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/provider.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/provider.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/provider.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.

Returns:

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

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

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

    Returns:
        List of MCPToolDefinition objects for each tool the server exposes.
    """
    response = await self._send_request("tools/list", {})
    tools: list[MCPToolDefinition] = []
    for tool_data in response.get("tools", []):
        tools.append(
            MCPToolDefinition(
                name=tool_data["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._discovered_tools: dict[
        str, tuple[MCPServerConnection, MCPToolDefinition]
    ] = {}

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.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool to invoke.	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.

    Args:
        tool_name: Name of the tool to invoke.
        arguments: Arguments to pass to the tool.

    Returns:
        MCPToolResult. If ``tool_name`` was not discovered, returns an
        error result without raising.
    """
    if tool_name not in self._discovered_tools:
        return MCPToolResult(
            tool_name=tool_name,
            content=f"Unknown MCP tool: {tool_name}",
            is_error=True,
        )
    conn, _defn = self._discovered_tools[tool_name]
    return await conn.call_tool(tool_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.

Returns:

Type	Description
list[MCPToolDefinition]	Flat list of MCPToolDefinition from all reachable servers.

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.

    Returns:
        Flat list of MCPToolDefinition from all reachable servers.
    """
    all_tools: list[MCPToolDefinition] = []
    for server_name, conn in self._servers.items():
        try:
            tools = await conn.list_tools()
            for tool in tools:
                self._discovered_tools[tool.name] = (conn, tool)
                all_tools.append(tool)
            logger.info(
                "Discovered %d tools from MCP server '%s'",
                len(tools),
                server_name,
            )
        except Exception:
            logger.warning(
                "Failed to discover tools from server '%s'",
                server_name,
                exc_info=True,
            )
    return all_tools

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 _tool_name, (_conn, defn) in self._discovered_tools.items():

        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.