A2aAgentExecutor sends the raw exception text to the A2A caller when the run fails #7385

"""What an A2A caller receives when the served ADK agent's run raises."""

import asyncio
import json
import logging
import warnings

import httpx
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import add_a2a_routes_to_fastapi, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentInterface
from fastapi import FastAPI
from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.agents import LlmAgent
from google.adk.models.base_llm import BaseLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

warnings.filterwarnings("ignore")
logging.disable(logging.CRITICAL)


class FailingModel(BaseLlm):  # stands in for a provider error
    async def generate_content_async(self, llm_request, stream=False):
        raise RuntimeError(
            "provider rejected the call: Traceback (most recent call last):\n"
            '  File "/srv/agent/.venv/lib/python3.12/site-packages/provider/client.py", line 900'
        )
        yield


async def main():
    card = AgentCard(
        name="agent",
        description="agent",
        version="1",
        capabilities=AgentCapabilities(streaming=True),
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        supported_interfaces=[
            AgentInterface(url="http://agent.test/a2a", protocol_binding="JSONRPC")
        ],
    )
    agent = LlmAgent(name="agent", model=FailingModel(model="fake"), instruction="x")
    executor = A2aAgentExecutor(
        runner=Runner(
            app_name="app",
            agent=agent,
            session_service=InMemorySessionService(),
            auto_create_session=True,
        )
    )
    app = FastAPI()
    add_a2a_routes_to_fastapi(
        app,
        jsonrpc_routes=create_jsonrpc_routes(
            DefaultRequestHandler(executor, InMemoryTaskStore(), card), rpc_url="/a2a"
        ),
    )
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://agent.test"
    ) as client:
        response = await client.post(
            "/a2a",
            headers={"A2A-Version": "1.0"},
            json={
                "jsonrpc": "2.0",
                "id": 1,
                "method": "SendMessage",
                "params": {
                    "message": {
                        "messageId": "m1",
                        "role": "ROLE_USER",
                        "parts": [{"text": "hello"}],
                    }
                },
            },
        )
    status = response.json()["result"]["task"]["status"]
    print(json.dumps(status, indent=2))


asyncio.run(main())