# Tenuo A2A Integration

Source: https://tenuo.ai/a2a

## Overview

Tenuo A2A adds **warrant-based authorization** to agent-to-agent communication. When Agent A delegates a task to Agent B, the warrant specifies exactly what Agent B is allowed to do.

```
┌─────────────┐                     ┌─────────────┐
│   Agent A   │   Task + Warrant    │   Agent B   │
│ (Orchestrator)│──────────────────▶│  (Worker)   │
│             │                     │             │
│             │◀──────────────────  │             │
│             │      Result         │             │
└─────────────┘                     └─────────────┘

Warrant says: "Agent B can only search arxiv.org for this task"
```

**Use cases:**
- Multi-agent systems where agents delegate tasks
- Orchestrators that dispatch work to specialized workers
- Agent networks with least-privilege access control

**Not for:** Single-agent tool enforcement (use `tenuo.openai` or `tenuo.langchain` instead)

---

## Installation

```bash
uv pip install "tenuo[a2a]"
```

---

## Quick Start (Minimal Example)

**Server (Worker Agent):**

```python
from tenuo.a2a import A2AServerBuilder

# Build server with fluent API
server = (A2AServerBuilder()
    .name("Worker")
    .url("https://worker.example.com")
    .key(my_signing_key)                    # Your identity
    .accept_warrants_from(orchestrator_key) # Who can give you tasks
    .build())

@server.skill("echo")
async def echo(msg: str) -> str:
    return f"Echo: {msg}"

# uvicorn server:server.app --port 8000
```

Or use the direct constructor:

```python
from tenuo.a2a import A2AServer

server = A2AServer(
    name="Worker",
    url="https://worker.example.com",
    public_key=my_public_key,
    trusted_issuers=[orchestrator_public_key],
)
```

**Client (Orchestrator):**

```python
from tenuo.a2a import A2AClientBuilder
from tenuo import Warrant

# Create warrant for this task
task_warrant = (Warrant.mint_builder()
    .capability("echo")
    .holder(worker_public_key)
    .ttl(300)
    .mint(orchestrator_key))

# Build client with default warrant
client = (A2AClientBuilder()
    .url("http://localhost:8000")
    .warrant(task_warrant, orchestrator_key)  # Pre-configure for repeated use
    .build())

# Send task (warrant already configured)
result = await client.send_task(
    "hello",
    skill="echo",
    arguments={"msg": "hello"},
)
print(result.output)  # "Echo: hello"
```

Or use the direct constructor:

```python
from tenuo.a2a import A2AClient

client = A2AClient("http://localhost:8000")
result = await client.send_task(
    "hello",
    skill="echo",
    arguments={"msg": "hello"},
    warrant=task_warrant,
    signing_key=orchestrator_key,
)
```

That's it. The warrant proves the orchestrator authorized this specific task.

---

## Full Example (With Constraints)

### Server (Worker)

```python
from tenuo.a2a import A2AServerBuilder
from tenuo.constraints import Subpath, UrlSafe

server = (A2AServerBuilder()
    .name("Research Agent")
    .url("https://research-agent.example.com")
    .key(my_signing_key)
    .accept_warrants_from(orchestrator_public_key)
    .build())

# Register skills with constraint bindings
@server.skill("search_papers", constraints={"sources": UrlSafe()})
async def search_papers(query: str, sources: list[str]) -> list[dict]:
    return await do_search(query, sources)

@server.skill("read_file", constraints={"path": Subpath("/data")})
async def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()

# uvicorn server:server.app --host 0.0.0.0 --port 8000
```

---

## Automated Registration (CSR Handshake)

A2A supports an automated handshake for agent registration, eliminating the need for out-of-band key sharing. This follows the Certificate Signing Request (CSR) pattern.

The connecting agent dynamically generates a self-signed challenge token to cryptographically prove key ownership. The server verifies this signature and uses a registered handler to decide what capabilities to grant, minting a fresh delegation warrant on the fly. 

**Server (Control Plane / Parent Agent):**

```python
from tenuo.a2a.types import VerifiedWarrantRequest

# The handler decides whether to grant the requested capabilities
async def registration_handler(req: VerifiedWarrantRequest, issue):
    if req.verified_key_hex not in ALLOWLIST:
        raise RegistrationDeniedError("Agent not approved")
    
    # Issue a new warrant bound to the requested capabilities
    await issue(capabilities=req.capabilities, ttl=86400) # 24 hrs

server = (A2AServerBuilder()
    .name("Control Plane")
    .url("https://control.example.com")
    .key(server_signing_key) # MUST be a SigningKey to issue warrants
    .trust(server_signing_key.public_key)
    .registration_handler(registration_handler) # Enable handshake
    .build())
```

**Client (Child Agent):**

```python
from tenuo.a2a import A2AClient
from tenuo import SigningKey

client = A2AClient("https://control.example.com")
worker_key = SigningKey.generate() 

# Request a warrant with specific capabilities
# The client automatically generates the self-signed challenge token
warrant = await client.request_warrant(
    signing_key=worker_key,
    capabilities={"search_papers": {}}
)

# You can now immediately use this warrant (and key) for tasks
result = await client.send_task(
    "Search for AI Agents papers",
    skill="search_papers",
    arguments={"query": "AI Agents"},
    warrant=warrant,
    signing_key=worker_key,
)
```

**Note:** Extension data (like AWS Nitro Enclaves or SGX TEE quotes) can be attached to the request via the `extensions` parameter in `request_warrant()` and inspected in the server handler via `req.extensions`.

### Client (Orchestrator)

```python
from tenuo.a2a import A2AClient
from tenuo.constraints import UrlSafe

# Discover agent capabilities
client = A2AClient("https://research-agent.example.com")
card = await client.discover()

# Attenuate warrant for this delegation
task_warrant = (my_warrant
    .grant_builder()
    .capability("search_papers", sources=UrlSafe(allow_domains=["arxiv.org"]))
    .holder(card.public_key)
    .ttl(300)
    .build(my_signing_key))

# Send task with warrant
result = await client.send_task(
    message="Find papers on capability-based security",
    warrant=task_warrant,
    skill="search_papers",
    arguments={"query": "capability-based security", "sources": ["https://arxiv.org"]},
)
```

### Streaming Tasks

For long-running tasks, use streaming to receive incremental updates:

```python
# Stream results as they arrive
async for update in client.send_task_streaming(
    message="Analyze these papers",
    warrant=task_warrant,
    skill="analyze_papers",
    arguments={"paper_ids": ["arxiv:2401.12345"]},
):
    if update.type.value == "status":
        print(f"Status: {update.data.get('status')}")
    elif update.type.value == "message":
        print(f"Chunk: {update.data.get('content')}")
    elif update.type.value == "complete":
        print(f"Done: {update.data.get('output')}")
```

The server emits SSE events for status updates, intermediate messages, and final completion.

**Stream timeout (DoS protection):**

```python
# Default timeout is 300 seconds (5 minutes)
async for update in client.send_task_streaming(
    ...,
    stream_timeout=600.0,  # 10 minute timeout
):
    ...
```

If the stream exceeds `stream_timeout`, a `TimeoutError` is raised. This prevents slow-drip DoS attacks where a malicious server holds connections indefinitely.

---

## Proof-of-Possession (PoP)

Proof-of-Possession adds an additional security layer by requiring the client to prove they control the private key associated with the warrant's holder.

### When to Use PoP

**Require PoP when:**
- Agents communicate over untrusted networks (Internet, shared infrastructure)
- Compliance requires cryptographic proof of authorization
- Protection against warrant theft is critical
- Multi-hop delegation across organizational boundaries

**PoP is optional when:**
- All agents run on trusted infrastructure (same data center, VPC)
- Network isolation provides security (private network, mTLS)
- Performance is critical and risk is low (every extra signature operation matters)

**Never skip PoP when:**
- Agents are on the public Internet
- Warrants have long TTLs (hours/days)
- Untrusted intermediaries exist in the call chain

### How PoP Works

PoP signatures prove that the caller possesses the private key corresponding to the warrant's `sub` (holder) field:

```
┌──────────────────────────────────────────────────────┐
│ Warrant (JWT):                                       │
│   sub: "z6Mk..."  ← Orchestrator's public key       │
│   grants: ["search"]                                 │
│   exp: 1234567890                                    │
│   Signature: <signed by control plane>              │
└──────────────────────────────────────────────────────┘
                       +
┌──────────────────────────────────────────────────────┐
│ PoP Signature:                                       │
│   sign(orchestrator_private_key, "search", args, ts) │
│   → Proves orchestrator controls the private key     │
└──────────────────────────────────────────────────────┘
                       =
               Authorization Proof
```

**What PoP Prevents:**
- **Warrant Theft**: If an attacker intercepts a warrant, they can't use it without the private key
- **Replay Attacks**: Each PoP signature includes a timestamp and is checked once
- **Man-in-the-Middle**: Modified arguments invalidate the PoP signature

### Client Usage

Enable PoP by passing `signing_key` to `send_task()`:

```python
from tenuo.a2a import A2AClient

client = A2AClient("https://worker.example.com")

# Without PoP (only warrant validation)
result = await client.send_task(
    "search for papers",
    warrant=my_warrant,
    skill="search",
    arguments={"query": "papers"},
)

# With PoP (warrant + signature proof)
result = await client.send_task(
    "search for papers",
    warrant=my_warrant,
    skill="search",
    arguments={"query": "papers"},
    signing_key=orchestrator_key,  # ← Proves possession
)
```

Or configure PoP by default using the builder:

```python
from tenuo.a2a import A2AClientBuilder

client = (A2AClientBuilder()
    .url("https://worker.example.com")
    .warrant(my_warrant, orchestrator_key)  # ← Pre-configure PoP
    .build())

# All requests automatically include PoP
result = await client.send_task(
    "search for papers",
    skill="search",
    arguments={"query": "papers"},
)
```

### Server Configuration

Control PoP requirements on the server:

```python
server = A2AServer(
    name="Worker",
    url="https://worker.example.com",
    public_key=worker_public_key,
    trusted_issuers=[control_plane_key],  # Required — warrants must be signed by these keys

    # PoP configuration
    require_pop=True,          # Reject requests without PoP (default: True)
)
```

> **Important:** Always configure `trusted_issuers`. Without it, the builder raises `ValueError`. This ensures only warrants signed by your control plane are accepted — self-signed warrants from attackers are rejected.

**Security Defaults:**
- `trusted_issuers` is **required** (fail-closed)
- `require_pop=True` by default (fail-safe)
- Can be disabled via `TENUO_A2A_REQUIRE_POP=false` environment variable
- If `require_pop=True` but client doesn't provide PoP → `PopRequiredError`

### Performance Impact

Enabling PoP adds two extra Ed25519 signature operations per request on top of warrant verification:

- **Without PoP:** warrant verification only.
- **With PoP:** warrant verification + client-side PoP signing + server-side PoP verification.

All three operations are local and offline. See [Performance Benchmarks](./api-reference#performance-benchmarks) for measured timings.

**Recommendation:** Always use PoP in production unless you have network-level security (mTLS + VPC).

### Error Handling

```python
from tenuo.a2a import PopRequiredError, PopVerificationError

try:
    result = await client.send_task("search", warrant=warrant, skill="search", arguments={})
except PopRequiredError:
    print("Server requires PoP signature - add signing_key parameter")
except PopVerificationError as e:
    print(f"PoP signature invalid: {e}")
    # Possible causes:
    #   - Wrong signing key (not matching warrant.sub)
    #   - Arguments modified after signing
    #   - Clock skew between client/server
```

### Debugging PoP Issues

**Issue:** `PopVerificationError: Signature verification failed`

**Causes:**
1. **Wrong signing key**: Key doesn't match warrant's `sub` field
2. **Modified arguments**: Arguments changed after PoP computation
3. **Clock skew**: Client/server clocks differ significantly

**Debug:**
```python
# Verify signing key matches warrant holder
assert warrant.sub == str(signing_key.public_key)

# Log PoP computation
import logging
logging.getLogger("tenuo.a2a.client").setLevel(logging.DEBUG)
# Shows: "Generated PoP signature for skill 'search'"
```

---

## Server Configuration

```python
server = A2AServer(
    # Required
    name="Agent Name",                    # Display name
    url="https://agent.example.com",      # Public URL (for audience validation)
    public_key=my_public_key,             # This agent's public key
    trusted_issuers=[...],                # List of trusted issuer public keys
    
    # Optional (shown with defaults)
    trust_delegated=True,                 # Accept warrants delegated from trusted issuers
    require_warrant=True,                 # Reject tasks without warrants
    require_audience=True,                # Require warrant audience matches our URL
    check_replay=True,                    # Enforce jti uniqueness
    replay_window=3600,                   # Seconds to remember jti values
    max_chain_depth=10,                   # Maximum delegation chain length
    
    # Audit
    audit_log=sys.stderr,                 # Destination (file, callable, or stderr)
    audit_format="json",                  # "json" or "text"
)
```

### Trust Model

The server trusts warrants based on `trusted_issuers`:

1. **Direct Trust**: Warrant signed by a trusted issuer → accepted
2. **Delegated Trust** (if `trust_delegated=True`): Warrant with valid chain back to trusted issuer → accepted

```
┌─────────────────────┐
│    Trusted Root     │  ← In trusted_issuers
│   (Control Plane)   │
└──────────┬──────────┘
           │ delegates
           ▼
┌─────────────────────┐
│   Orchestrator A    │  ← Warrant signed by root
└──────────┬──────────┘
           │ delegates
           ▼
┌─────────────────────┐
│     Worker B        │  ← Warrant with chain [root → A → B]
└─────────────────────┘
```

### Skill Constraints

Constraints bind warrant parameters to skill parameters:

```python
@server.skill("read_file", constraints={"path": Subpath("/data")})
async def read_file(path: str) -> str:
    # "path" constraint checked against warrant's path constraint
    # Blocked if: warrant allows Subpath("/data") but arg is "/etc/passwd"
    ...
```

**Constraint binding validation** happens at startup:

```python
# This raises ConstraintBindingError at startup:
@server.skill("read_file", constraints={"file_path": Subpath("/data")})  # "file_path" not a param
async def read_file(path: str) -> str:  # param is "path"
    ...
```

---

## Client Configuration

```python
client = A2AClient(
    url="https://agent.example.com",
    
    # Optional
    pin_key="z6Mk...",    # Expected public key (raises KeyMismatchError if different)
    timeout=30.0,          # Request timeout in seconds
)
```

### Key Pinning

Pin the expected public key to prevent TOFU (Trust On First Use) attacks:

```python
# If agent returns different key, raises KeyMismatchError
client = A2AClient(
    "https://research-agent.example.com",
    pin_key="z6MkResearchAgentKey123"  # From your config/secrets
)

card = await client.discover()  # Fails if key doesn't match
```

### Key Format Compatibility

A2A accepts public keys in multiple formats:

```python
# All of these work:
server = (A2AServerBuilder()
    .key(signing_key)  # PublicKey object
    .accept_warrants_from("a1b2c3...")  # Hex (64 chars)
    .accept_warrants_from("z6MkpT...")  # Multibase (base58btc)
    .accept_warrants_from("did:key:z6MkpT...")  # W3C DID
    .build())
```

All formats are automatically normalized for comparison. Multibase and DID support requires `uv pip install base58`.

---

## Agent Card (Discovery)

Agents expose their capabilities via `/.well-known/agent.json`:

```json
{
  "name": "Research Agent",
  "url": "https://research-agent.example.com",
  "skills": [
    {
      "id": "search_papers",
      "name": "Search Papers",
      "x-tenuo-constraints": {
        "sources": {"type": "UrlSafe", "required": true}
      }
    }
  ],
  "x-tenuo": {
    "version": "0.1.0",
    "required": true,
    "public_key": "z6Mk..."
  }
}
```

---

## Delegation Chains

When delegating through multiple agents, the full chain is transmitted as a single header.

The server validates:
1. Root warrant is from a trusted issuer
2. Each link: child issuer = parent holder
3. Skills narrow monotonically (no privilege escalation)
4. Chain depth ≤ `max_chain_depth`

### WarrantStack Transport

The current implementation packs the entire delegation chain into a **single** `X-Tenuo-Warrant` header using WarrantStack encoding, rather than the legacy two-header approach (`X-Tenuo-Warrant` + `X-Tenuo-Warrant-Chain`).

The client encodes the chain with `encode_warrant_stack` and the server decodes it with `decode_warrant_stack_base64`:

```python
from tenuo import encode_warrant_stack

# Client sends delegation chain as a single header
chain = [root_warrant, child_warrant]
stack_b64 = encode_warrant_stack(chain)
# stack_b64 goes in X-Tenuo-Warrant header
# Server automatically detects and unpacks WarrantStack
```

This simplifies proxy and load-balancer configurations (one header to forward instead of two) and avoids ordering ambiguities in multi-hop chains.

---

## Human Approval

Define gates and approvers on the warrant. On retry, attach `SignedApproval` objects via header or param. See [Human Approvals](approvals.md) for minting and signing.

```python
from tenuo.a2a import A2AClient, ApprovalRequiredError, InsufficientApprovalsError
from tenuo.approval import sign_approval

try:
    result = await client.send_task(skill="transfer", arguments={"amount": 5000})
except ApprovalRequiredError as e:
    # -32019 — gate fired; e.data has request_hash, min_approvals, skill
    signed = sign_approval(approval_request, approver_key)  # from your UI flow
    result = await client.send_task(
        skill="transfer",
        arguments={"amount": 5000},
        approvals=[signed],  # encoded into X-Tenuo-Approvals
    )
except InsufficientApprovalsError as e:
    # -32020 — collect additional signatures; check e.data["required"] vs ["received"]
    ...
```

**Wire encoding:** `X-Tenuo-Approvals` = `base64(JSON(["base64(CBOR SignedApproval)", ...]))`. Same outer wrapper as FastAPI.

| A2A JSON-RPC | Wire code | When | Key `data` fields |
|--------------|-----------|------|-------------------|
| **-32019** | 1707 | Gate fired, no approvals | `request_hash`, `min_approvals`, `skill` |
| **-32020** | 1700 | Partial multi-sig | `required`, `received` |
| **-32021** | 1701 | Invalid / malformed approval | `reason` |

> **Note:** A2A `-32002` is **invalid signature** (1100), not approval. Do not reuse MCP's `-32002` semantics on A2A.

---

## Error Handling

All A2A errors inherit from `A2AError` and map to JSON-RPC error codes with canonical wire codes:

```python
from tenuo.a2a import (
    A2AError,
    MissingWarrantError,      # -32001: Warrant required but not provided
    InvalidSignatureError,     # -32002: Signature verification failed
    UntrustedIssuerError,      # -32003: Issuer not in trusted_issuers
    WarrantExpiredError,       # -32004: Warrant has expired
    AudienceMismatchError,     # -32005: Audience doesn't match server URL
    ReplayDetectedError,       # -32006: jti already used
    SkillNotGrantedError,      # -32007: Skill not in warrant grants
    ConstraintViolationError,  # -32008: Argument violates constraint
    ChainInvalidError,         # -32010: Delegation chain validation failed
    KeyMismatchError,          # -32012: Public key doesn't match pinned key
    ApprovalRequiredError,     # -32019: Approval gate fired
    InsufficientApprovalsError,  # -32020: Partial multi-sig
    InvalidApprovalError,      # -32021: Bad signed approval
)

try:
    result = await client.send_task(...)
except SkillNotGrantedError as e:
    print(f"Skill {e.data['skill']} not in granted: {e.data['granted_skills']}")
except A2AError as e:
    print(f"A2A error {e.code}: {e.message}")
```

### Wire Code Support

A2A error responses now include canonical Tenuo wire codes (1000-2199) for cross-protocol compatibility:

```json
{
  "jsonrpc": "2.0",
  "error": {
    "code": -32008,
    "message": "Constraint violation",
    "data": {
      "tenuo_code": 1501,
      "field": "amount",
      "reason": "Value exceeds maximum"
    }
  },
  "id": "task_123"
}
```

This enables:
- **Cross-protocol debugging**: Same wire codes used in HTTP, gRPC, and JSON-RPC
- **Precise error mapping**: JSON-RPC code -32008 maps to canonical code 1501
- **Machine-readable errors**: Clients can programmatically handle specific error types

| A2A JSON-RPC Code | Canonical Wire Code | Name |
|-------------------|---------------------|------|
| -32001 | — (A2A-specific) | Missing warrant |
| -32002 | 1100 | Invalid signature |
| -32003 | 1406 | Untrusted issuer |
| -32004 | 1300 | Warrant expired |
| -32007 | 1500 | Tool not authorized |
| -32008 | 1501 | Constraint violation |
| -32010 | 1405 | Chain invalid |
| -32019 | 1707 | Approval required (gate fired) |
| -32020 | 1700 | Insufficient approvals (partial multi-sig) |
| -32021 | 1701 | Invalid approval |

See [wire format specification](./spec/wire-format-v1#appendix-a-error-code-reference) for the complete list.

---

## Accessing the Warrant

Inside a skill, access the current warrant via context:

```python
from tenuo.a2a import current_task_warrant

@server.skill("my_skill")
async def my_skill(query: str) -> str:
    warrant = current_task_warrant.get()
    if warrant:
        print(f"Warrant issuer: {warrant.iss}")
        print(f"Warrant subject: {warrant.sub}")
    return "done"
```

---

## Audit Logging

Server emits structured audit events:

```python
# JSON format (default)
{"timestamp": "...", "event": "warrant_validated", "skill": "search", "outcome": "allowed", ...}

# Text format
[WARRANT_VALIDATED] search: allowed
```

Custom audit handler:

```python
async def my_audit_handler(event: AuditEvent):
    await send_to_siem(event.to_dict())

server = A2AServer(..., audit_log=my_audit_handler)
```

---

## Example: Full Multi-Agent System

```python
# control_plane.py
from tenuo import SigningKey, Warrant

control_key = SigningKey.from_env("CONTROL_PLANE_KEY")

def issue_orchestrator_warrant(orchestrator_pubkey):
    return (Warrant.mint_builder()
        .capability("search_papers", {})
        .capability("read_file", {"path": Subpath("/data")})
        .holder(orchestrator_pubkey)
        .ttl(86400)  # 24 hours
        .mint(control_key))
```

```python
# orchestrator.py
from tenuo.a2a import A2AClient

async def delegate_research(topic: str, my_warrant, my_key, target_pubkey):
    client = A2AClient("https://research-agent.example.com")
    
    # Attenuate warrant for this specific task
    task_warrant = (my_warrant
        .grant_builder()
        .capability("search_papers", sources=UrlSafe(allow_domains=["arxiv.org"]))
        .holder(target_pubkey)
        .ttl(300)
        .build(my_key))
    
    return await client.send_task(
        message=f"Research: {topic}",
        warrant=task_warrant,
        skill="search_papers",
        arguments={"query": topic, "sources": ["https://arxiv.org"]},
    )
```

```python
# research_agent.py
from tenuo.a2a import A2AServer
from tenuo.constraints import UrlSafe

server = A2AServer(
    name="Research Agent",
    url="https://research-agent.example.com",
    public_key=my_public_key,
    trusted_issuers=[control_plane_public_key],
)

@server.skill("search_papers", constraints={"sources": UrlSafe()})
async def search_papers(query: str, sources: list[str]) -> list[dict]:
    # Only allowed URLs pass through
    return await search_arxiv(query, sources)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(server.app, host="0.0.0.0", port=8000)
```

---

## API Reference

See [API Reference](./api-reference) for complete type signatures.

## Protocol Specification

For the wire format and protocol details, see the [Protocol Spec](./spec/protocol-spec-v1) and [Wire Format](./spec/wire-format-v1).