`symfonic.tools.mcp`¶

mcp ¶

symfonic.tools.mcp -- MCP (Model Context Protocol) server adapter.

Provides: - MCPToolDefinition / MCPToolResult -- frozen dataclass models - MCPServerConnection -- structural Protocol (runtime_checkable) - JSONRPCMCPConnection -- HTTP JSON-RPC 2.0 transport (requires httpx) - MCPToolProvider -- discovery + routing bridge to LangChain

JSONRPCMCPConnection ¶

JSONRPCMCPConnection(server_url: str)

MCP server connection via JSON-RPC 2.0 over HTTP.

Implements the MCP wire protocol: - tools/list — discover available tools - tools/call — invoke a tool by name

Requires httpx::

pip install symfonic-core[mcp]

Parameters:

Name	Type	Description	Default
`server_url`	`str`	Full HTTP(S) URL to the MCP server endpoint.	required

Example::

conn = JSONRPCMCPConnection("http://localhost:3000/mcp")
tools = await conn.list_tools()
result = await conn.call_tool("search", {"query": "force majeure"})

Source code in src/symfonic/tools/mcp/connection.py

def __init__(self, server_url: str) -> None:
    self._url = server_url
    self._request_id = 0
    self._initialized = False
    # Pooled HTTP client -- lazily created on first request so
    # construction does not require httpx to be installed and
    # subsequent requests reuse the connection pool / keep-alive.
    self._client: Any | None = None

call_tool `async` ¶

call_tool(tool_name: str, arguments: dict[str, Any]) -> MCPToolResult

Send tools/call request and return parsed result.

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	Name of the tool to invoke.	required
`arguments`	`dict[str, Any]`	Arguments matching the tool's input schema.	required

Returns:

Type	Description
`MCPToolResult`	MCPToolResult with concatenated text content from the response.

Source code in src/symfonic/tools/mcp/connection.py

async def call_tool(
    self,
    tool_name: str,
    arguments: dict[str, Any],
) -> MCPToolResult:
    """Send tools/call request and return parsed result.

    Args:
        tool_name: Name of the tool to invoke.
        arguments: Arguments matching the tool's input schema.

    Returns:
        MCPToolResult with concatenated text content from the response.
    """
    response = await self._send_request(
        "tools/call",
        {"name": tool_name, "arguments": arguments},
    )
    content_parts = response.get("content", [])
    text_parts = [p.get("text", "") for p in content_parts if p.get("type") == "text"]
    is_error: bool = response.get("isError", False)
    return MCPToolResult(
        tool_name=tool_name,
        content="\n".join(text_parts),
        is_error=is_error,
    )

close `async` ¶

close() -> None

Release connection state and close the pooled HTTP client.

Source code in src/symfonic/tools/mcp/connection.py

async def close(self) -> None:
    """Release connection state and close the pooled HTTP client."""
    self._initialized = False
    if self._client is not None:
        try:
            await self._client.aclose()
        except Exception:  # pragma: no cover -- defensive
            logger.debug("Error closing pooled httpx client", exc_info=True)
        self._client = None

list_tools `async` ¶

list_tools() -> list[MCPToolDefinition]

Send tools/list request and parse response.

A row without a usable name is skipped with a warning rather than raising. tool_data["name"] used to be an unguarded index, and since MCPToolProvider.discover_tools catches per server, a single malformed entry discarded every other tool that server advertised.

Returns:

Type	Description
`list[MCPToolDefinition]`	List of MCPToolDefinition objects for each well-formed tool the
`list[MCPToolDefinition]`	server exposes.

Source code in src/symfonic/tools/mcp/connection.py

async def list_tools(self) -> list[MCPToolDefinition]:
    """Send tools/list request and parse response.

    A row without a usable ``name`` is skipped with a warning rather than
    raising. ``tool_data["name"]`` used to be an unguarded index, and since
    ``MCPToolProvider.discover_tools`` catches per *server*, a single
    malformed entry discarded every other tool that server advertised.

    Returns:
        List of MCPToolDefinition objects for each well-formed tool the
        server exposes.
    """
    response = await self._send_request("tools/list", {})
    tools: list[MCPToolDefinition] = []
    for tool_data in response.get("tools", []):
        name = tool_data.get("name") if isinstance(tool_data, dict) else None
        if not isinstance(name, str) or not name:
            logger.warning(
                "Skipping a malformed tool row from %s: no usable 'name'",
                self._url,
            )
            continue
        tools.append(
            MCPToolDefinition(
                name=name,
                description=tool_data.get("description", ""),
                input_schema=tool_data.get("inputSchema", {}),
                server_url=self._url,
            )
        )
    return tools

MCPServerConnection ¶

Bases: Protocol

Abstract interface for connecting to MCP servers.

Any object that implements list_tools, call_tool, and close with the correct signatures satisfies this protocol via structural subtyping — no inheritance required.

Example::

class MyConnection:
    async def list_tools(self) -> list[MCPToolDefinition]: ...
    async def call_tool(self, tool_name, arguments) -> MCPToolResult: ...
    async def close(self) -> None: ...

assert isinstance(MyConnection(), MCPServerConnection)

call_tool `async` ¶

call_tool(tool_name: str, arguments: dict[str, Any]) -> MCPToolResult

Execute a tool on the server.

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	Name of the tool to invoke.	required
`arguments`	`dict[str, Any]`	Key-value arguments matching the tool's input schema.	required

Returns:

Type	Description
`MCPToolResult`	MCPToolResult with the server's response content.

Source code in src/symfonic/tools/mcp/protocol.py

async def call_tool(
    self,
    tool_name: str,
    arguments: dict[str, Any],
) -> MCPToolResult:
    """Execute a tool on the server.

    Args:
        tool_name: Name of the tool to invoke.
        arguments: Key-value arguments matching the tool's input schema.

    Returns:
        MCPToolResult with the server's response content.
    """
    ...

close `async` ¶

close() -> None

Close the connection and release any held resources.

Source code in src/symfonic/tools/mcp/protocol.py

async def close(self) -> None:
    """Close the connection and release any held resources."""
    ...

list_tools `async` ¶

list_tools() -> list[MCPToolDefinition]

Discover available tools from the server.

Returns:

Type	Description
`list[MCPToolDefinition]`	List of tool definitions reported by the MCP server.

Source code in src/symfonic/tools/mcp/protocol.py

async def list_tools(self) -> list[MCPToolDefinition]:
    """Discover available tools from the server.

    Returns:
        List of tool definitions reported by the MCP server.
    """
    ...

MCPToolDefinition `dataclass` ¶

MCPToolDefinition(name: str, description: str, input_schema: dict[str, Any] = dict(), server_url: str = '')

A tool discovered from an MCP server.

Attributes:

Name	Type	Description
`name`	`str`	Unique tool identifier as reported by the server.
`description`	`str`	Human-readable description of what the tool does.
`input_schema`	`dict[str, Any]`	JSON Schema object describing accepted arguments.
`server_url`	`str`	URL of the MCP server that owns this tool.

MCPToolProvider ¶

MCPToolProvider()

Bridge that connects to MCP servers and provides LangChain-compatible tools.

Registers one or more MCPServerConnection instances, discovers their tools, and routes tool calls to the correct server.

Usage::

provider = MCPToolProvider()
provider.add_server("legal", JSONRPCMCPConnection("http://localhost:3000"))
definitions = await provider.discover_tools()
lc_tools = provider.to_langchain_tools()

# Later, invoke a tool directly:
result = await provider.call_tool("search", {"query": "breach of contract"})

# Cleanup:
await provider.close_all()

Source code in src/symfonic/tools/mcp/provider.py

def __init__(self) -> None:
    self._servers: dict[str, MCPServerConnection] = {}
    self._catalogues: dict[
        str, tuple[MCPServerConnection, list[MCPToolDefinition]]
    ] = {}
    self._discovered_tools: dict[str, ToolRoute] = {}

add_server ¶

add_server(name: str, connection: MCPServerConnection) -> None

Register an MCP server connection under a logical name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logical label for this server (used in log messages).	required
`connection`	`MCPServerConnection`	Any object satisfying MCPServerConnection protocol.	required

Source code in src/symfonic/tools/mcp/provider.py

def add_server(self, name: str, connection: MCPServerConnection) -> None:
    """Register an MCP server connection under a logical name.

    Args:
        name: Logical label for this server (used in log messages).
        connection: Any object satisfying MCPServerConnection protocol.
    """
    self._servers[name] = connection

call_tool `async` ¶

call_tool(tool_name: str, arguments: dict[str, Any]) -> MCPToolResult

Execute a previously discovered tool by name.

The call goes to the connection the tool was discovered on, and is placed under the name that server advertised. When a name collided the exposed name carries a "{server}." prefix that no server ever sent, so passing the exposed name back over the wire would be a call for a tool that does not exist.

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	Name of the tool to invoke, as exposed by :meth:`discover_tools`.	required
`arguments`	`dict[str, Any]`	Arguments to pass to the tool.	required

Returns:

Type	Description
`MCPToolResult`	MCPToolResult. If `tool_name` was not discovered, returns an
`MCPToolResult`	error result without raising.

Source code in src/symfonic/tools/mcp/provider.py

async def call_tool(
    self,
    tool_name: str,
    arguments: dict[str, Any],
) -> MCPToolResult:
    """Execute a previously discovered tool by name.

    The call goes to the connection the tool was discovered on, and is
    placed under the name that *server* advertised. When a name collided
    the exposed name carries a ``"{server}."`` prefix that no server ever
    sent, so passing the exposed name back over the wire would be a call
    for a tool that does not exist.

    Args:
        tool_name: Name of the tool to invoke, as exposed by
            :meth:`discover_tools`.
        arguments: Arguments to pass to the tool.

    Returns:
        MCPToolResult. If ``tool_name`` was not discovered, returns an
        error result without raising.
    """
    route = self._discovered_tools.get(tool_name)
    if route is None:
        return MCPToolResult(
            tool_name=tool_name,
            content=f"Unknown MCP tool: {tool_name}",
            is_error=True,
        )
    return await route.connection.call_tool(route.wire_name, arguments)

close_all `async` ¶

close_all() -> None

Close all registered server connections.

Source code in src/symfonic/tools/mcp/provider.py

async def close_all(self) -> None:
    """Close all registered server connections."""
    for conn in self._servers.values():
        try:
            await conn.close()
        except Exception:
            logger.debug("Error closing MCP connection", exc_info=True)

discover_tools `async` ¶

discover_tools() -> list[MCPToolDefinition]

Discover all tools from all registered servers.

Failed servers are skipped with a warning — partial discovery is preferred over a hard failure when one server is unavailable.

Re-discovery replaces each answering server's catalogue rather than adding to it, so a tool the server has withdrawn stops being callable. A server that fails to answer keeps the catalogue it last advertised: replacement is per server precisely so a momentary outage is not a silent deregistration of a whole integration.

A row with no usable name is dropped with a warning and its siblings are kept. It used to raise out of list_tools and, because the except below is per server, take that server's whole catalogue with it.

Returns:

Type	Description
`list[MCPToolDefinition]`	Flat list of MCPToolDefinition from all reachable servers, each
`list[MCPToolDefinition]`	carrying the name it is exposed under (see
`list[MCPToolDefinition]`	mod:`symfonic.tools.mcp.routing`).

Source code in src/symfonic/tools/mcp/provider.py

async def discover_tools(self) -> list[MCPToolDefinition]:
    """Discover all tools from all registered servers.

    Failed servers are skipped with a warning — partial discovery is
    preferred over a hard failure when one server is unavailable.

    Re-discovery **replaces** each answering server's catalogue rather than
    adding to it, so a tool the server has withdrawn stops being callable.
    A server that fails to answer keeps the catalogue it last advertised:
    replacement is per server precisely so a momentary outage is not a
    silent deregistration of a whole integration.

    A row with no usable name is dropped with a warning and its siblings
    are kept. It used to raise out of ``list_tools`` and, because the
    ``except`` below is per server, take that server's whole catalogue
    with it.

    Returns:
        Flat list of MCPToolDefinition from all reachable servers, each
        carrying the name it is exposed under (see
        :mod:`symfonic.tools.mcp.routing`).
    """
    catalogues: dict[
        str, tuple[MCPServerConnection, list[MCPToolDefinition]]
    ] = {}
    for server_name, conn in self._servers.items():
        try:
            rows = await conn.list_tools()
        except Exception:
            logger.warning(
                "Failed to discover tools from server '%s'; keeping the "
                "catalogue it last advertised",
                server_name,
                exc_info=True,
            )
            previous = self._catalogues.get(server_name)
            if previous is not None:
                catalogues[server_name] = previous
            continue

        tools: list[MCPToolDefinition] = []
        for row in rows:
            if tool_name_of(row) is None:
                logger.warning(
                    "Skipping a malformed tool row from MCP server '%s': "
                    "no usable 'name'; its siblings are kept",
                    server_name,
                )
                continue
            tools.append(row)
        catalogues[server_name] = (conn, tools)
        logger.info(
            "Discovered %d tools from MCP server '%s'",
            len(tools),
            server_name,
        )

    self._catalogues = catalogues
    self._discovered_tools = resolve_routes(catalogues)
    return [route.definition for route in self._discovered_tools.values()]

to_langchain_tools ¶

to_langchain_tools() -> list[Any]

Convert discovered MCP tools to LangChain StructuredTools.

Each tool's name and description from the MCP server are mapped to the StructuredTool. Invocations are delegated back through MCPToolProvider.call_tool.

Returns:

Type	Description
`list[Any]`	List of LangChain StructuredTool instances, or an empty list
`list[Any]`	if langchain-core is not installed.

Source code in src/symfonic/tools/mcp/provider.py

def to_langchain_tools(self) -> list[Any]:
    """Convert discovered MCP tools to LangChain StructuredTools.

    Each tool's name and description from the MCP server are mapped
    to the StructuredTool. Invocations are delegated back through
    ``MCPToolProvider.call_tool``.

    Returns:
        List of LangChain StructuredTool instances, or an empty list
        if langchain-core is not installed.
    """
    try:
        from langchain_core.tools import StructuredTool
    except ImportError:
        logger.warning(
            "langchain-core not installed; cannot create StructuredTools"
        )
        return []

    tools: list[Any] = []
    for defn in (route.definition for route in self._discovered_tools.values()):

        async def _run(_name: str = defn.name, **kwargs: Any) -> str:
            result = await self.call_tool(_name, kwargs)
            return result.content

        tools.append(
            StructuredTool.from_function(
                coroutine=_run,
                name=defn.name,
                description=defn.description,
            )
        )
    return tools

MCPToolResult `dataclass` ¶

MCPToolResult(tool_name: str, content: str, is_error: bool = False, metadata: dict[str, Any] = dict())

Result from executing an MCP tool.

Attributes:

Name	Type	Description
`tool_name`	`str`	Name of the tool that was called.
`content`	`str`	Text content returned by the tool.
`is_error`	`bool`	True when the server reported an error response.
`metadata`	`dict[str, Any]`	Optional extra data from the server response.

symfonic.tools.mcp¶

mcp ¶

JSONRPCMCPConnection ¶

call_tool async ¶

close async ¶

list_tools async ¶

MCPServerConnection ¶

call_tool async ¶

close async ¶

list_tools async ¶

MCPToolDefinition dataclass ¶

MCPToolProvider ¶

add_server ¶

call_tool async ¶

close_all async ¶

discover_tools async ¶

to_langchain_tools ¶

MCPToolResult dataclass ¶

`symfonic.tools.mcp`¶

call_tool `async` ¶

close `async` ¶

list_tools `async` ¶

call_tool `async` ¶

close `async` ¶

list_tools `async` ¶

MCPToolDefinition `dataclass` ¶

call_tool `async` ¶

close_all `async` ¶

discover_tools `async` ¶

MCPToolResult `dataclass` ¶