a2a-lite

A2A Lite — Python

Start by learning A2A, then add deployment guarantees explicitly. A2A Lite provides a small API over the official A2A Python SDK, with authenticated runtimes, durable state and opt-in recovery when you need them. See the deployment guide for supported guarantees and limits.

Build A2A agents in 8 lines. Add features when you need them.

Wraps the official A2A Python SDK with a simple, decorator-based API for A2A v1.0.

v1.2.1 — A2A Protocol v1.0. Serves A2A v1.0 on a2a-sdk >= 1.2.2.

• Default transports: JSON-RPC + REST (HTTP+JSON) via agent.run(); card at /.well-known/agent-card.json.
• gRPC (experimental, opt-in): pip install 'a2a-lite[grpc]' then agent.run_grpc() / build_grpc_server() — not enabled by agent.run() alone.
• 1.2.1 is a documentation-only patch: refreshed examples and registry README links; runtime behavior and dependency requirements are unchanged from 1.2.0.
• No v0.3 compatibility. See MIGRATION.md.

from a2a_lite import Agent

agent = Agent(name="Bot", description="My bot")

@agent.skill("greet")
async def greet(name: str) -> str:
    return f"Hello, {name}!"

agent.run()

---

Learning, managed runtimes and durability

Start with the executable learning path, then use the runtime and persistence guide for authenticated profiles, real task IDs, rotating credentials, PostgreSQL state/events, durable callback delivery and opt-in checkpoint recovery. Agent.start() returns an explicit runtime handle with readiness, statistics and stop(); the default eight-line API remains available.

from a2a_lite import RuntimeConfig, ClientConfig, PersistenceConfig
# RuntimeConfig(profile="learning" | "authenticated" | "durable")
# @agent.skill("tracked", task=True) exposes a queryable protocol Task.
# @agent.skill("resumable", recoverable=True) requires managed persistence.

Installation

pip install a2a-lite
# or
uv add a2a-lite

Requirements: Python 3.10+ (pulls in a2a-sdk[http-server] >= 1.2.2 automatically)

Optional extras

pip install 'a2a-lite[sqlite]'     # single-host persistent learning examples
pip install 'a2a-lite[postgresql]' # managed durable state/events/callbacks
pip install 'a2a-lite[grpc]'       # experimental gRPC transport
pip install 'a2a-lite[oauth]'      # OAuth2 / JWT (PyJWT)
pip install 'a2a-lite[mcp]'        # MCP tool client
pip install 'a2a-lite[openai]'     # OpenAI skill helper
pip install 'a2a-lite[anthropic]'  # Anthropic skill helper
pip install 'a2a-lite[bedrock]'    # AWS Bedrock skill helper

---

Quick Start

1. Create an agent

from a2a_lite import Agent

agent = Agent(name="Calculator", description="Does math")

@agent.skill("add")
async def add(a: int, b: int) -> int:
    return a + b

@agent.skill("multiply")
async def multiply(a: int, b: int) -> int:
    return a * b

agent.run(port=8787)

2. Test it (no HTTP needed)

from a2a_lite import Agent, AgentTestClient

agent = Agent(name="Calculator", description="Does math")

@agent.skill("add")
async def add(a: int, b: int) -> int:
    return a + b

client = AgentTestClient(agent)
result = client.call("add", a=2, b=3)
assert result == 5

3. Call it over the network

curl -X POST http://localhost:8787/ \
  -H "Content-Type: application/json" \
  -H "A2A-Version: 1.0" \
  -d '{
    "jsonrpc": "2.0",
    "method": "SendMessage",
    "id": "1",
    "params": {
      "message": {
        "role": "ROLE_USER",
        "parts": [{"text": "{\"skill\": \"add\", \"params\": {\"a\": 2, \"b\": 3}}"}],
        "messageId": "msg-1"
      }
    }
  }'

---

Progressive Complexity

Level 1 — Basic Skills

from a2a_lite import Agent

agent = Agent(name="Bot", description="A bot")

@agent.skill("greet")
async def greet(name: str) -> str:
    return f"Hello, {name}!"

agent.run()

Level 2 — Pydantic Models

Pass dicts from callers — they're auto-converted to Pydantic models:

from pydantic import BaseModel

class User(BaseModel):
    name: str
    email: str

@agent.skill("create_user")
async def create_user(user: User) -> dict:
    return {"id": 1, "name": user.name}

Lists of models work too:

from typing import List

@agent.skill("count_users")
async def count_users(users: List[User]) -> int:
    return len(users)

Level 3 — Streaming

Just yield instead of return:

@agent.skill("chat", streaming=True)
async def chat(message: str):
    for word in message.split():
        yield word + " "

Level 4 — Middleware

Cross-cutting concerns without touching skill code:

@agent.middleware
async def log_requests(ctx, next):
    print(f"Calling: {ctx.skill}")
    result = await next()
    print(f"Result: {result}")
    return result

Built-in middleware:

from a2a_lite import logging_middleware, timing_middleware, retry_middleware, rate_limit_middleware

agent.add_middleware(logging_middleware)
agent.add_middleware(timing_middleware)
agent.add_middleware(rate_limit_middleware(max_per_minute=60))
agent.add_middleware(retry_middleware(max_retries=3))

Level 5 — File Handling

Accept and return files through the A2A protocol:

from a2a_lite import FilePart

@agent.skill("summarize")
async def summarize(doc: FilePart) -> str:
    content = await doc.read_text()
    return f"Summary: {content[:100]}..."

Level 6 — Task Tracking

Long-running operations with progress updates:

from a2a_lite import TaskContext

agent = Agent(name="Bot", description="A bot")

@agent.skill("process")
async def process(data: str, task: TaskContext) -> str:
    await task.update("working", "Starting...", progress=0.0)
    for i in range(10):
        await task.update("working", f"Step {i}/10", progress=i/10)
    return "Done!"

Level 7 — Authentication

API keys are hashed in memory using SHA-256 — plaintext keys are never stored.

from a2a_lite import Agent, APIKeyAuth

agent = Agent(
    name="SecureBot",
    description="A secure bot",
    auth=APIKeyAuth(keys=["secret-key-1", "secret-key-2"]),
)

Other auth providers:

from a2a_lite.auth import BearerAuth, OAuth2Auth

# Demo opaque bearer token; use OAuth2Auth below for JWT verification.
demo_tokens = {"example-token": "alice"}
agent = Agent(
    name="Bot", description="A bot",
    auth=BearerAuth(validator=demo_tokens.get),
)

# OAuth2 (requires: pip install 'a2a-lite[oauth]')
agent = Agent(
    name="Bot", description="A bot",
    auth=OAuth2Auth(issuer="https://auth.example.com", audience="my-api"),
)

Skills can receive auth results by type-hinting a parameter as AuthResult:

from a2a_lite.auth import AuthResult

@agent.skill("whoami")
async def whoami(auth: AuthResult) -> dict:
    return {"user": auth.user_id, "scopes": sorted(auth.scopes)}

Level 8 — CORS & Production Mode

agent = Agent(
    name="Bot",
    description="A bot",
    cors_origins=["https://myapp.com", "https://admin.myapp.com"],
    production=True,  # Warns if running over HTTP
)

production=True enables warnings; it does not select an authenticated or durable runtime profile. Use RuntimeConfig and the deployment guide for those guarantees.

Level 10 — Multi-Agent Communication

TaskHandle — track remote tasks

A handle requires a real task ID returned by the peer. Message-only responses cannot be polled or subscribed to; requesting a handle for one raises a clear error instead of inventing an ID. Discovery selects a compatible JSON-RPC interface regardless of card order.

from a2a_lite import Agent, AgentNetwork, TaskHandle

network = AgentNetwork()
network.add("data", "http://localhost:8787")

agent = Agent(name="Orchestrator", description="Delegates to specialist agents", network=network)

@agent.skill("process")
async def process(query: str) -> str:
    # Get a TaskHandle instead of just the result
    handle = await agent.delegate("data", "fetch", query=query, return_handle=True)
    print(f"Task ID: {handle.task_id}")
    print(f"Agent: {handle.agent_url}")
    print(f"Result: {handle.result}")
    return str(handle.result)

Task Lifecycle — poll and cancel remote tasks

Use convenience methods directly on the handle, or via the network by name:

# Poll status directly on the handle
status = await handle.get_status()
status = await handle.get_status(timeout=15.0)

# Or poll via network name
status = await network.get_task("data", handle.task_id)
status = await network.get_task("data", handle.task_id, timeout=15.0)

# Cancel directly on the handle
await handle.cancel()
await handle.cancel(timeout=15.0)

# Or cancel via network name
await network.cancel_task("data", handle.task_id)
await network.cancel_task("data", handle.task_id, timeout=15.0)

Low-level functions are also available:

from a2a_lite import get_remote_task, cancel_remote_task

status = await get_remote_task("http://localhost:8787", task_id)
await cancel_remote_task("http://localhost:8787", task_id)

Agent Card Discovery — inspect before calling

from a2a_lite import discover, AgentCardInfo

# Fetch an agent's capabilities before calling
card = await discover("http://localhost:8787")
print(f"Agent: {card.name} v{card.version}")
print(f"Skills: {[s['name'] for s in card.skills]}")
print(f"Streaming: {card.supports_streaming}")

# Auto-discover when registering
network.add("data", "http://localhost:8787", auto_discover=True)

# Validate skill exists before delegating
result = await agent.delegate("data", "fetch", query="hello", discover=True)

Client-Side SSE Streaming

When calling a remote streaming agent, consume its chunks as they arrive instead of waiting for the full response:

from a2a_lite import Agent, AgentNetwork, stream_remote_skill

# Agent consuming a remote streaming skill
@agent.skill("display")
async def display(topic: str) -> str:
    chunks = []
    stream = await agent.delegate("story", "tell_story", topic=topic, stream=True)
    async for chunk in stream:
        print(chunk, end="", flush=True)
        chunks.append(chunk)
    return "".join(chunks)

Or call directly:

from a2a_lite import stream_remote_skill

async for chunk in stream_remote_skill("http://localhost:8787", "tell_story", {"topic": "dragons"}):
    print(chunk, end="", flush=True)

Per-Task Push Notifications

Per-task callbacks require an authenticated identity, including inline configuration in SendMessage. Agents with auth=None reject callback configuration. Register callbacks for an existing protocol task owned by the authenticated caller:

from a2a_lite import set_task_push_notification, get_task_push_notification, delete_task_push_notification

# task_id must come from result.task.id in an A2A response.
headers = {"X-API-Key": "your-agent-key"}
await set_task_push_notification(
    "https://agent.example/rpc", task_id,
    "https://my-app.com/webhook", token="callback-token", headers=headers,
)
config = await get_task_push_notification("https://agent.example/rpc", task_id, headers=headers)
await delete_task_push_notification("https://agent.example/rpc", task_id, headers=headers)

Register during SendMessage using configuration.taskPushNotificationConfig to avoid missing a fast task's completion. See the complete example. Callbacks carry standard A2A task, status-update or artifact-update payloads with Content-Type: application/a2a+json; the old custom {skill, result, status} envelope is no longer used for per-task notifications. The standalone agent-level WebhookPushNotifier keeps its existing trusted configuration and envelope.

Client-supplied callbacks require public HTTPS. User info, fragments, private/loopback/link-local/shared addresses, unresolved hosts and redirects are rejected. Delivery has a ten-second total deadline, connects to a validated address while preserving TLS verification for the original hostname, and closes the response without buffering its body. Logs omit callback URLs and exception details. Local development should use a public HTTPS reverse proxy or mocked HTTP transport in tests; there is no insecure callback bypass.

Configured authentication now protects all protocol operations, including task reads, cancellation, streaming and push configuration. HTTP authentication failures return 401; gRPC returns UNAUTHENTICATED. Agent-card discovery remains public. With auth=None, messages and tasks retain their public anonymous mode, while dynamic callback operations return UnsupportedOperation until authentication is configured. Custom authentication must return a stable non-empty string AuthResult.user_id; OAuth tokens require a subject or email identity. In-memory SDK stores isolate tasks/configuration by this identity. Custom protocol_task_store implementations must honor the SDK ServerCallContext ownership contract.

agent.push_registry is now the SDK owner-scoped store with async set_info, get_info and delete_info methods; direct access is for trusted server code. Use the authenticated protocol operations above for clients. The exported legacy TaskPushRegistry remains a standalone local utility, but it does not register server callbacks. The internal PushNotificationMiddleware has been removed.

The HTTP app and managed gRPC server close SDK background tasks and callback connections during shutdown. Await server.stop() when using build_grpc_server(). Its card advertises only gRPC and is available through GetExtendedAgentCard.

Level 9 — Lifecycle Hooks

@agent.on_startup
async def startup():
    print("Agent starting...")

@agent.on_shutdown
async def shutdown():
    print("Agent stopping...")

@agent.on_complete
async def notify(skill_name, result, ctx):
    print(f"Skill {skill_name} completed with: {result}")

@agent.on_error
async def handle_error(error: Exception):
    return {"error": str(error), "type": type(error).__name__}

---

Testing

AgentTestClient

Synchronous test client for pytest:

from a2a_lite import Agent, AgentTestClient

agent = Agent(name="Bot", description="Test")

@agent.skill("greet")
async def greet(name: str) -> str:
    return f"Hello, {name}!"

@agent.skill("info")
async def info(name: str, age: int) -> dict:
    return {"name": name, "age": age}


def test_simple_result():
    client = AgentTestClient(agent)
    result = client.call("greet", name="World")
    assert result == "Hello, World!"


def test_dict_result():
    client = AgentTestClient(agent)
    result = client.call("info", name="Alice", age=30)
    assert result.data["name"] == "Alice"
    assert result.data["age"] == 30


def test_list_skills():
    client = AgentTestClient(agent)
    skills = client.list_skills()
    assert "greet" in skills
    assert "info" in skills

TestResult

Every client.call() returns a TestResult:

TestResult supports direct equality for simple values (result == 5), but use .data for subscripting (result.data["key"]).

AsyncAgentTestClient

For async test frameworks:

import pytest
from a2a_lite import AsyncAgentTestClient

@pytest.mark.asyncio
async def test_async():
    client = AsyncAgentTestClient(agent)
    result = await client.call("greet", name="World")
    assert result == "Hello, World!"
    await client.close()

Streaming Tests

def test_streaming():
    client = AgentTestClient(agent)
    results = client.stream("chat", message="hello world")
    assert len(results) == 2

---

CLI

Full reference: docs/cli.md.

a2a-lite init my-agent                          # Scaffold a basic project
a2a-lite create my-agent                        # Full project + tests + Docker
a2a-lite serve agent.py                         # Run an agent from a .py file
a2a-lite inspect http://localhost:8787          # Rich agent card & skills
a2a-lite info http://localhost:8787             # Compact plain-text info
a2a-lite test http://localhost:8787 greet -p name=World
a2a-lite discover http://localhost:8787 http://localhost:8788
a2a-lite doctor                                 # Local env (SDK, extras)
a2a-lite doctor http://localhost:8787           # + verify remote A2A v1.0
a2a-lite version

---

API Reference

Agent

Agent(
    name: str,                          # Required
    description: str,                   # Required
    version: str = "1.0.0",
    url: str = None,                    # Override auto-detected URL
    auth: AuthProvider = None,          # Authentication provider
    task_store: str | TaskStore = None, # Optional local application mirror
    cors_origins: List[str] = None,     # CORS allowed origins
    production: bool = False,           # Enable production warnings
    runtime: RuntimeConfig = None,      # Profile and resource limits
    persistence: PersistenceConfig = None, # Managed SQL state/events/recovery
)

Methods:

Skill Decorator

@agent.skill(
    name: str = None,             # Skill name (defaults to function name)
    description: str = None,      # Human-readable description
    tags: List[str] = None,       # Categorization tags
    streaming: bool = False,      # Enable streaming
    task: bool = False,           # Return a real protocol Task
    recoverable: bool = False,    # Opt into checkpoint recovery with managed persistence
)

Auth Providers

Special Parameter Types

Auto-injected when detected in skill function signatures:

Orchestration

---

Examples

Showcase: Battleship (playable A2A demo)

Repo-root demo (not under packages/python/examples/): human vs bot + agent-vs-agent arena, real A2A v1.0 SendMessage, web UI, no API key.

cd ../../examples/python/11_battleship_lite   # from packages/python
# or from repo root: examples/python/11_battleship_lite
pip install -e ../../../packages/python   # editable a2a-lite from this example directory
python agent.py    # http://localhost:8790/
# python arena.py  # http://localhost:8793/?mode=arena

Full docs: examples/python/11_battleship_lite/README.md

Progressive package examples

Also see the side-by-side Google SDK suite under examples/.

---

A2A Protocol Mapping

Everything maps directly to the underlying protocol — no magic, no lock-in.

Wire methods use A2A v1.0 names (SendMessage, not message/send) and the A2A-Version: 1.0 header.

---

License

MIT