Expose the docs assistant over A2A and consume it

The release coordinator needs answers from a docs assistant owned by another team. Serve that assistant over the Agent2Agent (A2A) protocol with to_a2a, then connect the coordinator through RemoteA2aAgent. You inspect the advertised agent card, send a rollback question and follow the streamed task updates back to the coordinator. Each side keeps its own agent definition and tools.

When you need this. When another team owns an agent with its own tools, instructions and model, such as a docs assistant, a release-notes reviewer or a deploy agent, we call it over A2A instead of importing its code. Each team deploys and changes its agent on its own schedule, and the caller uses the agent card and A2A messages instead of importing the remote implementation.

A2A puts the whole assistant behind one HTTP endpoint that another team’s app calls.

1. Install the ADK A2A extra

Download the complete lab ZIP, extract it, and open a terminal in adk-a2a/. The commands below use that directory and its pinned dependencies. Set GOOGLE_API_KEY in each terminal that runs an agent. The [a2a] extra of google-adk installs a2a-sdk with its HTTP server; the lock pins a2a-sdk 1.2.3. adk.dev marks ADK’s A2A support as Experimental. To run Gemma locally on either side, set DOCS_MODEL=ollama_chat/gemma4:e2b.

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt

2. Serve the assistant with to_a2a

"""Expose the docs assistant as an A2A server with to_a2a, served by uvicorn.

Run: python server.py            # serves http://localhost:8001 and /.well-known/agent-card.json
"""
import contextlib
import os
import socket
import threading
import time
from pathlib import Path

import uvicorn
from google.adk import Agent
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from google.adk.models.lite_llm import LiteLlm
from google.genai import types

HERE = Path(__file__).resolve().parent
# Gemini through ADK's native client; set DOCS_MODEL=ollama_chat/gemma4:e2b to run Gemma locally.
MODEL = os.environ.get("DOCS_MODEL", "gemini-3.8-flash")
PAGES = sorted(p.stem for p in (HERE / "docs").glob("*.md"))


def read_page(page: str) -> dict:
    """Read one page of the docuctl documentation: config, deploy, rollback or tokens."""
    if page not in PAGES:
        return {"error": f"No page {page!r}. Pages: {', '.join(PAGES)}."}
    root = (HERE / "docs").resolve()
    path = (root / f"{page}.md").resolve()
    if not path.is_relative_to(root) or not path.is_file():
        return {"error": f"No readable docs page {page!r}."}
    return {"page": page, "text": path.read_text(encoding="utf-8")}


def build_agent(model=None) -> Agent:
    return Agent(
        name="docs_assistant",
        description="Answers questions about the docuctl documentation: config, deploy, rollback and tokens.",
        model=model or (LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL),
        instruction="Answer from the docuctl documentation only. Read the page with read_page and name it. "
                    "If the docs do not cover the question, say so.",
        tools=[read_page],
        generate_content_config=types.GenerateContentConfig(
            temperature=0.2, thinking_config=types.ThinkingConfig(thinking_level="low")),
    )


def build_app(agent: Agent | None = None, port: int = 8001, card_port: int | None = None):
    """The Starlette app. The agent card advertises http://localhost:{card_port or port}/ as the RPC URL."""
    return to_a2a(agent or build_agent(), host="localhost", port=card_port or port)


def free_port() -> int:
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


@contextlib.contextmanager
def running(app, port: int, startup_timeout: float = 10.0):
    """Serve app with uvicorn in a background thread, for the recorder and the tests."""
    server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning"))
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    try:
        deadline = time.monotonic() + startup_timeout
        while not server.started:
            if not thread.is_alive():
                raise RuntimeError("A2A server failed to start; check its log and port")
            if time.monotonic() >= deadline:
                raise TimeoutError("A2A server startup timed out")
            time.sleep(0.05)
        yield f"http://localhost:{port}"
    finally:
        server.should_exit = True
        thread.join(timeout=10)


if __name__ == "__main__":
    uvicorn.run(build_app(), host="127.0.0.1", port=8001)

to_a2a(agent, host=, port=) returns a Starlette app. It wraps the agent in an A2aAgentExecutor with a default in-memory Runner, keeps tasks in an InMemoryTaskStore, and serves the agent card at /.well-known/agent-card.json. host and port do not choose where the server listens; uvicorn does that. They build the RPC URL that the card advertises, which matters as soon as the server sits behind a proxy.

The defaults suit a local run. The app keeps tasks in memory and the default runner keeps sessions in memory, so a restart loses running tasks and conversation state; pass task_store= and a runner= with persistent services for a long-running server. It also has no authentication and listens on 127.0.0.1 here. to_a2a accepts security_schemes for the card, and RemoteA2aAgent accepts auth_scheme and auth_credential; read ADK: A2A before you expose an agent beyond your machine.

The other way to serve an ADK agent over A2A is adk api_server --a2a, which serves every agent folder that has a hand-written agent.json card. to_a2a generates the card for you and gives you the app to deploy, which suits one agent per service.

Start the server in the first terminal and keep it running:

python server.py

The card is the contract. ADK built it from the agent: name and description come from the Agent, and each tool became a skill, so description is now the first thing another team’s model reads about this agent. Write it for them.

The server thread exits during startup. The helper raises RuntimeError; a live thread also has a startup deadline. Inspect the port and server log before accepting traffic rather than waiting on a dead thread.

A card whose RPC URL has another origin. The client fetches the card but refuses the RPC URL and emits an error event. Set to_a2a(host=, port=, protocol=) to the public address clients actually use. This origin check limits redirection; it does not authenticate the server.

3. Call it from another agent

"""A release coordinator that delegates docs questions to the remote docs assistant over A2A.

Run (with server.py running): python client.py "How do I roll back production to r41?"
"""
import asyncio
import os
import sys

import httpx
from google.adk import Agent, Runner
from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH, RemoteA2aAgent
from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
from google.genai import types

MODEL = os.environ.get("DOCS_MODEL", "gemini-3.8-flash")
DOCS_URL = os.environ.get("DOCS_A2A_URL", "http://localhost:8001")


def remote_docs(url: str = DOCS_URL, timeout: float = 60.0, httpx_client: httpx.AsyncClient | None = None):
    """The remote agent, described by the card it serves at the well-known path."""
    return RemoteA2aAgent(
        name="docs_assistant",
        description="Answers questions about the docuctl documentation.",
        agent_card=f"{url}{AGENT_CARD_WELL_KNOWN_PATH}",
        timeout=timeout,
        httpx_client=httpx_client,
    )


def build_agent(model=None, remote: RemoteA2aAgent | None = None) -> Agent:
    return Agent(
        name="release_coordinator",
        model=model or (LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL),
        instruction="You coordinate docuctl releases. Transfer every question about how docuctl works "
                    "to docs_assistant. Answer release scheduling questions yourself.",
        sub_agents=[remote or remote_docs()],
        generate_content_config=types.GenerateContentConfig(
            temperature=0.2, thinking_config=types.ThinkingConfig(thinking_level="low")),
    )


async def ask(text: str, agent: Agent | None = None) -> list:
    sessions = InMemorySessionService()
    runner = Runner(app_name="releases", agent=agent or build_agent(), session_service=sessions)
    try:
        session = await sessions.create_session(app_name="releases", user_id="dev")
        message = types.Content(role="user", parts=[types.Part(text=text)])
        return [e async for e in runner.run_async(user_id="dev", session_id=session.id, new_message=message)]
    finally:
        await runner.close()


def show(events: list) -> str:
    lines = []
    for event in events:
        for call in event.get_function_calls():
            lines.append(f"{event.author}: call {call.name} {call.args}")
        if event.error_code or event.error_message:
            lines.append(f"{event.author}: error {event.error_code} {event.error_message}")
        text = "".join(p.text or "" for p in (event.content.parts if event.content else None) or [] if not p.thought)
        if text.strip():
            lines.append(f"{event.author}: {text.strip()}")
    return "\n".join(lines)


if __name__ == "__main__":
    print(show(asyncio.run(ask(" ".join(sys.argv[1:])))))

RemoteA2aAgent is an agent whose work happens elsewhere. It takes the card URL, fetches the card on first use, and sends the conversation to the RPC URL in it. Because it is an agent, the coordinator lists it in sub_agents and reaches it with transfer_to_agent, exactly as it would a local sub-agent.

Open a second terminal in adk-a2a/, activate the same environment and inspect the card before asking the coordinator:

source .venv/bin/activate
curl -s http://localhost:8001/.well-known/agent-card.json
python client.py "How do I roll back production to r41?"

Here is what happens when the coordinator transfers: the remote assistant’s read_page call and its answer arrive in the coordinator’s own event stream, authored by docs_assistant, when the remote run uses that tool. The coordinator’s declarations contain transfer_to_agent; read_page belongs to the remote assistant, not to its own toolset.

A server that is down. The failed card fetch becomes an event with error_code and error_message, not a raised exception. Probe the card at startup and inspect error fields on each event; printing only text produces a blank answer.

A timeout that is ignored. When an httpx_client is supplied, RemoteA2aAgent(timeout=) does not configure it. Set the timeout on that httpx.AsyncClient, or let ADK create the client. HTTPX timeouts apply to connection, read, write and pool operations; a read timeout limits inactivity between chunks, not the total duration of an active event stream.

Handle errors alongside normal answer parts. A failed card fetch or origin mismatch appears as an event with error_message, so printing only text can produce a blank answer. A client timeout can cancel the local run; do not treat that cancellation as confirmation that provider processing or billing stopped.

Inspect the agent card and streamed task updates

On the first remote invocation, the client fetches the agent card, then sends a JSON-RPC message/stream POST. The message carries the user’s question and the coordinator’s own events so far, each quoted between markers that tell the remote model to treat them as data. That transcript crosses the boundary to the other team’s service, so review which context a coordinator is permitted to share before enabling a remote transfer.

Original recording. The saved A2A exchange uses the earlier shipctl documents and model replies. Current source uses docuctl; replaying this exchange is not a new docuctl model run.

The response is a stream of server-sent events, one per task update. The remote run’s function call and result travel as data parts in TASK_STATE_WORKING updates, the answer as an artifact, and the task ends in TASK_STATE_COMPLETED:

task: TASK_STATE_SUBMITTED
statusUpdate: TASK_STATE_WORKING
statusUpdate: TASK_STATE_WORKING (function_call)
statusUpdate: TASK_STATE_WORKING (function_response)
statusUpdate: TASK_STATE_WORKING (text)
artifactUpdate: the answer
statusUpdate: TASK_STATE_COMPLETED

When MCP exposes individual function tools, the calling model chooses their calls. MCP can also wrap an entire agent, as in Expose the docs assistant as an MCP server, so a remote model loop is not exclusive to A2A. This A2A example uses an agent card for discovery and task updates with states and artifacts. The docs team owns its remote model loop and implementation. The coordinator handles the returned task updates through its own Runner.

Test both sides offline

test_a2a.py serves the docs assistant with to_a2a and uvicorn on a free port, and replays the recorded model turns on both sides. The agent card, the JSON-RPC stream and RemoteA2aAgent are real.

python -m unittest -v test_a2a.py

"""Replay the recorded Gemini turns on both sides of a real A2A connection, offline.

Run: python -m unittest -v test_a2a.py
The server (to_a2a + uvicorn), the agent card, the JSON-RPC stream and RemoteA2aAgent are real;
both models are replaced by their recorded turns.
"""
import asyncio
import json
import logging
from unittest.mock import patch
import unittest
import tempfile
import warnings
from pathlib import Path

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

import httpx  # noqa: E402
from google.adk.models.base_llm import BaseLlm  # noqa: E402
from google.adk.models.llm_response import LlmResponse  # noqa: E402
from google.genai import types  # noqa: E402

import client  # noqa: E402
import record  # noqa: E402
import server  # noqa: E402

RECORD = json.loads(Path(__file__).with_name("recorded-runs.json").read_text(encoding="utf-8"))
ROLLBACK = RECORD["runs"]["rollback"]


def setUpModule():
    warnings.simplefilter("ignore")  # unittest resets warning filters before the tests run


class ReplayLlm(BaseLlm):
    turns: list
    delay: float = 0

    async def generate_content_async(self, llm_request, stream=False):
        await asyncio.sleep(self.delay)
        yield LlmResponse(content=types.Content.model_validate(self.turns.pop(0)))


def docs_server(delay: float = 0, card_port: int | None = None):
    port = server.free_port()
    remote = server.build_agent(ReplayLlm(model="replay", turns=list(ROLLBACK["remote_model_turns"]), delay=delay))
    return server.running(server.build_app(remote, port, card_port), port)


def ask(url: str, http_timeout: float = 60, **remote) -> tuple[str, list]:
    async def go():
        http, exchange = record.wire(http_timeout)
        coordinator = client.build_agent(ReplayLlm(model="replay", turns=list(ROLLBACK["model_turns"])),
                                         client.remote_docs(url, httpx_client=http, **remote))
        try:
            return client.show(await client.ask(ROLLBACK["question"], coordinator)), exchange
        finally:
            await http.aclose()
    return asyncio.run(go())


def states(exchange: list) -> list[str]:
    return [next(iter(e["result"].values())).get("status", {}).get("state", "artifact") for e in exchange[1]["response"]]


class A2ATests(unittest.TestCase):
    def test_a_server_thread_that_exits_does_not_hang_startup(self):
        class FailedServer:
            started = False
            should_exit = False
            def run(self):
                pass
        with patch.object(server.uvicorn, "Server", return_value=FailedServer()):
            with self.assertRaisesRegex(RuntimeError, "failed to start"):
                with server.running(object(), server.free_port(), startup_timeout=0.5):
                    self.fail("server never started")

    def test_the_card_describes_the_agent_and_its_tool(self):
        with docs_server() as url:
            card = httpx.get(f"{url}/.well-known/agent-card.json").json()
        self.assertEqual(card["supportedInterfaces"][0]["url"], url)
        self.assertEqual({k: v for k, v in card.items() if k != "supportedInterfaces"},
                         {k: v for k, v in json.loads(json.dumps(RECORD["agent_card"]).replace("shipctl", "docuctl")).items() if k != "supportedInterfaces"})

    def test_the_rollback_question_crosses_a2a(self):
        with docs_server() as url:
            transcript, exchange = ask(url)
        self.assertEqual(transcript, ROLLBACK["transcript"])
        self.assertEqual([(e["method"], e["path"]) for e in exchange], [("GET", "/.well-known/agent-card.json"), ("POST", "/")])
        self.assertEqual(exchange[1]["request"]["method"], "SendStreamingMessage")

    def test_the_server_streams_the_task_through_its_states(self):
        with docs_server() as url:
            _, exchange = ask(url)
        self.assertEqual(states(exchange), states(ROLLBACK["exchange"]))
        self.assertEqual(states(exchange)[0], "TASK_STATE_SUBMITTED")
        self.assertEqual(states(exchange)[-1], "TASK_STATE_COMPLETED")

    def test_a_card_whose_rpc_url_has_another_origin_is_refused(self):
        with docs_server(card_port=server.free_port()) as url:
            transcript, exchange = ask(url)
        self.assertIn("Agent card RPC URL must have the same origin as the location the card was fetched from", transcript)
        self.assertEqual([e["method"] for e in exchange], ["GET"])

    def test_a_server_that_is_down_returns_an_error_event(self):
        transcript, _ = ask(f"http://localhost:{server.free_port()}")
        self.assertIn("docs_assistant: error None Failed to initialize remote A2A agent", transcript)
        self.assertIn("All connection attempts failed", transcript)

    def test_the_timeout_argument_is_ignored_with_your_own_http_client(self):
        with docs_server(delay=1.0) as url:
            ignored, _ = ask(url, timeout=0.3)
            timed_out, _ = ask(url, http_timeout=0.3)
        self.assertEqual(ignored, ROLLBACK["transcript"])
        self.assertEqual(timed_out.splitlines()[-1], "docs_assistant: error None A2A request failed: Client Request timed out")
        self.assertEqual(timed_out, RECORD["failures"]["timeout"]["transcript"])


class DocsPathSafety(unittest.TestCase):
    def test_read_page_refuses_an_out_of_tree_symlink(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "docs").mkdir()
            (root / "outside.txt").write_text("private content")
            (root / "docs/config.md").symlink_to(root / "outside.txt")
            with patch.object(server, "HERE", root), patch.object(server, "PAGES", ["config"]):
                result = server.read_page("config")
            self.assertIn("error", result)
            self.assertNotIn("private content", str(result))


if __name__ == "__main__":
    unittest.main()

Check your understanding

• Build a remote MCP service with authentication and authorization: the token checks a remote agent endpoint needs, shown on an MCP server.

References

Official sources checked October 21, 2026, against google-adk 2.12.0 and a2a-sdk 1.2.3.

• ADK: A2A protocol
• ADK: Introduction to A2A
• ADK: Quickstart, exposing a remote agent
• ADK: Quickstart, consuming a remote agent
• A2A protocol specification