Skip to content

symfonic.tools.mcp.connection

connection

JSONRPCMCPConnection -- the MCP wire protocol over HTTP.

Split from :mod:symfonic.tools.mcp.provider so each module holds one class. The transport is sound and was not implicated in T4.2.2's findings; the provider on top of it was, which is exactly why they are worth reading apart.

Optional dependency: httpx>=0.27, via pip install symfonic-core[mcp].

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 
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 src/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 src/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 src/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