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 await agent.serve_async(). 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 Mistral models using the Mistral API.
if "MISTRAL_API_KEY" not in os.environ:
print("MISTRAL_API_KEY not found in environment!")
api_key = getpass("Please enter your MISTRAL_API_KEY: ")
os.environ["MISTRAL_API_KEY"] = api_key
print("MISTRAL_API_KEY set for this session!")
else:
print("MISTRAL_API_KEY found in environment.")
import os
from getpass import getpass
# This notebook communicates with Mistral models using the Mistral API.
if "MISTRAL_API_KEY" not in os.environ:
print("MISTRAL_API_KEY not found in environment!")
api_key = getpass("Please enter your MISTRAL_API_KEY: ")
os.environ["MISTRAL_API_KEY"] = api_key
print("MISTRAL_API_KEY set for this session!")
else:
print("MISTRAL_API_KEY found in environment.")
MISTRAL_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 [ ]:
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="mistral/mistral-small-latest",
description="I'm an agent to help with getting the time",
tools=[time_tool],
),
)
time_handle = await time.serve_async(A2AServingConfig(port=0))
server_port = time_handle.port
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="mistral/mistral-small-latest",
description="I'm an agent to help with getting the time",
tools=[time_tool],
),
)
time_handle = await time.serve_async(A2AServingConfig(port=0))
server_port = time_handle.port
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)
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:54288/",
"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, http_kwargs={"timeout": 30.0})
# 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, http_kwargs={"timeout": 30.0})
# Close the httpx client when done
await httpx_client.aclose()
╭──────────────────────────────────────────── CALL_LLM: mistral/mistral-medium-latest ─────────────────────────────────────────────╮ │ ╭─ 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-27T08:20:44-04:00", │ │ │ │ "is_dst": true │ │ │ │ } │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭──────────────────────────────────────────── CALL_LLM: mistral/mistral-medium-latest ─────────────────────────────────────────────╮ │ ╭─ OUTPUT ────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ The current time in the America/New_York timezone is 8:20 AM on June 27, 2025. If you want to know the time │ │ │ │ in another timezone, please let me know. │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ │ ╭─ USAGE ─────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ { │ │ │ │ "input_tokens": 295, │ │ │ │ "output_tokens": 42 │ │ │ │ } │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
In [6]:
Copied!
print(response.model_dump_json(indent=2))
print(response.model_dump_json(indent=2))
{
"id": "fb2bed8d-5a97-4200-9d85-9435fa7e5056",
"jsonrpc": "2.0",
"result": {
"artifacts": null,
"contextId": "2bc7f223-b5ea-4e26-a135-329ecd15fb11",
"history": [
{
"contextId": "2bc7f223-b5ea-4e26-a135-329ecd15fb11",
"extensions": null,
"kind": "message",
"messageId": "35685ee6f31c428aa9b93c1e98eb9fc8",
"metadata": null,
"parts": [
{
"kind": "text",
"metadata": null,
"text": "What time is it?"
}
],
"referenceTaskIds": null,
"role": "user",
"taskId": "d740f7e4-d44d-4df8-bc7e-f82379ff803c"
}
],
"id": "d740f7e4-d44d-4df8-bc7e-f82379ff803c",
"kind": "task",
"metadata": null,
"status": {
"message": {
"contextId": "2bc7f223-b5ea-4e26-a135-329ecd15fb11",
"extensions": null,
"kind": "message",
"messageId": "64bb6ebe-4f8b-4579-a509-8ba3c3d59a51",
"metadata": null,
"parts": [
{
"kind": "text",
"metadata": null,
"text": "{\"result\":\"The current time in the America/New_York timezone is 8:20 AM on June 27, 2025. If you want to know the time in another timezone, please let me know.\"}"
}
],
"referenceTaskIds": null,
"role": "agent",
"taskId": "d740f7e4-d44d-4df8-bc7e-f82379ff803c"
},
"state": "completed",
"timestamp": "2025-06-27T12:20:45.987764+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!
await time_handle.shutdown()
print("Agent server has been shut down!")
await time_handle.shutdown()
print("Agent server has been shut down!")
Agent server has been shut down!