A2aAgentExecutor reports a tool awaiting human confirmation as INPUT_REQUIRED, not AUTH_REQUIRED (A2A spec 7.6) #7810

"""A tool awaiting human confirmation is reported as INPUT_REQUIRED, not AUTH_REQUIRED.

Each case serves an ADK agent with ADK's own A2aAgentExecutor, in process, and
sends it one blocking A2A SendMessage. Scripted models: no network, no creds.

  confirmation  tool with require_confirmation=True    -> adk_request_confirmation
  credential    tool calling request_credential (OAuth) -> adk_request_credential
  relayed       a chat agent delegating (RemoteA2aAgent mode="task") to the
                confirmation agent: the state its *own* task reports
"""

import asyncio
import logging
import uuid
import warnings

import httpx
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import (
    add_a2a_routes_to_fastapi,
    create_agent_card_routes,
    create_jsonrpc_routes,
)
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentInterface
from fastapi import FastAPI
from fastapi.openapi.models import OAuth2, OAuthFlowAuthorizationCode, OAuthFlows
from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.auth.auth_credential import AuthCredential, AuthCredentialTypes, OAuth2Auth
from google.adk.auth.auth_tool import AuthConfig
from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext
from google.genai import types

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


def delete_records(table: str) -> dict:
    """Deletes every record in a table. Destructive: needs a human's approval."""
    return {"status": "deleted", "table": table}


def read_calendar(tool_context: ToolContext) -> dict:
    """Reads the user's calendar. Needs an OAuth token."""
    tool_context.request_credential(
        AuthConfig(
            auth_scheme=OAuth2(
                flows=OAuthFlows(
                    authorizationCode=OAuthFlowAuthorizationCode(
                        authorizationUrl="https://example.test/auth",
                        tokenUrl="https://example.test/token",
                        scopes={"calendar": "read"},
                    )
                )
            ),
            raw_auth_credential=AuthCredential(
                auth_type=AuthCredentialTypes.OAUTH2,
                oauth2=OAuth2Auth(client_id="id", client_secret="secret"),
            ),
        )
    )
    return {"status": "pending"}


class CallOnce(BaseLlm):
    """Calls one tool, then (if ever resumed) says done."""

    tool: str
    args: dict = {}

    async def generate_content_async(self, llm_request, stream=False):
        if llm_request.contents[-1].parts[-1].function_response:
            part = types.Part(text="done")
        else:
            part = types.Part(
                function_call=types.FunctionCall(name=self.tool, args=self.args)
            )
        yield LlmResponse(content=types.Content(role="model", parts=[part]))


def card(name):
    return AgentCard(
        name=name,
        description=name,
        version="1",
        capabilities=AgentCapabilities(streaming=True),
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        supported_interfaces=[
            AgentInterface(url=f"http://{name}.test/a2a", protocol_binding="JSONRPC")
        ],
    )


def serve(agent):
    """Serve an ADK agent with ADK's own A2aAgentExecutor, in process."""
    runner = Runner(
        app_name=agent.name,
        agent=agent,
        session_service=InMemorySessionService(),
        auto_create_session=True,
    )
    c = card(agent.name)
    handler = DefaultRequestHandler(
        A2aAgentExecutor(runner=runner), InMemoryTaskStore(), c
    )
    app = FastAPI()
    add_a2a_routes_to_fastapi(
        app,
        agent_card_routes=create_agent_card_routes(c),
        jsonrpc_routes=create_jsonrpc_routes(handler, rpc_url="/a2a"),
    )
    return httpx.AsyncClient(transport=httpx.ASGITransport(app=app))


async def task_state(name, client):
    """Send one blocking SendMessage and return the task's state and call names."""
    body = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "SendMessage",
        "params": {
            "message": {
                "messageId": uuid.uuid4().hex,
                "role": "ROLE_USER",
                "parts": [{"text": "go"}],
            }
        },
    }
    r = await client.post(
        f"http://{name}.test/a2a", json=body, headers={"A2A-Version": "1.0"}
    )
    task = r.json()["result"]["task"]
    calls = [
        p["data"].get("name")
        for p in task["status"].get("message", {}).get("parts", [])
        if "data" in p
    ]
    return task["status"]["state"], calls


def approver():
    return LlmAgent(
        name="approver",
        model=CallOnce(model="fake", tool="delete_records", args={"table": "t"}),
        tools=[FunctionTool(delete_records, require_confirmation=True)],
    )


async def main():
    cases = {
        "confirmation": approver(),
        "credential": LlmAgent(
            name="calendar",
            model=CallOnce(model="fake", tool="read_calendar"),
            tools=[read_calendar],
        ),
        "relayed": LlmAgent(
            name="coordinator",
            model=CallOnce(model="fake", tool="approver"),
            sub_agents=[
                RemoteA2aAgent(
                    name="approver",
                    description="approver",
                    agent_card=card("approver"),
                    mode="task",
                    httpx_client=serve(approver()),
                )
            ],
        ),
    }
    for label, agent in cases.items():
        state, calls = await task_state(agent.name, serve(agent))
        print(f"{label:13} {state:26} {calls}")


asyncio.run(main())