Repository navigation
A2aAgentExecutor support for x402-a2a #3009
Description
Problem
A2aAgentExecutor doesn't play nicely with x402-a2a.
In particular x402-a2a expects ADK agents to:
raise x402PaymentRequiredException(...)and that to be caught by x402ServerExecutor but the error is instead caught by A2aAgentExecutor if you wrap them the following way:
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:
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:
agent_executor = InnerA2aAgentExecutor(runner=runner)
agent_executor = x402SellerExecutor(delegate=agent_executor)
agent_executor = OuterA2aAgentExecutor(agent_executor)
Activity
klateefa commentedon Nov 26, 2025
@matiasanaya Thank you for reporting this. I understand the request is currently under review. Please keep me posted if anything else is needed. Thanks for your patience.
DeanChensj commentedon Oct 14, 2026
Thank you for the detailed write-up and code example!
A2aAgentExecutorConfignow supportsexecute_interceptors(ExecuteInterceptorwithbefore_agent,after_event, andafter_agenthooks), which allows middleware-style inspection and customization of theRequestContext, outgoing A2A events, and terminalTaskStatusUpdateEvents around agent execution.We are closing this issue for now as part of our backlog triage, but if the current
ExecuteInterceptorhooks don't cover a specific requirement for yourx402-a2aintegration, please feel free to open a new issue and we'd be glad to take a closer look!github-actions commentedon Oct 14, 2026
Are you satisfied with the resolution of your issue?
Yes
No