Metadata-Version: 2.5
Name: a2a-lite
Version: 1.2.0
Summary: Simplified wrapper for Google's A2A Protocol SDK
Project-URL: Homepage, https://github.com/xvierd/a2a-lite
Project-URL: Repository, https://github.com/xvierd/a2a-lite
Project-URL: Documentation, https://github.com/xvierd/a2a-lite/tree/main/packages/python
Project-URL: Bug Tracker, https://github.com/xvierd/a2a-lite/issues
Project-URL: Changelog, https://github.com/xvierd/a2a-lite/blob/main/ROADMAP.md
Author-email: A2A Lite Contributors <hello@a2a-lite.dev>
License-Expression: MIT
Keywords: a2a,agents,ai,protocol,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: a2a-sdk[http-server]<2.0,>=1.2.2
Requires-Dist: httpx>=0.25.0
Requires-Dist: pydantic>=2.0
Requires-Dist: rich>=13.0
Requires-Dist: starlette>=1.3.1
Requires-Dist: typer>=0.9.0
Requires-Dist: uvicorn>=0.30.0
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.30; extra == 'anthropic'
Provides-Extra: bedrock
Requires-Dist: boto3>=1.35.0; extra == 'bedrock'
Provides-Extra: dev
Requires-Dist: a2a-sdk[encryption,sqlite]<2.0,>=1.2.2; extra == 'dev'
Requires-Dist: cryptography>=50.0.0; extra == 'dev'
Requires-Dist: httpx>=0.25; extra == 'dev'
Requires-Dist: pyjwt[crypto]>=2.15.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest-cov>=4.0; extra == 'dev'
Requires-Dist: pytest>=9.0.3; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Provides-Extra: docs
Requires-Dist: mkdocs-material>=9.0; extra == 'docs'
Requires-Dist: mkdocstrings[python]>=0.24; extra == 'docs'
Provides-Extra: grpc
Requires-Dist: a2a-sdk[grpc]<2.0,>=1.2.2; extra == 'grpc'
Provides-Extra: mcp
Requires-Dist: mcp<2.0,>=1.30.0; extra == 'mcp'
Provides-Extra: oauth
Requires-Dist: cryptography>=50.0.0; extra == 'oauth'
Requires-Dist: pyjwt[crypto]>=2.15.0; extra == 'oauth'
Provides-Extra: openai
Requires-Dist: openai>=1.0; extra == 'openai'
Provides-Extra: postgresql
Requires-Dist: a2a-sdk[encryption,postgresql]<2.0,>=1.2.2; extra == 'postgresql'
Requires-Dist: cryptography>=50.0.0; extra == 'postgresql'
Provides-Extra: sqlite
Requires-Dist: a2a-sdk[encryption,sqlite]<2.0,>=1.2.2; extra == 'sqlite'
Requires-Dist: cryptography>=50.0.0; extra == 'sqlite'
Description-Content-Type: text/markdown

# A2A Lite — Python

[![PyPI](https://img.shields.io/pypi/v/a2a-lite?label=PyPI&logo=pypi&logoColor=white)](https://pypi.org/project/a2a-lite/)
[![GitHub](https://img.shields.io/badge/GitHub-a2a--lite-blue?logo=github)](https://github.com/xvierd/a2a-lite)

> **A2A Lite is designed for learning and prototyping.** It's the friendly on-ramp to Google's A2A Protocol — get familiar with agent-to-agent concepts with minimal boilerplate before diving into the full [google/a2a-python](https://github.com/a2aproject/a2a-python) SDK. Perfect for courses, POCs, and demos.

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

Wraps the official [A2A Python SDK](https://github.com/a2aproject/a2a-python) with a simple, decorator-based API. 100% protocol-compatible.

> **v1.2.0 — 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.
> - **No v0.3 compatibility.** See [MIGRATION.md](../../MIGRATION.md).

```python
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](../../examples/python/12_production_path), then use the [runtime and persistence guide](docs/production.md) 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.

```python
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

```bash
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

```bash
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

```python
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)

```python
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

```bash
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

```python
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:

```python
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:

```python
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`:

```python
@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:

```python
@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:

```python
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:

```python
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:

```python
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 7 — Authentication

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

```python
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:

```python
from a2a_lite.auth import BearerAuth, OAuth2Auth

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

# 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`:

```python
from a2a_lite.auth import AuthResult

@agent.skill("whoami")
async def whoami(auth: AuthResult) -> dict:
    return {"user": auth.identity, "scheme": auth.scheme}
```

### Level 8 — CORS & Production Mode

```python
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 — 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.

```python
from a2a_lite import Agent, AgentNetwork, TaskHandle

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

agent = Agent(name="Orchestrator", 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:

```python
# 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:

```python
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

```python
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:

```python
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 = []
    async for chunk in agent.delegate("story", "tell_story", topic=topic, stream=True):
        print(chunk, end="", flush=True)
        chunks.append(chunk)
    return "".join(chunks)
```

Or call directly:
```python
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:

```python
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](examples/18_per_task_push.py). 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

```python
@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:

```python
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`:

| 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 for simple values (`result == 5`), but use `.data` for subscripting (`result.data["key"]`).

### AsyncAgentTestClient

For async test frameworks:

```python
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

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

---

## CLI

Full reference: [docs/cli.md](docs/cli.md).

```bash
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

```python
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
)
```

**Methods:**

| Method | Description |
|--------|-------------|
| `@agent.skill(name, **config)` | Register a skill via decorator |
| `@agent.middleware` | Register middleware via decorator |
| `agent.add_middleware(fn)` | Register middleware function |
| `@agent.on_complete` | Register completion hook |
| `@agent.on_startup` | Register startup hook |
| `@agent.on_shutdown` | Register shutdown hook |
| `@agent.on_error` | Register error handler |
| `agent.run(port=8787)` | Start HTTP server (JSON-RPC + REST + agent card) |
| `agent.get_app()` | Get the ASGI app (for custom deployment) |
| `agent.build_agent_card(...)` | Build the v1.0 agent card (`grpc_url=` adds a GRPC interface) |
| `agent.build_grpc_server(...)` | **Experimental** — build a `grpc.aio` server (`a2a-lite[grpc]`) |
| `agent.run_grpc(port=50051)` | **Experimental** — run gRPC only (standalone; pair with `run()` for HTTP) |
| `agent.delegate(...)` / `AgentNetwork` | Call remote agents over A2A |

### Skill Decorator

```python
@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
)
```

### 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 (requires `a2a-lite[oauth]`) |
| `NoAuth()` | No auth (default) |

### Special Parameter Types

Auto-injected when detected in skill function signatures:

| Type | Description |
|------|-------------|
| `TaskContext` | Task lifecycle management using the real protocol task ID; no separate `task_store` required |
| `AuthResult` | Authentication result injection |
| `FilePart` | File upload handling |
| `DataPart` | Structured data handling |

### Orchestration

| Export | Description |
|--------|-------------|
| `AgentNetwork` | Registry of named remote agents |
| `TaskHandle` | Handle to a remote task with `task_id` and `result` |
| `AgentCardInfo` | Parsed agent card with `name`, `version`, `skills`, `supports_streaming` |
| `discover(url)` | Fetch a remote agent's card from `/.well-known/agent-card.json` |
| `get_remote_task(url, task_id)` | Poll the status of a remote task |
| `cancel_remote_task(url, task_id)` | Request cancellation of a remote task |
| `stream_remote_skill(url, skill, params)` | Stream SSE chunks from a remote agent skill |

---

## 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.

```bash
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](../../examples/python/11_battleship_lite/README.md)

### Progressive package examples

| Example | What it shows |
|---------|---------------|
| [01_hello_world.py](examples/01_hello_world.py) | Simplest agent (8 lines) |
| [02_calculator.py](examples/02_calculator.py) | Multiple skills |
| [03_async_agent.py](examples/03_async_agent.py) | Async operations & lifecycle hooks |
| [04_multi_agent/](examples/04_multi_agent) | Two agents communicating |
| [05_with_llm.py](examples/05_with_llm.py) | OpenAI / Anthropic integration |
| [06_pydantic_models.py](examples/06_pydantic_models.py) | Auto Pydantic conversion |
| [06_task_handle_discovery.py](examples/06_task_handle_discovery.py) | TaskHandle + card discovery |
| [07_middleware.py](examples/07_middleware.py) | Middleware pipeline |
| [07_client_streaming.py](examples/07_client_streaming.py) | Client-side SSE streaming |
| [08_streaming.py](examples/08_streaming.py) | Streaming responses |
| [09_testing.py](examples/09_testing.py) | Built-in TestClient |
| [10_file_handling.py](examples/10_file_handling.py) | File upload & processing |
| [11_task_tracking.py](examples/11_task_tracking.py) | Progress updates |
| [12_with_auth.py](examples/12_with_auth.py) | Authentication |
| [13_mcp_tools.py](examples/13_mcp_tools.py) | MCP server integration |
| [14_multi_agent_network.py](examples/14_multi_agent_network.py) | AgentNetwork + delegate() |
| [15_llm_openai.py](examples/15_llm_openai.py) | OpenAI LLM skill |
| [16_llm_anthropic.py](examples/16_llm_anthropic.py) | Anthropic LLM skill |
| [17_llm_bedrock.py](examples/17_llm_bedrock.py) | AWS Bedrock LLM skill |
| [18_per_task_push.py](examples/18_per_task_push.py) | Per-task push notifications |
| [19_capability_negotiation.py](examples/19_capability_negotiation.py) | Capability negotiation |
| [20_streaming_negotiation.py](examples/20_streaming_negotiation.py) | Streaming negotiation |
| [**Battleship showcase**](../../examples/python/11_battleship_lite/) | Playable multi-agent A2A demo + UI |

Also see the side-by-side Google SDK suite under [examples/](../../examples/).

---

## A2A Protocol Mapping

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

| A2A Lite | A2A Protocol |
|----------|--------------|
| `@agent.skill()` | Agent Skills |
| `agent.run()` | JSON-RPC + REST (HTTP+JSON) server |
| `agent.run_grpc()` | gRPC transport (**experimental**, extra `[grpc]`) |
| `streaming=True` | SSE / `SendStreamingMessage` |
| `TaskContext.update()` | Task lifecycle states |
| `FilePart` | A2A File parts |
| `DataPart` | A2A Data parts |
| `Artifact` | A2A Artifacts |
| `APIKeyAuth` / `BearerAuth` / `OAuth2Auth` | `securitySchemes` on the agent card |
| `AgentNetwork` / `delegate()` | Remote `SendMessage` |
| `TaskHandle` | Remote task id + result |
| `discover(url)` | `GET /.well-known/agent-card.json` |
| `get_remote_task` / `cancel_remote_task` | `GetTask` / `CancelTask` |
| `stream_remote_skill` | Client-side SSE consumer |

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

---

## License

MIT
