the-nexus/docs/A2A_PROTOCOL.md

# A2A Protocol for Fleet-Wizard Delegation

Implements Google's [Agent2Agent (A2A) Protocol v1.0](https://github.com/google/A2A) for the Timmy Foundation fleet.

## What This Is

Instead of passing notes through humans (Telegram, Gitea issues), fleet wizards can now discover each other's capabilities and delegate tasks autonomously through a machine-native protocol.

```
┌─────────┐    A2A Protocol    ┌─────────┐
│  Timmy  │ ◄────────────────► │  Ezra   │
│  (You)  │   JSON-RPC / HTTP  │ (CI/CD) │
└────┬────┘                    └─────────┘
     │  ╲                          ╲
     │   ╲  Agent Card Discovery    ╲  Task Delegation
     │    ╲   GET /agent.json        ╲  POST /a2a/v1
     ▼     ▼                         ▼
┌──────────────────────────────────────────┐
│          Fleet Registry                   │
│     config/fleet_agents.json              │
└──────────────────────────────────────────┘
```

## Components

| File | Purpose |
|------|---------|
| `nexus/a2a/types.py` | A2A data types — Agent Card, Task, Message, Part, JSON-RPC |
| `nexus/a2a/card.py` | Agent Card generation from `~/.hermes/agent_card.yaml` |
| `nexus/a2a/client.py` | Async client for sending tasks to other agents |
| `nexus/a2a/server.py` | FastAPI server for receiving A2A tasks |
| `nexus/a2a/registry.py` | Fleet agent discovery (local file + Gitea backends) |
| `bin/a2a_delegate.py` | CLI tool for fleet delegation |
| `config/agent_card.example.yaml` | Example agent card config |
| `config/fleet_agents.json` | Fleet registry with all wizards |

## Quick Start

### 1. Configure Your Agent Card

```bash
cp config/agent_card.example.yaml ~/.hermes/agent_card.yaml
# Edit with your agent name, URL, skills, and auth
```

### 2. List Fleet Agents

```bash
python bin/a2a_delegate.py list
```

### 3. Discover Agents by Skill

```bash
python bin/a2a_delegate.py discover --skill ci-health
python bin/a2a_delegate.py discover --tag devops
```

### 4. Send a Task

```bash
python bin/a2a_delegate.py send --to ezra --task "Check CI pipeline health"
python bin/a2a_delegate.py send --to allegro --task "Analyze the codebase" --wait
```

### 5. Fetch an Agent Card

```bash
python bin/a2a_delegate.py card --agent ezra
```

## Programmatic Usage

### Client (Sending Tasks)

```python
from nexus.a2a.client import A2AClient, A2AClientConfig
from nexus.a2a.types import Message, Role, TextPart

config = A2AClientConfig(auth_token="your-token", timeout=30.0, max_retries=3)
client = A2AClient(config=config)

try:
    # Discover agent
    card = await client.get_agent_card("https://ezra.example.com")
    print(f"Found: {card.name} with {len(card.skills)} skills")

    # Delegate task
    task = await client.delegate(
        "https://ezra.example.com/a2a/v1",
        text="Check CI pipeline health",
        skill_id="ci-health",
    )

    # Wait for result
    result = await client.wait_for_completion(
        "https://ezra.example.com/a2a/v1",
        task.id,
    )
    print(f"Result: {result.artifacts[0].parts[0].text}")

    # Audit log
    for entry in client.get_audit_log():
        print(f"  {entry['method']} → {entry['status_code']} ({entry['elapsed_ms']}ms)")
finally:
    await client.close()
```

### Server (Receiving Tasks)

```python
from nexus.a2a.server import A2AServer
from nexus.a2a.types import AgentCard, Task, AgentSkill, TextPart, Artifact, TaskStatus, TaskState

# Define your handler
async def ci_handler(task: Task, card: AgentCard) -> Task:
    # Do the work
    result = "CI pipeline healthy: 5/5 passed"

    task.artifacts.append(
        Artifact(parts=[TextPart(text=result)], name="ci_report")
    )
    task.status = TaskStatus(state=TaskState.COMPLETED)
    return task

# Build agent card
card = AgentCard(
    name="Ezra",
    description="CI/CD specialist",
    skills=[AgentSkill(id="ci-health", name="CI Health", description="Check CI", tags=["ci"])],
)

# Start server
server = A2AServer(card=card, auth_token="your-token")
server.register_handler("ci-health", ci_handler)
await server.start(host="0.0.0.0", port=8080)
```

### Registry (Agent Discovery)

```python
from nexus.a2a.registry import LocalFileRegistry

registry = LocalFileRegistry()  # Reads config/fleet_agents.json

# List all agents
for agent in registry.list_agents():
    print(f"{agent.name}: {agent.description}")

# Find agents by capability
ci_agents = registry.list_agents(skill="ci-health")
devops_agents = registry.list_agents(tag="devops")

# Get endpoint
url = registry.get_endpoint("ezra")
```

## A2A Protocol Reference

### Endpoints

| Endpoint | Method | Purpose |
|----------|--------|---------|
| `/.well-known/agent-card.json` | GET | Agent Card discovery |
| `/agent.json` | GET | Agent Card fallback |
| `/a2a/v1` | POST | JSON-RPC endpoint |
| `/a2a/v1/rpc` | POST | JSON-RPC alias |

### JSON-RPC Methods

| Method | Purpose |
|--------|---------|
| `SendMessage` | Send a task and get a Task object back |
| `GetTask` | Get task status by ID |
| `ListTasks` | List tasks (cursor pagination) |
| `CancelTask` | Cancel a running task |
| `GetAgentCard` | Get the agent's card via RPC |

### Task States

| State | Terminal? | Meaning |
|-------|-----------|---------|
| `TASK_STATE_SUBMITTED` | No | Task acknowledged |
| `TASK_STATE_WORKING` | No | Actively processing |
| `TASK_STATE_COMPLETED` | Yes | Success |
| `TASK_STATE_FAILED` | Yes | Error |
| `TASK_STATE_CANCELED` | Yes | Canceled |
| `TASK_STATE_INPUT_REQUIRED` | No | Needs more input |
| `TASK_STATE_REJECTED` | Yes | Agent declined |

### Part Types (discriminated by JSON key)

- `TextPart` — `{"text": "hello"}`
- `FilePart` — `{"raw": "base64...", "mediaType": "image/png"}` or `{"url": "https://..."}`
- `DataPart` — `{"data": {"key": "value"}}`

## Authentication

Agents declare auth in their Agent Card. Supported schemes:
- **Bearer token**: `Authorization: Bearer <token>`
- **API key**: `X-API-Key: <token>` (or custom header name)

Configure in `~/.hermes/agent_card.yaml`:

```yaml
auth:
  scheme: "bearer"
  token_env: "A2A_AUTH_TOKEN"  # env var containing the token
```

## Fleet Registry

The fleet registry (`config/fleet_agents.json`) lists all wizards and their capabilities. Agents can be registered via:

1. **Local file** — `LocalFileRegistry` reads/writes JSON directly
2. **Gitea** — `GiteaRegistry` stores cards in a repo for distributed discovery

## Testing

```bash
pytest tests/test_a2a.py -v
```

Covers:
- Type serialization roundtrips
- Agent Card building from YAML
- Registry operations (register, list, filter)
- Server integration (SendMessage, GetTask, ListTasks, CancelTask)
- Authentication (required, success)
- Custom handler routing
- Error handling

## Phase Status

- [x] Phase 1 — Agent Card & Discovery
- [x] Phase 2 — Task Delegation
- [x] Phase 3 — Security & Reliability

## Linked Issue

[#1122](https://forge.alexanderwhitestone.com/Timmy_Foundation/the-nexus/issues/1122)