orcaqubits/a2a-server

A2A Server Implementation

Before writing code

Fetch live docs:

1. Fetch https://a2a-protocol.org/latest/specification/ for the server-side protocol requirements
2. Web-search site:github.com a2aproject a2a-python server or a2aproject a2a-js server for SDK server classes
3. Web-search site:github.com a2aproject a2a-samples server for reference server implementations
4. Fetch the target SDK README for server setup patterns

Conceptual Architecture

What an A2A Server Does

An A2A server is the agent's network endpoint that:

1. Serves the Agent Card at /.well-known/agent-card.json
2. Accepts JSON-RPC 2.0 requests at the agent's URL
3. Routes requests to the appropriate method handler
4. Manages task state (creation, updates, terminal states)
5. Executes the agent's logic (the "brain" — typically an LLM or deterministic logic)
6. Returns responses in the correct JSON-RPC format

Request Processing Flow

HTTP POST → JSON-RPC parse → Method routing → Handler execution → Task state update → JSON-RPC response

Methods to Implement

Every A2A server must handle:

| Method | Required | Description | |--------|----------|-------------| | message/send | Yes | Receive a message, create/update task, return result | | message/stream | If streaming | Same as send but returns SSE stream | | tasks/get | Yes | Return task by ID | | tasks/cancel | Yes | Cancel a task | | tasks/resubscribe | If streaming | Re-subscribe to task's SSE stream | | Push notification methods | If supported | Configure/manage push notification callbacks |

Task State Management

The server is responsible for task state transitions:

• Create tasks in submitted state when receiving new messages
• Transition to working when processing begins
• Transition to input-required when more input is needed
• Transition to terminal state (completed, failed, canceled, rejected) when done
• Never transition from a terminal state to a non-terminal state

Handler Pattern

The SDK typically provides a handler interface you implement:

class MyAgentHandler:
    async def handle_message(task_id, message) -> TaskResult:
        # Your agent logic here
        # 1. Parse the input message parts
        # 2. Process with LLM or business logic
        # 3. Return result with updated task state

Server Components

1. HTTP Server — Handles HTTP requests (FastAPI, Express, built-in SDK server)
2. JSON-RPC Router — Parses JSON-RPC and routes to method handlers
3. Task Store — Persists task state (in-memory for dev, database for production)
4. Agent Handler — Your custom logic that processes tasks
5. Agent Card Endpoint — Serves the discovery document

Task Storage

• In-memory — Dict/Map of task IDs to task objects (development only)
• Redis — Fast key-value storage for task state (good for single-region)
• Database — PostgreSQL, MongoDB, etc. (production, multi-region)
• SDK built-in — Some SDKs provide task store abstractions

Concurrency

A2A servers must handle concurrent requests:

• Multiple tasks running simultaneously
• Same task receiving updates while processing
• Streaming responses held open while processing continues
• Proper locking/synchronization for task state updates

Best Practices

• Use the SDK's built-in server utilities rather than implementing raw JSON-RPC
• Implement proper task state validation — reject invalid transitions
• Add request logging with task IDs for debugging
• Set reasonable timeouts for long-running tasks
• Handle graceful shutdown — complete or cancel in-flight tasks
• Return proper JSON-RPC errors for invalid requests
• Validate incoming messages against expected input modes
• Use async/await for I/O-bound operations (LLM calls, external APIs)

Fetch the SDK documentation for exact server class names, constructor parameters, middleware patterns, and handler interfaces before implementing.