Protocol Bridges¶

litestar_mcp.agent.bridges provides four bidirectional adapters (agent_to_mcp, mcp_to_tools, discover_mcp_tools, and agent_to_a2a) that connect Litestar MCP route handlers, remote Streamable HTTP MCP servers, local Agent / AgentGroup / AgentRuntime instances, and official A2A 1.0 endpoints without duplicating tool schemas or handler logic.

Bridge Reference Matrix¶

Complete Example: Bridging MCP Routes, AgentRuntime, and A2A 1.0¶

The snippet below demonstrates all three bridge directions in a single self-contained Litestar application:

1. mcp_to_tools(catalog_mcp) converts the @get("/inventory/{sku:str}", mcp_tool="lookup_inventory") route handler into a Tool attached to inventory_agent.
2. agent_to_mcp(inventory_agent, path="/mcp") exposes both the bridged lookup_inventory tool and InventorySkillController.check_reorder_threshold over MCP at /mcp.
3. agent_to_a2a(inventory_agent, path="/a2a", base_url="https://agents.example.com", runtime=runtime) mounts the agent's AgentRuntime as an A2A 1.0 AgentExecutor serving /.well-known/agent-card.json, JSON-RPC 2.0 at /a2a, and HTTP+JSON REST routes.

"""Snippet: Bidirectional protocol bridges connecting Litestar MCP routes, AgentRuntime, and A2A 1.0."""

# start-example
from litestar import Litestar, get

from litestar_mcp import LitestarMCP, SkillController, tool
from litestar_mcp.agent import (
    Agent,
    AgentChatController,
    AgentRuntime,
    MockModelClient,
    ModelDelta,
    agent_to_a2a,
    agent_to_mcp,
    mcp_to_tools,
)


@get("/inventory/{sku:str}", mcp_tool="lookup_inventory")
async def lookup_inventory(sku: str) -> dict[str, str | int]:
    """Look up warehouse stock levels for a product SKU."""
    return {"sku": sku, "available": 42}


class InventorySkillController(SkillController):
    """Domain skill providing reorder threshold checks for inventory agents."""

    name = "inventory"
    description = "Verify warehouse stock and reorder thresholds"
    tags = ("inventory", "warehouse")
    min_stock = 10

    @tool
    def check_reorder_threshold(self, sku: str, available: int) -> dict[str, str | bool]:
        """Determine whether a SKU has fallen below its reorder threshold."""
        return {"sku": sku, "reorder_required": available < self.min_stock}


catalog_mcp = LitestarMCP()
catalog_app = Litestar(route_handlers=[lookup_inventory], plugins=[catalog_mcp])
bridged_tools = mcp_to_tools(catalog_mcp)

model = MockModelClient(responses=[[ModelDelta(event_type="delta", text="SKU espresso-beans has 42 units in stock.")]])
inventory_agent = Agent(
    name="inventory_specialist",
    description="Answers warehouse inventory questions using bridged MCP tools.",
    instructions="Check inventory availability before confirming stock counts.",
    tools=bridged_tools,
    skills=[InventorySkillController],
    model=model,
)
runtime = AgentRuntime(inventory_agent, default_model=model)
mcp_bridge_plugin = agent_to_mcp(inventory_agent, path="/mcp")
a2a_bridge_plugin = agent_to_a2a(
    inventory_agent,
    path="/a2a",
    base_url="https://agents.example.com",
    runtime=runtime,
)
app = Litestar(
    route_handlers=[AgentChatController],
    plugins=[runtime, mcp_bridge_plugin, a2a_bridge_plugin],
)
# end-example


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

1. Exposing Agents Over MCP (agent_to_mcp)¶

agent_to_mcp(agent_or_group, path="/mcp") inspects an Agent or AgentGroup and builds a LitestarMCP plugin:

• Skill Controller Deduplication: Collects all SkillController classes or instances from the agent (or from group.coordinator and group.specialists.values()) and passes them to MCPConfig(base_path=path, skill_controllers=skills_list).
• Tool Registration: Registers each unique Tool in ag.tool_set via plugin.register_tool(t). Because agent_to_mcp reads ag.tool_set rather than group.tools_for(ag), internal transfer_to_agent handoff tools are never leaked to external MCP clients.

If you also want to expose a full multi-turn AgentRuntime execution as a single composite MCP tool callable via tools/call, wrap runtime.run_turn with @tool:

from litestar_mcp import tool
from litestar_mcp.agent import AgentRuntime, TurnRequest
from litestar_mcp.core.context import ToolContext


def runtime_as_mcp_tool(runtime: AgentRuntime, *, name: str = "ask_agent", description: str = "Run an agent turn."):
    """Wrap an AgentRuntime as a single callable Tool for MCP clients."""

    @tool(name=name, description=description)
    async def _invoke(prompt: str, ctx: ToolContext, session_id: str | None = None) -> dict[str, str]:
        response = await runtime.run_turn(
            TurnRequest(user_message=prompt, session_id=session_id, context=ctx)
        )
        return {
            "session_id": response.session_id,
            "turn_id": response.turn_id,
            "agent_name": response.agent_name,
            "output": response.output,
        }

    return _invoke

2. Consuming Local & Remote MCP Tools (mcp_to_tools & discover_mcp_tools)¶

Local LitestarMCP Route Handlers (mcp_to_tools)¶

When passed a LitestarMCP plugin instance after route discovery, mcp_to_tools(plugin) iterates over plugin.discovered_tools and wraps each handler's underlying callable in a Tool(handler_fn, name=tool_name). This lets an in-process Agent invoke your existing @get / @post MCP-marked route handlers with full ToolContext and DI resolution.

Remote Streamable HTTP Servers (discover_mcp_tools)¶

To equip an Agent with tools hosted on an external MCP server (or an in-process ASGI app tested over httpx2.ASGITransport), use discover_mcp_tools with MCPStreamableHTTPClient (litestar_mcp.MCPStreamableHTTPClient):

import httpx2
from litestar_mcp import MCPStreamableHTTPClient
from litestar_mcp.agent import Agent, discover_mcp_tools


async def build_remote_equipped_agent(http_client: httpx2.AsyncClient) -> Agent:
    """Discover tools from a remote Streamable HTTP MCP endpoint and bind them to an Agent."""
    mcp_client = MCPStreamableHTTPClient(
        endpoint="https://mcp.internal.example.com/mcp",
        client=http_client,
    )
    remote_tools = await discover_mcp_tools(mcp_client)
    return Agent(
        name="ops_assistant",
        description="Executes operational queries against remote MCP services.",
        instructions="Use discovered remote tools to inspect service status.",
        tools=remote_tools,
    )

discover_mcp_tools invokes await client.list_tools() and delegates to mcp_to_tools(client, tools=raw_tools), which constructs a Tool.from_schema(name=..., description=..., parameters=..., call=...) for each remote tool. When the LLM calls one of those tools during an AgentRuntime turn, the synthesized async call coroutine forwards the validated keyword arguments to await client.call_tool(target_name, kwargs).

3. Adapting AgentRuntime to Official A2A 1.0 (agent_to_a2a)¶

agent_to_a2a (litestar_mcp.agent.agent_to_a2a, enabled via litestar-mcp[a2a]) bridges an Agent or AgentGroup and its AgentRuntime into an official a2a-sdk 1.0 LitestarA2A plugin:

When an A2A client sends a message to the bridged endpoint:

1. _BridgeAgentExecutor.execute enqueues an initial Task(status=TaskStatus(state=TaskState.TASK_STATE_WORKING)) event if context.current_task is None.
2. It resolves the caller's owner key via owner_resolver(context.call_context), constructs a ToolContext(user={"id": owner_key}, session_id=context.context_id, turn_id=context.task_id), and invokes await runtime.run_turn(TurnRequest(...)).
3. Because session_id is bound to context.context_id, multi-turn A2A conversations sharing a context_id automatically resume their owner-isolated transcript in runtime.sessions.
4. The executor emits a TaskArtifactUpdateEvent(artifact=Artifact(artifact_id="response", parts=[Part(text=turn_res.content)]), last_chunk=True) followed by TaskStatusUpdateEvent(status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED)).