> ## Documentation Index
> Fetch the complete documentation index at: https://docs.emergence.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# A2A Protocol Primer

> Agent Cards, JSON-RPC methods, SSE streaming, and the task lifecycle — everything you need to understand agent communication on CRAFT.

# A2A Protocol Primer

Google's A2A protocol (now under the Linux Foundation) is the communication standard that connects all agents on CRAFT. Understanding A2A is prerequisite knowledge for every agent author, regardless of which framework you choose — it defines the contract your agent must honour at its network boundary.

## What A2A Is

A2A is an open, vendor-neutral protocol for agent-to-agent communication. Key properties:

* **Transport**: JSON-RPC 2.0 over HTTP, with Server-Sent Events (SSE) for streaming responses
* **Discovery**: Agent Cards served at a well-known URL describe capabilities
* **Framework-agnostic**: works with Google ADK, Pydantic AI, LangGraph, or any custom implementation
* **Task-centric**: every interaction creates a Task with a defined lifecycle

A2A defines the *protocol*, not the implementation. CRAFT agents are free to use any framework internally; they must speak A2A at their boundary.

<Note>
  CRAFT's reference agents currently serve `protocolVersion: "0.3.0"` (verified
  against the live text2sql agent's `/.well-known/agent-card.json`). The [A2A
  specification](https://a2a-protocol.org/latest/specification/) reached **v1.0**
  in early 2026 — the platform's `a2a-sdk` Python library (`pip install a2a-sdk`)
  will track upstream as the SDK catches up. The `a2a-sdk` provides the canonical
  types and server utilities (`A2AStarletteApplication`, `DefaultRequestHandler`,
  `AgentCard`, etc.).
</Note>

## Agent Cards — Identity and Discovery

An Agent Card is a JSON document that describes what an agent does, what it can accept, and how to reach it. Every CRAFT agent publishes its card at:

```
GET <base-url>/.well-known/agent-card.json
```

This endpoint is **unauthenticated** — any orchestrator can fetch it to understand the agent's capabilities before invoking it.

### Agent Card Structure

```json theme={null}
{
  "name": "Insights Agent",
  "description": "Main analytics agent — plans queries, coordinates sub-agents, analyzes data.",
  "url": "https://<insights-agent-host>",
  "version": "1.4.0",
  "protocolVersion": "0.3.0",
  "provider": {
    "organization": "EmergenceAI",
    "url": "https://emergence.ai"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": false
  },
  "skills": [
    {
      "id": "data_query",
      "name": "Data Query",
      "description": "Query databases using natural language",
      "tags": ["sql", "query", "database"],
      "examples": [
        "Show me Q4 revenue by region",
        "What are the top 10 customers by sales?"
      ]
    },
    {
      "id": "data_analysis",
      "name": "Data Analysis",
      "description": "Analyze data, generate charts, and produce insights",
      "tags": ["analysis", "insights", "charts"]
    }
  ],
  "defaultInputModes": ["text", "data", "file"],
  "defaultOutputModes": ["text", "data", "file"],
  "securitySchemes": {
    "bearer": { "type": "http", "scheme": "bearer" }
  },
  "security": [{ "bearer": [] }]
}
```

### Capabilities

| Capability               | Effect when `true`                                                         |
| ------------------------ | -------------------------------------------------------------------------- |
| `streaming`              | Agent supports `message/stream` (SSE); callers may use real-time streaming |
| `pushNotifications`      | Agent supports webhook push for task completion                            |
| `stateTransitionHistory` | Agent returns full task state transition history in responses              |

Clients should read capabilities from the Agent Card before calling methods that depend on them. Calling `message/stream` against an agent with `streaming: false` returns `UnsupportedOperationError`.

### Skills

Skills are discrete capability declarations within an Agent Card. They help orchestrators understand what tasks the agent can handle and inform LLM routing decisions.

```python theme={null}
# Pydantic AI / a2a SDK — define skills in code
from a2a.types import AgentSkill

skill = AgentSkill(
    id="data_analysis",
    name="Data Analysis",
    description="Performs data analysis tasks related to semiconductors.",
    tags=["semiconductor", "data analysis"],
    examples=["analyze wafer data", "predict defect rates"],
)
```

**Claude Agent SDK and LangGraph** also require explicit skill authoring — add `AgentSkill` objects to your `AgentCard` regardless of framework:

```python theme={null}
# Claude Agent SDK / LangGraph — same AgentSkill definition
from a2a.types import AgentSkill

skill = AgentSkill(
    id="enterprise-analysis",
    name="Enterprise Data Analysis",
    description="Answers business questions by querying governed data sources and generating SQL.",
    tags=["analytics", "sql", "enterprise"],
    examples=["What is our MRR by region?", "Show top 10 customers by revenue last quarter"],
)
```

**Google ADK and `to_a2a()`**: ADK's `to_a2a()` constructs a minimal Agent Card from the agent's `name`, `description`, and `instruction` — but does not populate `AgentSkill` definitions. For production CRAFT registrations, author explicit `AgentSkill` objects even when using ADK, to give orchestrators rich capability descriptions for routing decisions.

**Claude Agent SDK — Wrapping as an A2A Agent:**

The Claude Anthropic SDK does not natively produce A2A `AgentSkill` definitions. Wrap a Claude call inside a standard A2A server:

```python theme={null}
from a2a.types import AgentSkill
import anthropic

client = anthropic.AsyncAnthropic()

skill = AgentSkill(
    id="data-analysis",
    name="Data Analysis",
    description="Analyzes enterprise data using Claude and CRAFT's data catalog.",
    tags=["analytics", "sql", "enterprise"],
    examples=["What is our MRR by region?", "Identify anomalies in yesterday's orders"],
)

# The A2A server calls the Claude API when tasks arrive — see streaming-idioms.mdx
# for the full TaskUpdater pattern
async def handle_task(message: str) -> str:
    response = await client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
    )
    return response.content[0].text
```

**LangGraph — Wrapping as an A2A Agent:**

```python theme={null}
from a2a.types import AgentSkill
from langgraph.prebuilt import create_react_agent

skill = AgentSkill(
    id="graph-workflow",
    name="Graph Workflow",
    description="Executes multi-step analysis workflows using LangGraph state machines.",
    tags=["workflow", "stateful", "multi-step"],
)

# Build the LangGraph agent; expose via A2AStarletteApplication (see streaming-idioms.mdx
# for the TaskUpdater + astream integration pattern)
graph = create_react_agent("anthropic:claude-opus-4-8", tools=[...])
```

## JSON-RPC Methods

The A2A protocol exposes these JSON-RPC methods on your agent's root endpoint (`POST /`):

| Method              | Use Case                                           | Response                            |
| ------------------- | -------------------------------------------------- | ----------------------------------- |
| `message/send`      | Fire-and-forget, await a single complete response  | JSON-RPC response with Task         |
| `message/stream`    | Real-time SSE stream of status and artifact events | SSE event stream                    |
| `tasks/get`         | Poll for current task state                        | Task object                         |
| `tasks/cancel`      | Cancel an in-progress task                         | Acknowledgment                      |
| `tasks/resubscribe` | Reconnect to a dropped SSE stream                  | SSE event stream from current state |

Most CRAFT agents use `message/stream` exclusively — it is the preferred method for all production interactions because it delivers progressive status updates to the user.

## Message Structure

Every request to a CRAFT agent is a **Message** containing one or more **Parts**:

```json theme={null}
{
  "jsonrpc": "2.0",
  "method": "message/stream",
  "params": {
    "message": {
      "role": "user",
      "messageId": "<uuid>",
      "contextId": "<session-id>",
      "parts": [
        { "kind": "text", "text": "Show me Q4 revenue by region" },
        {
          "kind": "data",
          "data": {
            "type": "datasource",
            "resource_uri": "data:acme-corp:ml-project:analytics-db",
            "datasource_type": "database",
            "datasource_name": "Analytics Database",
            "selected_schemas": [
              { "schema_name": "public", "schema_fqn": "analytics-db.analytics_db.public" }
            ]
          }
        }
      ]
    },
    "contextId": "<session-id>"
  },
  "id": "req-1"
}
```

The A2A spec uses camelCase for protocol fields (`messageId`, `contextId`,
`taskId`). The platform's `a2a-sdk` server also accepts snake\_case input
(`message_id`, `context_id`) and normalizes, but responses are always
camelCase — match the spec to keep request/response symmetric.

### Part Types

| Part Type  | `kind` value | Use Case                                                                     |
| ---------- | ------------ | ---------------------------------------------------------------------------- |
| `TextPart` | `"text"`     | Natural language input or agent response                                     |
| `DataPart` | `"data"`     | Structured JSON — datasource context, artifact references, selection context |
| `FilePart` | `"file"`     | Binary content referenced by URI (`asset://artifacts/chart.png`)             |

<Note>
  The `context_id` groups messages into a conversation session. CRAFT agents use this to retrieve conversation history from the task store and maintain coherent multi-turn dialogue.
</Note>

## Task Lifecycle

Every `message/send` or `message/stream` call creates or resumes a **Task**. Tasks follow a strict state machine:

```
submitted ──► working ──► completed
                      ──► failed
             ──► input-required
             ──► canceled
             ──► rejected
```

Terminal states (`completed`, `failed`, `canceled`, `rejected`) are final — a task in a terminal state cannot be restarted. To continue a conversation, the client sends a new message with the same `context_id`.

## SSE Streaming — Event Sequence

When a client calls `message/stream`, the server responds with `Content-Type: text/event-stream` and pushes events as the agent works. Each `data:` line is a complete JSON-RPC 2.0 response object.

### Event Types

| Result Type               | When Used                                            |
| ------------------------- | ---------------------------------------------------- |
| `Task`                    | First event — the created task (state = `submitted`) |
| `TaskStatusUpdateEvent`   | Lifecycle state changes and intermediate messages    |
| `TaskArtifactUpdateEvent` | Completed artifacts (results, charts, files)         |

### Protocol Rules

1. The first event MUST be a `Task` or a `Message`
2. Intermediate events are `TaskStatusUpdateEvent` (state=`working`) and `TaskArtifactUpdateEvent`
3. The final event MUST be `TaskStatusUpdateEvent` with `final: true` and a terminal state
4. The SSE connection closes after the final event

### Wire Example — Text2SQL Stream

The following is a real-world event sequence from the CRAFT text2sql agent:

```
data: {"kind":"status-update","task_id":"t1","final":false,
       "status":{"state":"working","message":{"role":"agent",
       "parts":[{"kind":"text","text":"Analyzing database schema"}]}}}

data: {"kind":"status-update","task_id":"t1","final":false,
       "status":{"state":"working","message":{"role":"agent",
       "parts":[{"kind":"text","text":"Running query"}]}}}

data: {"kind":"artifact-update","task_id":"t1",
       "artifact":{"artifact_id":"abc","name":"query_summary",
       "parts":[{"kind":"text","text":"Top 10 customers by revenue"}]}}

data: {"kind":"artifact-update","task_id":"t1","last_chunk":true,
       "artifact":{"artifact_id":"def","name":"query_results",
       "parts":[{"kind":"data","data":{"type":"artifact",
       "uri":"data:acme:proj:results-parquet",
       "resource_type":"parquet",
       "metadata":{"row_count":10,"columns":["customer_name","total_revenue"]}}}]}}

data: {"kind":"status-update","task_id":"t1","final":true,
       "status":{"state":"completed"}}
```

### Chunked Artifact Streaming

Large artifacts can be delivered in chunks using the `append` and `lastChunk` flags:

```json theme={null}
{"artifactUpdate": {"artifact": {"artifactId": "a1", "parts": [...chunk1...], "append": false, "lastChunk": false}}}
{"artifactUpdate": {"artifact": {"artifactId": "a1", "parts": [...chunk2...], "append": true,  "lastChunk": false}}}
{"artifactUpdate": {"artifact": {"artifactId": "a1", "parts": [...chunk3...], "append": true,  "lastChunk": true}}}
```

## Agent-to-Agent Communication Flow

The following sequence diagram shows how an orchestrator delegates a task to a sub-agent using the A2A protocol.

```mermaid theme={null}
%%{init: {'theme': 'base', 'themeVariables': {'lineColor': '#555555', 'fontFamily': 'sans-serif', 'edgeLabelBackground': '#ffffff'}}}%%
sequenceDiagram
    participant UI as UI / Client
    participant Orch as Orchestrator Agent
    participant Reg as Assets Registry
    participant Sub as Sub-Agent (e.g. text2sql)

    UI->>Orch: POST / message/stream
    note over Orch: Task submitted

    Orch->>Reg: GET /assets/agents (resolve sub-agent URL)
    Reg-->>Orch: agent_card.json (url, capabilities, skills)

    Orch->>Sub: POST / message/stream (A2A JSON-RPC)
    Sub-->>Orch: SSE: Task{state=submitted}
    Sub-->>Orch: SSE: StatusUpdate{state=working, "Analyzing schema"}
    Sub-->>Orch: SSE: ArtifactUpdate{name=query_results, last_chunk=true}
    Sub-->>Orch: SSE: StatusUpdate{state=completed, final=true}

    Orch-->>UI: SSE: StatusUpdate{state=working, "Running analysis"}
    Orch-->>UI: SSE: ArtifactUpdate{chart + summary}
    Orch-->>UI: SSE: StatusUpdate{state=completed, final=true}
```

## Authentication

Agent Cards declare supported authentication schemes in `securitySchemes`. CRAFT agents use JWT bearer tokens — the orchestrator injects them when calling sub-agents.

```json theme={null}
{
  "securitySchemes": {
    "bearer": { "type": "http", "scheme": "bearer", "bearerFormat": "JWT" }
  },
  "security": [{ "bearer": [] }]
}
```

Agents validate the token, extract the `user_id` and `org_id`, and use them to scope all database queries and audit logs. A small `UserContextBuilder` that reads the `Authorization` header and exposes a typed context object to your dependencies is the canonical pattern for Pydantic AI + `a2a` server integrations.

## Reconnection

If an SSE connection drops mid-stream, the client can reconnect using `tasks/resubscribe`:

```json theme={null}
{
  "jsonrpc": "2.0",
  "method": "tasks/resubscribe",
  "params": { "taskId": "task-123" }
}
```

The server resumes the event stream from the task's current state. Your agent implementation does not need to handle this explicitly — `A2AStarletteApplication` and `DefaultRequestHandler` manage reconnection automatically.

## Next Steps

<CardGroup cols={2}>
  <Card title="Your First Agent" icon="play" href="/guides/agent-author/your-first-agent">
    Build your first agent using Google ADK, Pydantic AI, or another supported framework.
  </Card>

  <Card title="Multi-Agent Patterns" icon="diagram-project" href="/guides/agent-author/multi-agent-patterns">
    Orchestration, delegation, and fan-out across multiple A2A agents.
  </Card>
</CardGroup>