Serve an Agent with A2A¶
Once you've built an agent, a common question might be: "how do I let other developers/applications access it?". Enter the A2A protocol by Google! A2A is "An open protocol enabling communication and interoperability between opaque agentic applications". Any-Agent provides support for serving an agent over A2A, as simple as calling agent.serve(). In this tutorial, we'll build and serve an agent using any-agent, and show how you can serve and interact with it via the A2A protocol.
This tutorial assumes basic familiarity with any-agent: if you haven't used any-agent before you may also find the Creating your first agent cookbook to be useful.
Note: because this tutorial relies upon advanced stdio/stderr communication using the MCP Server, it cannot be run on Google Colab.
Install Dependencies¶
any-agent uses the python asyncio module to support async functionality. When running in Jupyter notebooks, this means we need to enable the use of nested event loops. We'll install any-agent and enable this below using nest_asyncio.
In [ ]:
Copied!
%pip install 'any-agent[a2a]'
import nest_asyncio
nest_asyncio.apply()
%pip install 'any-agent[a2a]'
import nest_asyncio
nest_asyncio.apply()
In [2]:
Copied!
import os
from getpass import getpass
# This notebook communicates with OpenAI GPT models using the OpenAI API.
if "OPENAI_API_KEY" not in os.environ:
print("OPENAI_API_KEY not found in environment!")
api_key = getpass("Please enter your OPENAI_API_KEY: ")
os.environ["OPENAI_API_KEY"] = api_key
print("OPENAI_API_KEY set for this session!")
else:
print("OPENAI_API_KEY found in environment.")
import os
from getpass import getpass
# This notebook communicates with OpenAI GPT models using the OpenAI API.
if "OPENAI_API_KEY" not in os.environ:
print("OPENAI_API_KEY not found in environment!")
api_key = getpass("Please enter your OPENAI_API_KEY: ")
os.environ["OPENAI_API_KEY"] = api_key
print("OPENAI_API_KEY set for this session!")
else:
print("OPENAI_API_KEY found in environment.")
OPENAI_API_KEY found in environment.
Configure and run the server¶
Let's give our agent the very simple capability to access the current time through a Model Context Protocol (MCP) server. For this demo, we'll use the async method agent.serve_async so that we can easily run both the server and client from inside the notebook.
In [3]:
Copied!
import asyncio
import httpx
from any_agent import AgentConfig, AnyAgent
from any_agent.config import MCPStdio
from any_agent.serving import A2AServingConfig
# This MCP Tool relies upon uvx https://docs.astral.sh/uv/getting-started/installation/
time_tool = MCPStdio(
command="uvx",
args=["mcp-server-time", "--local-timezone=America/New_York"],
tools=[
"get_current_time",
],
)
time = await AnyAgent.create_async(
"tinyagent", # See all options in https://mozilla-ai.github.io/any-agent/
AgentConfig(
model_id="gpt-4.1-mini",
description="I'm an agent to help with getting the time",
tools=[time_tool],
),
)
(time_task, time_server) = await time.serve_async(A2AServingConfig(port=0))
server_port = time_server.servers[0].sockets[0].getsockname()[1]
max_attempts = 20
poll_interval = 0.5
attempts = 0
server_url = f"http://localhost:{server_port}"
async with httpx.AsyncClient() as client:
while True:
try:
# Try to make a basic GET request to check if server is responding
await client.get(server_url, timeout=1.0)
print(f"Server is ready at {server_url}")
break
except (httpx.RequestError, httpx.TimeoutException):
# Server not ready yet, continue polling
pass
await asyncio.sleep(poll_interval)
attempts += 1
if attempts >= max_attempts:
msg = f"Could not connect to {server_url}. Tried {max_attempts} times with {poll_interval} second interval."
raise ConnectionError(msg)
import asyncio
import httpx
from any_agent import AgentConfig, AnyAgent
from any_agent.config import MCPStdio
from any_agent.serving import A2AServingConfig
# This MCP Tool relies upon uvx https://docs.astral.sh/uv/getting-started/installation/
time_tool = MCPStdio(
command="uvx",
args=["mcp-server-time", "--local-timezone=America/New_York"],
tools=[
"get_current_time",
],
)
time = await AnyAgent.create_async(
"tinyagent", # See all options in https://mozilla-ai.github.io/any-agent/
AgentConfig(
model_id="gpt-4.1-mini",
description="I'm an agent to help with getting the time",
tools=[time_tool],
),
)
(time_task, time_server) = await time.serve_async(A2AServingConfig(port=0))
server_port = time_server.servers[0].sockets[0].getsockname()[1]
max_attempts = 20
poll_interval = 0.5
attempts = 0
server_url = f"http://localhost:{server_port}"
async with httpx.AsyncClient() as client:
while True:
try:
# Try to make a basic GET request to check if server is responding
await client.get(server_url, timeout=1.0)
print(f"Server is ready at {server_url}")
break
except (httpx.RequestError, httpx.TimeoutException):
# Server not ready yet, continue polling
pass
await asyncio.sleep(poll_interval)
attempts += 1
if attempts >= max_attempts:
msg = f"Could not connect to {server_url}. Tried {max_attempts} times with {poll_interval} second interval."
raise ConnectionError(msg)
/Users/nbrake/scm/any-agent/.venv/lib/python3.13/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html from .autonotebook import tqdm as notebook_tqdm
Server is ready at http://localhost:61416
Call the agent using A2AClient¶
Now that the agent is listening on localhost, we can communicate with it from any other application that supports A2A. For this tutorial we'll use the a2a python SDK, but any client that implements the A2A Protocol would also work.
A2A Agent Card¶
Before giving the agent a task, we first retrieve the agent's model card, which is a description of the agents capabilities. This helps the client understand what the agent can do, and what A2A features are available.
In [4]:
Copied!
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import AgentCard
# Create the httpx client
httpx_client = httpx.AsyncClient()
agent_card: AgentCard = await A2ACardResolver(
httpx_client,
base_url=f"http://localhost:{server_port}",
).get_agent_card(http_kwargs=None)
print(agent_card.model_dump_json(indent=2))
client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import AgentCard
# Create the httpx client
httpx_client = httpx.AsyncClient()
agent_card: AgentCard = await A2ACardResolver(
httpx_client,
base_url=f"http://localhost:{server_port}",
).get_agent_card(http_kwargs=None)
print(agent_card.model_dump_json(indent=2))
client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
{
"capabilities": {
"extensions": null,
"pushNotifications": false,
"stateTransitionHistory": false,
"streaming": false
},
"defaultInputModes": [
"text"
],
"defaultOutputModes": [
"text"
],
"description": "I'm an agent to help with getting the time",
"documentationUrl": null,
"iconUrl": null,
"name": "any_agent",
"provider": null,
"security": null,
"securitySchemes": null,
"skills": [
{
"description": "Get current time in a specific timezones",
"examples": null,
"id": "any_agent-get_current_time",
"inputModes": null,
"name": "get_current_time",
"outputModes": null,
"tags": []
}
],
"supportsAuthenticatedExtendedCard": null,
"url": "http://localhost:61416/",
"version": "0.1.0"
}
Make a request of the agent¶
Now that we've connected to the agent with the A2AClient, we're ready to make a request of the agent!
In [5]:
Copied!
from uuid import uuid4
from a2a.types import MessageSendParams, SendMessageRequest
send_message_payload = {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "What time is it?"}],
"messageId": uuid4().hex,
},
}
request = SendMessageRequest(
id=str(uuid4()), params=MessageSendParams(**send_message_payload)
)
response = await client.send_message(request)
# Close the httpx client when done
await httpx_client.aclose()
from uuid import uuid4
from a2a.types import MessageSendParams, SendMessageRequest
send_message_payload = {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "What time is it?"}],
"messageId": uuid4().hex,
},
}
request = SendMessageRequest(
id=str(uuid4()), params=MessageSendParams(**send_message_payload)
)
response = await client.send_message(request)
# Close the httpx client when done
await httpx_client.aclose()
╭──────────────────────────────────────────── CALL_LLM: gpt-4.1-mini ─────────────────────────────────────────────╮ │ ╭─ INPUT ─────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ [ │ │ │ │ { │ │ │ │ "role": "system", │ │ │ │ "content": "You are an agent - please keep going until the user's query is completely resolved, before │ │ │ │ }, │ │ │ │ { │ │ │ │ "role": "user", │ │ │ │ "content": "What time is it?" │ │ │ │ } │ │ │ │ ] │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ │ ╭─ OUTPUT ────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ [ │ │ │ │ { │ │ │ │ "tool.name": "get_current_time", │ │ │ │ "tool.args": "{\"timezone\":\"America/New_York\"}" │ │ │ │ } │ │ │ │ ] │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ │ ╭─ USAGE ─────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ { │ │ │ │ "input_tokens": 227, │ │ │ │ "output_tokens": 18 │ │ │ │ } │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭──────────────────────────────────────── EXECUTE_TOOL: get_current_time ─────────────────────────────────────────╮ │ ╭─ Input ─────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ { │ │ │ │ "timezone": "America/New_York" │ │ │ │ } │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ │ ╭─ OUTPUT ────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ { │ │ │ │ "timezone": "America/New_York", │ │ │ │ "datetime": "2025-06-13T13:28:21-04:00", │ │ │ │ "is_dst": true │ │ │ │ } │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭──────────────────────────────────────────── CALL_LLM: gpt-4.1-mini ─────────────────────────────────────────────╮ │ ╭─ OUTPUT ────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ The current time in New York (America/New_York timezone) is approximately 1:28 PM on June 13, 2025. If you │ │ │ │ want to know the time in a different location or timezone, please let me know! │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ │ ╭─ USAGE ─────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ { │ │ │ │ "input_tokens": 295, │ │ │ │ "output_tokens": 49 │ │ │ │ } │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
In [6]:
Copied!
print(response.model_dump_json(indent=2))
print(response.model_dump_json(indent=2))
{
"id": "d9357217-38e7-40b5-ab6a-d6a06f83af0c",
"jsonrpc": "2.0",
"result": {
"artifacts": null,
"contextId": "6bde1392-d951-4876-9686-0bc30cd7dbb7",
"history": [
{
"contextId": "6bde1392-d951-4876-9686-0bc30cd7dbb7",
"extensions": null,
"kind": "message",
"messageId": "56f9f3de356d48dbbd495882860252f9",
"metadata": null,
"parts": [
{
"kind": "text",
"metadata": null,
"text": "What time is it?"
}
],
"referenceTaskIds": null,
"role": "user",
"taskId": "894a7ed3-b1c7-4578-b8f8-f5845538859b"
}
],
"id": "894a7ed3-b1c7-4578-b8f8-f5845538859b",
"kind": "task",
"metadata": null,
"status": {
"message": {
"contextId": "6bde1392-d951-4876-9686-0bc30cd7dbb7",
"extensions": null,
"kind": "message",
"messageId": "5ae21424-4882-4fd7-b386-6b620cd0a5a1",
"metadata": null,
"parts": [
{
"kind": "text",
"metadata": null,
"text": "{\"result\":\"The current time in New York (America/New_York timezone) is approximately 1:28 PM on June 13, 2025. If you want to know the time in a different location or timezone, please let me know!\"}"
}
],
"referenceTaskIds": null,
"role": "agent",
"taskId": "894a7ed3-b1c7-4578-b8f8-f5845538859b"
},
"state": "completed",
"timestamp": "2025-06-13T17:28:24.396695+00:00"
}
}
}
Cleanup¶
In order to shut down the A2A server, we can set the should_exit property of the server, which will cause the server to shutdown and the asyncio task to complete.
In [7]:
Copied!
import asyncio
time_server.should_exit = True
time_task.cancel()
wait_count = 0
max_wait_seconds = 10
while True:
if time_task.done():
print("Agent server task has completed!")
break
elif wait_count >= max_wait_seconds:
print(f"Timed out after {max_wait_seconds} seconds waiting for server to exit")
break
else:
print("Waiting for server to exit")
await asyncio.sleep(1)
wait_count += 1
import asyncio
time_server.should_exit = True
time_task.cancel()
wait_count = 0
max_wait_seconds = 10
while True:
if time_task.done():
print("Agent server task has completed!")
break
elif wait_count >= max_wait_seconds:
print(f"Timed out after {max_wait_seconds} seconds waiting for server to exit")
break
else:
print("Waiting for server to exit")
await asyncio.sleep(1)
wait_count += 1
Waiting for server to exit Agent server task has completed!