a2akit

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.

---

• One-liner setup

  ---

  A2AServer wires storage, broker, event bus, and endpoints automatically.

• Streaming

  ---

  Word-by-word artifact streaming via SSE with emit_text_artifact.

• Cancellation

  ---

  Cooperative and force-cancel with configurable timeout fallback.

• Multi-turn

  ---

  request_input() / request_auth() for conversational flows.

• Middleware

  ---

  Pipeline for auth extraction, header injection, and payload sanitization.

• Lifecycle Hooks

  ---

  Fire-and-forget callbacks on terminal state transitions.

• Type-safe

  ---

  Full type hints, py.typed marker, PEP 561 compliant.

• Pluggable backends

  ---

  Swap in PostgreSQL, SQLite, Redis, or RabbitMQ (coming soon).

Quick Start

from 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 --reload

curl -X POST http://localhost:8000/v1/message:send \
  -H "Content-Type: application/json" \
  -d '{"message":{"role":"user","parts":[{"text":"hello"}],"messageId":"1"}}'

---

Get Started Architecture