google / google/adk-python

A2aAgentExecutor support for `x402-a2a`

Abierto
#3,009 1 comentario 0 reacciones 2 asignados Reclamado por @yeesian Ver en GitHub
agent engine needs review
Lenguaje dominante
Python
Estrellas
21.5k
Forks
4k
Merge medio
1 d 14 h
PR fusionados (30 d)
37

Descripción

## 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)
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.