A2A 1.0, Protocol Bridges & Transports API¶

API reference for the Agent-to-Agent (A2A 1.0) protocol plugin (A2AConfig, LitestarA2A), bidirectional MCP and A2A protocol bridges (agent_to_mcp, mcp_to_tools, discover_mcp_tools, agent_to_a2a), in-process stdio serving (MCPStdioContext, ASGIStreamingTransport, run_stdio_async, run_stdio), the stdio-to-HTTP proxy bridge (run_stdio_streamable_http_bridge, run_bridge), and MCPStreamableHTTPClient.

For architectural guides, see Protocol Bridges, A2A 1.0 Protocol & HTTP+JSON Binding, Multi-Tenant A2A Routing, and Stdio, HTTP Bridge & Tool Spec.

A2A 1.0 Plugin & Configuration (`litestar_mcp.a2a`)¶

A2AConfig ¶

Configure the Litestar routes wrapping the official A2A SDK.

The context builder receives the request and prepared SDK context, including the requested tenant and protocol metadata. It may asynchronously authorize that tenant and return an authorized context; request values never overwrite its result.

Handlers report extension activation through the context state's a2a_activated_extensions set of requested, advertised URI strings. Set it before returning a result or yielding the first stream event: response headers cannot reflect later activation.

stream_cleanup_timeout bounds the wait for cooperative producer cleanup after a response ends. It must be positive and finite; expiry is logged as incomplete cleanup. Application finalizers must themselves tolerate cancellation.

Parameters:

Name	Type	Description	Default
`path`	`str`	Mount path of the A2A JSON-RPC route. Must start with `/`; a trailing slash is removed.	`'/a2a'`
`agent_card_path`	`str`	Path that serves the agent card. Must start with `/`; a trailing slash is removed.	`'/.well-known/agent-card.json'`
`guards`	`Sequence[Guard]`	Litestar guards applied to the JSON-RPC route. The agent card route is public.	`()`
`route_opt`	`dict[str, Any]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`
`agent_card_route_opt`	`dict[str, Any]`	dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)	`<class 'dict'>`
`tenant_header`	`str \| None`	Optional HTTP header name used as a fallback tenant source when neither the request payload nor the URL path specifies a tenant.	`'X-A2A-Tenant'`
`enable_http_json`	`bool`	Mount A2A 1.0 HTTP+JSON REST transport routes alongside the JSON-RPC route.	`True`
`context_builder`	`Callable[[Request[Any, Any, Any], ServerCallContext], ServerCallContext \| Awaitable[ServerCallContext]] \| None`	Optional `(request, context)` callback that authorizes the prepared SDK call context and returns it, directly or as an awaitable.	`None`
`include_in_schema`	`bool`	Include A2A routes in the OpenAPI schema.	`False`
`agent_card_max_age`	`int`	`Cache-Control` `max-age` for the agent card, in seconds. Must not be negative.	`300`
`stream_cleanup_timeout`	`float`	Seconds to wait for producer cleanup after a response ends.	`5.0`

Attributes:

Name	Type	Description
`path`	`str`	Mount path of the A2A JSON-RPC route. Must start with `/`; a trailing slash is removed.
`agent_card_path`	`str`	Path that serves the agent card. Must start with `/`; a trailing slash is removed.
`guards`	`Sequence[Guard]`	Litestar guards applied to the JSON-RPC route. The agent card route is public.
`route_opt`	`dict[str, Any]`	Extra `opt` entries merged into the JSON-RPC and HTTP+JSON routes, for example an authentication policy.
`agent_card_route_opt`	`dict[str, Any]`	Extra `opt` entries merged into the public agent card route.
`tenant_header`	`str \| None`	Optional HTTP header name used as a fallback tenant source when neither the request payload nor the URL path specifies a tenant.
`enable_http_json`	`bool`	Mount A2A 1.0 HTTP+JSON REST transport routes alongside the JSON-RPC route.
`context_builder`	`Callable[[Request[Any, Any, Any], ServerCallContext], ServerCallContext \| Awaitable[ServerCallContext]] \| None`	Optional `(request, context)` callback that authorizes the prepared SDK call context and returns it, directly or as an awaitable.
`include_in_schema`	`bool`	Include A2A routes in the OpenAPI schema.
`agent_card_max_age`	`int`	`Cache-Control` `max-age` for the agent card, in seconds. Must not be negative.
`stream_cleanup_timeout`	`float`	Seconds to wait for producer cleanup after a response ends.

LitestarA2A ¶

Bases: InitPluginProtocol

Register an A2A 1.0 request handler on Litestar-native routes.

This request-only adapter declines JSON-RPC notifications without executing them and returns HTTP 204. Explicit null request IDs still receive responses.

Attributes:

Name	Type	Description
`request_handler`	`RequestHandler`	Default: `DefaultRequestHandler(agent_executor=request_handler, task_store=InMemoryTaskStore(), agent_card=agent_card) if isinstance(request_handler, AgentExecutor) else request_handler`

on_app_init ¶

on_app_init(app_config)

Register the A2A JSON-RPC, agent card, and optional HTTP+JSON routes on the application.

Parameters:

Name	Type	Description	Default
`app_config`	`AppConfig`	The Litestar application configuration being initialized.	required

Returns:

Type	Description
`AppConfig`	The updated application configuration.

on_shutdown ¶

on_shutdown(app)

Close the A2A request handler if it defines an aclose coroutine or method.

Parameters:

Name	Type	Description	Default
`app`	`Litestar`	The shutting-down Litestar application instance.	required

Bidirectional Protocol Bridges (`litestar_mcp.agent.bridges`)¶

agent_to_mcp ¶

agent_to_mcp(agent_or_group, path='/mcp')

Expose tools, prompts, and skills of an Agent or AgentGroup as a LitestarMCP plugin.

mcp_to_tools ¶

mcp_to_tools(client, *, tools=None)

Convert remote tools discovered from an MCP client or LitestarMCP plugin into callable Tool instances.

discover_mcp_tools ¶

discover_mcp_tools(client_or_url, *, url='/mcp')

Asynchronously discover tools from a remote MCP HTTP endpoint using httpx2 and wrap them as Tools.

agent_to_a2a ¶

agent_to_a2a(
    agent_or_group,
    path="/a2a",
    *,
    base_url,
    runtime=None,
    task_store=None,
    owner_resolver=None,
    security_schemes=None,
    security_requirements=None,
    route_opt=None,
    agent_card_route_opt=None,
    tenant_header="X-A2A-Tenant",
    enable_http_json=True,
)

Mount an Agent or AgentGroup as an A2A 1.0 Litestar plugin.

Parameters:

Name	Type	Description	Default
`agent_or_group`	`Agent \| AgentGroup`	Single agent or multi-agent group to expose over A2A.	required
`path`	`str`	Route path or absolute URL for the A2A JSON-RPC endpoint.	`'/a2a'`
`base_url`	`str`	Required base URL used to construct the advertised AgentInterface when path is a relative route path.	required
`runtime`	`AgentRuntime \| None`	Optional AgentRuntime instance used to execute incoming requests. Defaults to AgentRuntime(target=agent_or_group).	`None`
`task_store`	`Any \| None`	Optional A2A TaskStore override. Defaults to an owner-scoped InMemoryTaskStore.	`None`
`owner_resolver`	`Callable[[ServerCallContext], str] \| None`	Optional callable resolving owner string from ServerCallContext.	`None`
`security_schemes`	`dict[str, Any] \| None`	Optional OpenAPI/A2A security schemes for the AgentCard.	`None`
`security_requirements`	`list[Any] \| None`	Optional security requirements for the AgentCard.	`None`
`route_opt`	`dict[str, Any] \| None`	Optional route `opt` mapping applied to A2A transport routes.	`None`
`agent_card_route_opt`	`dict[str, Any] \| None`	Optional route `opt` mapping applied to the agent card route.	`None`
`tenant_header`	`str \| None`	Optional HTTP header name used as a fallback tenant source.	`'X-A2A-Tenant'`
`enable_http_json`	`bool`	Whether to mount and advertise the A2A 1.0 HTTP+JSON transport.	`True`

Returns:

Type	Description
`LitestarA2A`	Configured LitestarA2A plugin instance.

In-Process Stdio Transport (`litestar_mcp.mcp.stdio`)¶

MCPStdioContext ¶

Runtime identity context for standalone MCP stdio transports.

Parameters:

Name	Type	Description	Default
`client_id`	`str`	Client identifier attached to stdio MCP requests.	`'stdio'`
`owner_id`	`str \| None`	Optional owner identifier used to scope background tasks created over stdio.	`None`
`user`	`Any`	Pre-populated `request.user` principal injected into synthesized ASGI scopes.	`None`
`auth`	`Any`	Pre-populated `request.auth` context injected into synthesized ASGI scopes.	`None`
`session`	`Mapping[str, Any] \| None`	Optional session dictionary injected into synthesized ASGI scopes.	`None`
`state`	`Mapping[str, Any] \| None`	Optional initial request state mapping injected into synthesized ASGI scopes.	`None`

Attributes:

Name	Type	Description
`client_id`	`str`	Client identifier attached to stdio MCP requests.
`owner_id`	`str \| None`	Optional owner identifier used to scope background tasks created over stdio.
`user`	`Any`	Pre-populated `request.user` principal injected into synthesized ASGI scopes.
`auth`	`Any`	Pre-populated `request.auth` context injected into synthesized ASGI scopes.
`session`	`Mapping[str, Any] \| None`	Optional session dictionary injected into synthesized ASGI scopes.
`state`	`Mapping[str, Any] \| None`	Optional initial request state mapping injected into synthesized ASGI scopes.

ASGIStreamingTransport ¶

Bases: AsyncBaseTransport

Run an ASGI application in-process and stream its response body to httpx2.

httpx2.ASGITransport buffers every http.response.body chunk until the application returns, so Server-Sent Events never reach the client while the stream is open. This transport returns the response as soon as http.response.start arrives, hands each body chunk to the response stream as it is sent, and cancels the application task when the response is closed, or when the caller is cancelled before http.response.start arrives, so an aborted request observes http.disconnect and task cancellation. The httpx2 read timeout bounds the wait for http.response.start and for each body chunk (httpx2.ReadTimeout).

Parameters:

Name	Type	Description	Default
`app`	`ASGIApp`	The ASGI application to call for every request.	required
`client`	`tuple[str, int]`	The `scope["client"]` tuple presented to the application.	`_DEFAULT_CLIENT`
`root_path`	`str`	The `scope["root_path"]` presented to the application.	`''`
`shutdown_timeout`	`float`	Positive finite seconds allowed for application cleanup. Expiry is logged; cancellation-resistant application code may outlive the response.	`5.0`

handle_async_request ¶

handle_async_request(request)

Dispatch request to the application and return a streaming response.

run_stdio_async ¶

run_stdio_async(
    app,
    *,
    stdio_context=None,
    headers=None,
    token_provider=None,
    header_name=DEFAULT_AUTH_HEADER_NAME,
    token_prefix=BEARER_TOKEN_PREFIX,
    sse_read_timeout=300.0,
    stdin=None,
    stdout=None,
    stderr=None,
    max_message_size=DEFAULT_MAX_STDIN_MESSAGE_SIZE,
    shutdown_timeout=5.0,
)

Serve a Litestar MCP endpoint to a local stdio client without a socket.

shutdown_timeout bounds in-process request cleanup and application shutdown after native lifespan entry succeeds. Bridge errors and body cancellation are preserved if that shutdown fails or times out.

Startup and its unwind follow Litestar's native semantics, including exception grouping and chaining. Application startup, lifespan, and shutdown hooks must bound and shield their own cleanup where needed.

run_stdio ¶

run_stdio(
    app,
    *,
    stdio_context=None,
    headers=None,
    token_provider=None,
    header_name=DEFAULT_AUTH_HEADER_NAME,
    token_prefix=BEARER_TOKEN_PREFIX,
    sse_read_timeout=300.0,
    stdin=None,
    stdout=None,
    stderr=None,
    max_message_size=DEFAULT_MAX_STDIN_MESSAGE_SIZE,
    shutdown_timeout=5.0,
)

Synchronously run run_stdio_async; KeyboardInterrupt exits 0.

Stdio-to-Streamable HTTP Bridge (`litestar_mcp.mcp.bridge`)¶

run_stdio_streamable_http_bridge ¶

run_stdio_streamable_http_bridge(
    endpoint,
    *,
    transport=None,
    client_info=None,
    headers=None,
    token_provider=None,
    header_name=DEFAULT_AUTH_HEADER_NAME,
    token_prefix=BEARER_TOKEN_PREFIX,
    timeout=30.0,
    sse_read_timeout=300.0,
    stdin=None,
    stdout=None,
    stderr=None,
    max_message_size=DEFAULT_MAX_STDIN_MESSAGE_SIZE,
)

Bridge local stdio JSON-RPC to a Litestar Streamable HTTP MCP endpoint.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	Full MCP Streamable HTTP endpoint URL.	required
`transport`	`AsyncBaseTransport \| None`	Optional httpx2 transport. Pass `litestar_mcp.mcp.stdio.ASGIStreamingTransport` to serve an in-process ASGI app instead of a remote endpoint.	`None`
`client_info`	`Mapping[str, str] \| None`	Default `io.modelcontextprotocol/clientInfo` injected into requests that do not carry one.	`None`
`headers`	`Mapping[str, str] \| None`	Static HTTP headers sent to the endpoint.	`None`
`token_provider`	`TokenProvider \| None`	Optional callable that returns a fresh token per request.	`None`
`header_name`	`str`	Header used for token auth.	`DEFAULT_AUTH_HEADER_NAME`
`token_prefix`	`str`	Prefix prepended to token values.	`BEARER_TOKEN_PREFIX`
`timeout`	`float`	HTTP connect, write, and pool timeout in seconds.	`30.0`
`sse_read_timeout`	`float \| None`	Read timeout for streaming SSE responses. `None` or `0` disables quiet-period timeouts for server streams.	`300.0`
`stdin`	`ByteReceiveStream \| None`	Optional byte receive stream for tests or embedding.	`None`
`stdout`	`ByteSendStream \| None`	Optional byte send stream for tests or embedding.	`None`
`stderr`	`Any \| None`	Optional diagnostic text stream. Defaults to `sys.stderr`.	`None`
`max_message_size`	`int`	Maximum bytes allowed for one newline-delimited JSON-RPC message, applied both to stdin lines and to server-sent events read from the endpoint. Set to `-1` to disable the limit.	`DEFAULT_MAX_STDIN_MESSAGE_SIZE`

Returns:

Type	Description
`int`	Process-style exit code. `0` means clean EOF/shutdown; non-zero means
`int`	a transport or pump error was surfaced to the local stdio client.

run_bridge ¶

run_bridge(
    endpoint,
    *,
    headers=None,
    token_provider=None,
    header_name=DEFAULT_AUTH_HEADER_NAME,
    token_prefix=BEARER_TOKEN_PREFIX,
    timeout=30.0,
    sse_read_timeout=300.0,
    stdout=None,
    stderr=None,
    max_message_size=DEFAULT_MAX_STDIN_MESSAGE_SIZE,
)

Synchronously run the stdio bridge for CLI integrations.

Streamable HTTP Client (`litestar_mcp.mcp.client`)¶

MCPStreamableHTTPClient ¶

Request/response MCP client over Streamable HTTP using httpx2.

Attributes:

Name	Type	Description
`tools`	`list[dict[str, Any]]`	Return cached list of discovered tools.

request ¶

request(method, params=None)

Send JSON-RPC request to MCP endpoint and parse response.

list_tools ¶

list_tools()

List all remote tools, following pagination cursors.

call_tool ¶

call_tool(name, arguments=None)

Call a remote tool by name with arguments and extract structured or text content.

MCPClientError ¶

Bases: LitestarMCPError

A JSON-RPC error or isError tool result returned by an MCP server.

A2A 1.0, Protocol Bridges & Transports API¶

A2A 1.0 Plugin & Configuration (litestar_mcp.a2a)¶

A2AConfig ¶

LitestarA2A ¶

on_app_init ¶

on_shutdown ¶

Bidirectional Protocol Bridges (litestar_mcp.agent.bridges)¶

agent_to_mcp ¶

mcp_to_tools ¶

discover_mcp_tools ¶

agent_to_a2a ¶

In-Process Stdio Transport (litestar_mcp.mcp.stdio)¶

MCPStdioContext ¶

ASGIStreamingTransport ¶

handle_async_request ¶

run_stdio_async ¶

run_stdio ¶

Stdio-to-Streamable HTTP Bridge (litestar_mcp.mcp.bridge)¶

run_stdio_streamable_http_bridge ¶

run_bridge ¶

Streamable HTTP Client (litestar_mcp.mcp.client)¶

MCPStreamableHTTPClient ¶

request ¶

list_tools ¶

call_tool ¶

MCPClientError ¶

A2A 1.0 Plugin & Configuration (`litestar_mcp.a2a`)¶

Bidirectional Protocol Bridges (`litestar_mcp.agent.bridges`)¶

In-Process Stdio Transport (`litestar_mcp.mcp.stdio`)¶

Stdio-to-Streamable HTTP Bridge (`litestar_mcp.mcp.bridge`)¶

Streamable HTTP Client (`litestar_mcp.mcp.client`)¶