google / google/adk-python

A2aAgentExecutor support for `x402-a2a`

Đang mở
#3,009 1 bình luận 0 reaction 2 người được giao Được @yeesian nhận Xem trên GitHub
agent engine needs review
Ngôn ngữ chính
Python
Star
21.5k
Fork
4k
Merge trung bình
1 ngày 22 giờ
Pull request đã merge (30 ngày)
31

Mô tả

## Problem
`A2aAgentExecutor` doesn't play nicely with [x402-a2a](https://github.com/google-agentic-commerce/a2a-x402).
In particular `x402-a2a` expects ADK agents to:
```python
raise x402PaymentRequiredException(...)
```
and that to be caught by `x402ServerExecutor` but the error is instead caught by `A2aAgentExecutor` if you wrap them the following way:
```python
agent_executor = A2aAgentExecutor(runner=runner)
agent_executor = x402SellerExecutor(delegate=agent_executor)
```

## Solution

Support middleware in `A2aAgentExecutorConfig`, re-write `_handle_request` to be wrapped by the middleware.

## Alternative Solution

Another approach would be to split the A2aAgentExecutor into two, an inner and an outer one. Something like:
```python
class OuterA2aAgentExecutor(AgentExecutor):
def __init__(
self,
delegate: AgentExecutor,
):
super().__init__()
self._delegate = delegate

async def cancel(self, context: RequestContext, event_queue: EventQueue):
return await self._delegate.cancel(context, event_queue)

async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
):
if not context.message:
raise ValueError("A2A request must have a message")

assert context.task_id, "A2A request must have a task ID"
assert context.context_id, "A2A request must have a context ID"

# for new task, create a task submitted event
if not context.current_task:
await event_queue.enqueue_event(
TaskStatusUpdateEvent(
task_id=context.task_id,
status=TaskStatus(
state=TaskState.submitted,
message=context.message,
timestamp=datetime.now(timezone.utc).isoformat(),
),
context_id=context.context_id,
final=False,
)
)
try:
await self._delegate.execute(context, event_queue)
except Exception as e:
logger.error("Error handling A2A request: %s", e, exc_info=True)
# Publish failure event
try:
await event_queue.enqueue_event(
TaskStatusUpdateEvent(
task_id=context.task_id,
status=TaskStatus(
state=TaskState.failed,
timestamp=datetime.now(timezone.utc).isoformat(),
message=Message(
message_id=str(uuid.uuid4()),
role=Role.agent,
parts=[Part(TextPart(text=str(e)))],
),
),
context_id=context.context_id,
final=True,
)
)
except Exception as enqueue_error:
logger.error(
"Failed to publish failure event: %s", enqueue_error, exc_info=True
)

class InnerA2aAgentExecutor(A2aAgentExecutor):
@override
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
):
await self._handle_request(context, event_queue)
```
This would allow to do the following:
```python
agent_executor = InnerA2aAgentExecutor(runner=runner)
agent_executor = x402SellerExecutor(delegate=agent_executor)
agent_executor = OuterA2aAgentExecutor(agent_executor)
```

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.