A2aAgentExecutor support for `x402-a2a`
- Lingua principale
- Python
- Stelle
- 21.5k
- Fork
- 4k
- Merge medio
- 1g 14h
- PR unite (30g)
- 37
Descrizione
## 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)
```
Guida per i contributori
Apri la guida per i contributori
Valutazione
Questa issue non è ancora stata valutata.