Skip to content

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 symfonic/tools/mcp/connection.py 
46
47
48
49
50
51
52
53
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 symfonic/tools/mcp/connection.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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 symfonic/tools/mcp/connection.py 
114
115
116
117
118
119
120
121
122
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 symfonic/tools/mcp/connection.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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 symfonic/tools/mcp/protocol.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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 symfonic/tools/mcp/protocol.py 
56
57
58
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 symfonic/tools/mcp/protocol.py 
32
33
34
35
36
37
38
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 symfonic/tools/mcp/provider.py 
56
57
58
59
60
61
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 symfonic/tools/mcp/provider.py 
63
64
65
66
67
68
69
70
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 symfonic/tools/mcp/provider.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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 symfonic/tools/mcp/provider.py 
199
200
201
202
203
204
205
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 symfonic/tools/mcp/provider.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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 symfonic/tools/mcp/provider.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
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.