A2A agent framework in one import.
Build Agent-to-Agent (A2A) protocol agents with minimal boilerplate. Streaming, cancellation, multi-turn conversations, and artifact handling — batteries included.
pip install a2akitWith optional extras:
pip install a2akit[langgraph] # LangGraph integration
pip install a2akit[otel] # OpenTelemetry tracing & metricsfrom a2akit import A2AServer, AgentCardConfig, TaskContext, Worker
class EchoWorker(Worker):
async def handle(self, ctx: TaskContext) -> None:
await ctx.complete(f"Echo: {ctx.user_text}")
server = A2AServer(
worker=EchoWorker(),
agent_card=AgentCardConfig(
name="Echo Agent",
description="Echoes your input back.",
version="0.1.0",
),
)
app = server.as_fastapi_app()uvicorn my_agent:app --reloadfrom a2akit import A2AClient
async with A2AClient("http://localhost:8000") as client:
result = await client.send("Hello, agent!")
print(result.text)
async for chunk in client.stream_text("Stream me"):
print(chunk, end="")Enable the built-in debug interface to test your agent in the browser:
app = server.as_fastapi_app(debug=True)Open http://localhost:8000/chat — chat with your agent and inspect tasks in real time.
- Debug UI — built-in browser interface for chat + task inspection (
debug=True) - One-liner setup —
A2AServerwires storage, broker, event bus, and endpoints - A2AClient — auto-discovers agents, supports send/stream/cancel/subscribe
- Capabilities — explicit opt-in for streaming and state transition history, enforced on server and client
- Streaming — word-by-word artifact streaming via SSE
- Cancellation — cooperative and force-cancel with timeout fallback
- Multi-turn —
request_input()/request_auth()for conversational flows - Direct reply —
reply_directly()for simple request/response without task tracking - Middleware — pipeline for auth extraction, header injection, payload sanitization
- Lifecycle hooks — fire-and-forget callbacks on terminal state transitions
- Dependency injection — shared infrastructure with automatic lifecycle management
- OpenTelemetry — opt-in distributed tracing and metrics (
pip install a2akit[otel]) - Pluggable backends — PostgreSQL, SQLite, and more (Redis, RabbitMQ coming soon)
- Type-safe — full type hints,
py.typedmarker, PEP 561 compliant
MIT