a2a-lite 0.2.1

Simplified wrapper for Google's A2A Protocol SDK

A2A Lite - Python

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

Wraps the official A2A Python SDK with a simple, intuitive API.

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()

---

Installation

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

Requirements: Python 3.10+

---

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

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

curl -X POST http://localhost:8787/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "id": "1",
    "params": {
      "message": {
        "role": "user",
        "parts": [{"type": "text", "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 (Just Works)

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:
    # 'user' is already a User instance — auto-converted from 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)

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

Level 4: Middleware

@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.use(logging_middleware)
agent.use(timing_middleware)
agent.use(rate_limit_middleware(max_per_minute=60))
agent.use(retry_middleware(max_retries=3))

Level 5: Human-in-the-Loop

from a2a_lite import InteractionContext

@agent.skill("wizard")
async def wizard(ctx: InteractionContext) -> dict:
    name = await ctx.ask("What's your name?")
    role = await ctx.ask("Role?", options=["Dev", "Manager"])

    if await ctx.confirm(f"Create {name} as {role}?"):
        return {"created": name}
    return {"cancelled": True}

Level 6: File Handling

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 7: Task Tracking

from a2a_lite import TaskContext

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

@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 8: Authentication

from a2a_lite import Agent, APIKeyAuth

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

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

Other auth providers:

from a2a_lite.auth import BearerAuth, OAuth2Auth

# Bearer/JWT
agent = Agent(
    name="Bot", description="A bot",
    auth=BearerAuth(secret="your-jwt-secret"),
)

# OAuth2
agent = Agent(
    name="Bot", description="A bot",
    auth=OAuth2Auth(issuer="https://auth.example.com", audience="my-api"),
)

Level 9: CORS and 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
)

Level 10: Webhooks

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

---

Testing

AgentTestClient

The synchronous test client for use with 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")
    # Simple values support direct equality
    assert result == "Hello, World!"


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


def test_text_access():
    client = AgentTestClient(agent)
    result = client.call("greet", name="World")
    # .text gives the raw string
    assert result.text == '"Hello, World!"'


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 with:

Property	Description
.data	Parsed Python object (dict, list, int, str, etc.)
.text	Raw text string from the response
.json()	Parse text as JSON (raises on invalid JSON)
.raw_response	Full A2A response dict

TestResult supports direct equality comparison 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

---

Task Store

The TaskStore provides async-safe task lifecycle management:

from a2a_lite.tasks import TaskStore, TaskStatus

store = TaskStore()

# All operations are async and thread-safe
task = await store.create(task_id="task-1", skill="process")
task = await store.get("task-1")
await store.update("task-1", status=TaskStatus.WORKING, progress=0.5)
tasks = await store.list()
await store.delete("task-1")

---

Agent Discovery

Find agents on your local network via mDNS:

from a2a_lite import AgentDiscovery

# Advertise your agent
agent.run(port=8787, enable_discovery=True)

# Discover other agents
discovery = AgentDiscovery()
agents = await discovery.discover(timeout=5.0)
for a in agents:
    print(f"{a.name} at {a.url}")

---

CLI

a2a-lite init my-agent          # Create new project
a2a-lite serve agent.py         # Run agent from file
a2a-lite serve agent.py -r      # Run with hot reload
a2a-lite inspect http://...     # View agent capabilities
a2a-lite test http://... skill  # Test a skill
a2a-lite discover               # Find local agents
a2a-lite version                # Show version

---

Full 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, # "memory" or custom TaskStore
    cors_origins: List[str] = None,     # CORS allowed origins
    production: bool = False,           # Enable production warnings
    enable_discovery: bool = False,     # mDNS discovery
)

Methods:

Method	Description
@agent.skill(name, **config)	Register a skill via decorator
@agent.middleware	Register middleware via decorator
agent.use(middleware)	Register middleware function
@agent.on_complete	Register completion hook
agent.run(port=8787)	Start the server
agent.get_app()	Get the ASGI app (for custom deployment)

Skill Decorator

@agent.skill(
    name: str,                    # Skill name (required)
    description: str = None,      # Human-readable description
    tags: List[str] = None,       # Categorization tags
    streaming: bool = False,      # Enable streaming
)

Auth Providers

Provider	Usage
APIKeyAuth(keys=[...])	API key auth (keys hashed with SHA-256)
BearerAuth(secret=...)	JWT/Bearer token auth
OAuth2Auth(issuer=..., audience=...)	OAuth2 auth
NoAuth()	No auth (default)

Special Parameter Types

These are auto-injected when detected in skill signatures:

Type	Description
TaskContext	Task lifecycle management (requires task_store)
InteractionContext	Human-in-the-loop interactions
FilePart	File upload handling

---

Examples

Example	What it shows
01_hello_world.py	Simplest agent (8 lines)
02_calculator.py	Multiple skills
06_pydantic_models.py	Auto Pydantic conversion
08_streaming.py	Streaming responses
09_testing.py	Testing your agents
11_human_in_the_loop.py	Ask user questions
12_file_handling.py	Handle files
13_task_tracking.py	Progress updates
14_with_auth.py	Authentication

---

100% A2A Protocol Compatible

A2A Lite wraps the official A2A Python SDK. Every feature maps to real A2A protocol concepts:

A2A Lite	A2A Protocol
@agent.skill()	Agent Skills
streaming=True	SSE Streaming
InteractionContext.ask()	input-required state
TaskContext.update()	Task lifecycle states
FilePart	A2A File parts
APIKeyAuth / BearerAuth	Security schemes

---

License

Apache 2.0