__main__.py
"""
Basic A2A Server
================
Starts a local A2A server backed by an Agno agent executor.
"""
import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
AgentAuthentication,
AgentCapabilities,
AgentCard,
AgentSkill,
)
from basic_agent import BasicAgentExecutor
# ---------------------------------------------------------------------------
# Create A2A Application
# ---------------------------------------------------------------------------
def create_server() -> A2AStarletteApplication:
skill = AgentSkill(
id="agno_agent",
name="Agno Agent",
description="Agno Agent",
tags=["Agno agent"],
examples=["hi", "hello"],
)
agent_card = AgentCard(
name="Agno Agent",
description="Agno Agent",
url="http://localhost:9999/",
version="1.0.0",
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities=AgentCapabilities(),
skills=[skill],
authentication=AgentAuthentication(schemes=["public"]),
)
request_handler = DefaultRequestHandler(
agent_executor=BasicAgentExecutor(),
task_store=InMemoryTaskStore(),
)
return A2AStarletteApplication(agent_card=agent_card, http_handler=request_handler)
# ---------------------------------------------------------------------------
# Run Server
# ---------------------------------------------------------------------------
if __name__ == "__main__":
server = create_server()
uvicorn.run(server.build(), host="0.0.0.0", port=9999, timeout_keep_alive=10)
basic_agent.py
"""
Basic A2A Agent Executor
========================
Implements an A2A executor that routes incoming text to an Agno agent.
"""
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.types import Part, TextPart
from a2a.utils import new_agent_text_message
from agno.agent import Agent, Message, RunOutput
from agno.models.openai import OpenAIChat
from typing_extensions import override
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
model=OpenAIChat(id="gpt-5.2"),
)
# ---------------------------------------------------------------------------
# Create Executor
# ---------------------------------------------------------------------------
class BasicAgentExecutor(AgentExecutor):
"""Test AgentProxy implementation."""
def __init__(self):
self.agent = agent
@override
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
message: Message = Message(role="user", content="")
for part in context.message.parts:
if isinstance(part, Part):
if isinstance(part.root, TextPart):
message.content = part.root.text
break
result: RunOutput = await self.agent.arun(message)
event_queue.enqueue_event(new_agent_text_message(result.content))
@override
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
raise Exception("Cancel not supported")
# ---------------------------------------------------------------------------
# Run Example
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print(
"Run `python cookbook/05_agent_os/interfaces/a2a/basic_agent/__main__.py` to start the A2A server."
)
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U agno a2a openai uvicorn
3
Export your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"
$Env:OPENAI_API_KEY="your_openai_api_key_here"
4
Run the example
Save the code blocks above as
__main__.py and basic_agent.py in the same directory, then run:python __main__.py