google / google/adk-python

A2aAgentExecutor support for `x402-a2a`

Ouverte
#3,009 1 commentaire 0 réactions 2 personnes assignées Réclamée par @yeesian Voir sur GitHub
agent engine needs review
Langage dominant
Python
Étoiles
21.5k
Forks
4k
Merge moyen
1 j 22 h
PR mergées (30 j)
31

Description

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

Guide de contribution

Ouvrir le guide de contribution

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.