A2A 1.0 Protocol & HTTP+JSON¶

LitestarA2A (litestar_mcp.a2a.LitestarA2A, enabled via litestar-mcp[a2a]) provides a first-party, Litestar-native transport layer for the official Agent-to-Agent Python SDK (a2a-sdk>=1.1.4). It mounts /.well-known/agent-card.json discovery, A2A 1.0 JSON-RPC 2.0 (POST /a2a), and native A2A 1.0 HTTP+JSON REST routes without requiring Starlette, FastAPI, or Uvicorn.

Quickstart with LitestarA2A¶

Pass an official a2a.types.AgentCard and either an a2a.server.request_handlers.RequestHandler (such as DefaultRequestHandler) or an a2a.server.agent_execution.AgentExecutor directly to LitestarA2A:

from uuid import uuid4

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill, Message, Part, Role
from a2a.utils.errors import UnsupportedOperationError
from litestar import Litestar

from litestar_mcp.a2a import LitestarA2A


class EchoExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        await event_queue.enqueue_event(
            Message(message_id=str(uuid4()), role=Role.ROLE_AGENT, parts=[Part(text=context.get_user_input())])
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise UnsupportedOperationError(message="This agent returns messages without creating tasks")


def build() -> Litestar:
    card = AgentCard(
        name="Support agent",
        description="Answers support questions",
        version="1.0.0",
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        skills=[AgentSkill(id="echo", name="Echo", description="Repeat a message", tags=["example"])],
        capabilities=AgentCapabilities(streaming=True),
        supported_interfaces=[
            AgentInterface(
                url="https://example.com/a2a",
                protocol_binding="JSONRPC",
                protocol_version="1.0",
            )
        ],
    )
    return Litestar(plugins=[LitestarA2A(card, EchoExecutor())])

When you pass an AgentExecutor instance instead of a RequestHandler, LitestarA2A.__init__ automatically wraps it in DefaultRequestHandler(agent_executor=request_handler, task_store=InMemoryTaskStore(), agent_card=agent_card).

A2AConfig Reference¶

Configure mount paths, transport bindings, route security options, caching, and stream cleanup via A2AConfig (litestar_mcp.a2a.A2AConfig):

Agent Card Discovery (/.well-known/agent-card.json)¶

At application initialization (on_app_init), LitestarA2A validates agent_card and pre-renders its ProtoJSON payload:

1. Interface Validation (_validate_card): agent_card.supported_interfaces must advertise at least one absolute http:// or https:// URL for protocol_version="1.0" (or a 1.0.x patch version) with protocol_binding="JSONRPC" or "HTTP+JSON" whose URL path ends with A2AConfig.path. Using a suffix check allows the AgentCard to advertise a public reverse-proxy URL or ASGI root_path prefix.
2. SHA-256 ETag & 304 Not Modified: LitestarA2A computes a quoted SHA-256 ETag ("<sha256>") over the serialized card bytes and sets Cache-Control: public, max-age={agent_card_max_age}. Incoming GET /.well-known/agent-card.json requests with a matching If-None-Match header (using RFC 9110 weak comparison, including W/ prefixes and *) receive an empty 304 Not Modified response.
3. Route Collision Detection: If any existing route on the Litestar application already occupies A2AConfig.path or A2AConfig.agent_card_path, on_app_init raises ValueError("A2A route collision: ...").
4. Skill Metadata & agent_card_route_opt: When bridging agents with agent_to_a2a, attached SkillController instances automatically compile into AgentSkill entries on the card via SkillController.to_agent_skill(). When configuring LitestarA2A directly, you can tag routes and card handlers via agent_card_route_opt (for example agent_card_route_opt={"a2a_skill": ...} or public security policy markers) while keeping execution routes protected by route_opt.

Note: An AgentCard's AgentSkill entries describe high-level A2A task capabilities advertised to peer agents and are distinct from MCP Agent Skills (SEP-2640 SKILL.md bundles served over /mcp via skills/list and skills/load).

Dual Transport Bindings: JSON-RPC 2.0 & HTTP+JSON REST¶

Every request to an A2A execution endpoint must include the A2A-Version: 1.0 header (1.0.x patch versions are also accepted). Missing, malformed, or legacy (0.3) version headers are rejected with VersionNotSupportedError.

1. JSON-RPC 2.0 Binding (POST /a2a and POST /a2a/{tenant:str})¶

The JSON-RPC transport dispatches all 11 A2A 1.0 methods using protobuf JSON (ParseDict / MessageToDict):

JSON-RPC notifications (well-formed requests without an "id" member) are declined with HTTP 204 No Content before version checking and never invoke the handler. Requests with an explicit "id": null are treated as correlated requests and receive a JSON-RPC response.

2. Native HTTP+JSON REST Binding (enable_http_json=True)¶

When A2AConfig.enable_http_json is True (the default), LitestarA2A mounts the full A2A 1.0 HTTP+JSON REST surface under A2AConfig.path (and its tenant-prefixed counterpart /{base_prefix}/{tenant:str}...), returning application/a2a+json on unary responses and structured REST error payloads (build_rest_error_payload) on failures:

"""Snippet: LitestarA2A with A2A 1.0 HTTP+JSON REST routes, tenant header resolution, and context_builder."""

# start-example
from typing import Any
from uuid import uuid4

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.context import ServerCallContext
from a2a.server.events import EventQueue
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill, Message, Part, Role
from a2a.utils.errors import UnsupportedOperationError
from litestar import Litestar, Request, get
from litestar.exceptions import PermissionDeniedException

from litestar_mcp.a2a import A2AConfig, LitestarA2A

ALLOWED_TENANTS = frozenset({"acme", "globex"})


@get(
    "/tickets/{ticket_id:str}",
    opt={
        "a2a_skill": {
            "id": "ticket_lookup",
            "name": "Ticket Lookup",
            "description": "Retrieve tenant support ticket status by identifier.",
            "tags": ["support", "tickets"],
        }
    },
)
async def get_ticket(ticket_id: str) -> dict[str, str]:
    """Retrieve a tenant support ticket by identifier."""
    return {"ticket_id": ticket_id, "status": "open"}


class TenantSupportExecutor(AgentExecutor):
    """Execute A2A requests within the validated tenant context."""

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        """Enqueue a tenant-scoped agent response message."""
        tenant = context.call_context.tenant if context.call_context is not None else "default"
        reply_text = f"[{tenant}] Processed: {context.get_user_input()}"
        await event_queue.enqueue_event(
            Message(
                message_id=str(uuid4()),
                role=Role.ROLE_AGENT,
                parts=[Part(text=reply_text)],
            )
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        """Report cancellation as unsupported for stateless message responses."""
        raise UnsupportedOperationError(message="Cancellation is not supported for stateless messages")


async def authorize_tenant(request: Request[Any, Any, Any], context: ServerCallContext) -> ServerCallContext:
    """Validate the resolved A2A tenant from URL path or X-A2A-Tenant header."""
    del request
    if context.tenant not in ALLOWED_TENANTS:
        raise PermissionDeniedException(detail="Authorized tenant is required")
    context.state["tenant"] = context.tenant
    return context


card = AgentCard(
    name="Multi-Tenant Support Agent",
    description="Handles tenant-isolated support inquiries over JSON-RPC 2.0 and HTTP+JSON REST",
    version="1.0.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[
        AgentSkill(
            id="ticket_lookup",
            name="Ticket Lookup",
            description="Retrieve tenant support ticket status by identifier.",
            tags=["support", "tickets"],
        )
    ],
    capabilities=AgentCapabilities(streaming=True),
    supported_interfaces=[
        AgentInterface(
            url="https://agents.example.com/a2a",
            protocol_binding="JSONRPC",
            protocol_version="1.0",
        ),
        AgentInterface(
            url="https://agents.example.com/a2a",
            protocol_binding="HTTP+JSON",
            protocol_version="1.0",
        ),
    ],
)

config = A2AConfig(
    path="/a2a",
    enable_http_json=True,
    tenant_header="X-A2A-Tenant",
    route_opt={"requires_tenant": True},
    agent_card_route_opt={"a2a_skill": "public_card"},
    context_builder=authorize_tenant,
)

app = Litestar(
    route_handlers=[get_ticket],
    plugins=[LitestarA2A(card, TenantSupportExecutor(), config)],
)
# end-example


def build() -> Litestar:
    """Return the configured Litestar application for snippet smoke testing."""
    return app

Resource Lifetime & Cooperative SSE Cleanup¶

Streaming A2A operations (SendStreamingMessage and SubscribeToTask across both JSON-RPC and HTTP+JSON) run inside a background producer task managed by start_stream, prefetch_stream, and StreamCleanupMiddleware:

1. First-Event Prefetch: prefetch_stream awaits the first event (or early error/header activation) before committing response headers so extension headers and initial setup errors are reflected accurately.
2. Bounded Backpressure & Disconnect Cleanup: A bounded anyio memory object stream (capacity=1) provides backpressure between the RequestHandler async generator and the ASGI response writer. When the client disconnects or the stream completes, A2AConfig.stream_cleanup_timeout (default 5.0 seconds) bounds the wait for the producer's async with aclosing(stream): finalizer.
3. Executor Service Scopes: Because Litestar HTTP route handlers return a ServerSentEvent response object before the stream finishes (and because background tasks may continue after returnImmediately or client disconnect), AgentExecutor implementations must open their own service or Dishka request scopes from the verified context.call_context rather than relying on request-scoped HTTP dependencies.
4. Application Shutdown Hook: LitestarA2A.on_shutdown automatically calls await request_handler.aclose() (when defined) during Litestar shutdown.

Extension Activation (a2a_activated_extensions)¶

A2A 1.0 extensions declared in AgentCard.capabilities.extensions (AgentExtension(uri=..., required=...)) are negotiated per request via the A2A-Extensions HTTP header:

• Required Extensions: Before invoking context_builder or the handler, LitestarA2A checks that every AgentExtension with required=True is present in context.requested_extensions. If any required extension URI is missing, LitestarA2A raises ExtensionSupportRequiredError.
• Activating Requested Extensions: To activate one or more requested extensions for a call, store a set[str] of requested, advertised extension URIs in context.state["a2a_activated_extensions"] inside context_builder or inside your RequestHandler / AgentExecutor before returning a unary result or yielding the first SSE stream event:

  EXTENSION_URI = "https://example.com/extensions/workspace-label"
  
  if EXTENSION_URI in context.requested_extensions:
      context.state["a2a_activated_extensions"] = {EXTENSION_URI}
  

• Response Header Emission: LitestarA2A validates that context.state["a2a_activated_extensions"] is a subset of requested & advertised safe header URIs and emits A2A-Extensions: <comma-separated sorted URIs> on the HTTP response.