Skip to content

MCP Integration

This section documents the Model Context Protocol (MCP) integration capabilities of the ToolRegistry library.

Architecture Overview

The MCP integration enables seamless communication with Model Context Protocol servers, allowing LLM applications to utilize tools provided by external MCP servers. The architecture follows a client-server communication model:

Core Components

  1. MCPToolWrapper: A wrapper class that provides both synchronous and asynchronous communication with MCP servers

  2. Handles tool execution via MCP protocol

  3. Supports various content types (text, image, embedded resources)

  4. Manages client transport and communication lifecycle

  5. MCPTool: A tool class that wraps MCP tool specifications

  6. Preserves original tool metadata and descriptions

  7. Converts MCP schemas to ToolRegistry format

  8. Supports namespace organization

  9. MCPIntegration: The main integration class that orchestrates server communication

  10. Manages client connections to MCP servers
  11. Discovers available tools from servers
  12. Handles transport abstraction for different connection types

Communication Architecture

  • Transport Layer: Supports multiple transport types (HTTP, WebSocket, file-based)
  • Protocol Layer: Implements MCP specification for tool discovery and execution
  • Content Processing: Handles multiple content types with post-processing

Key Features

  • Support for various transport types (URL, file paths, server instances)
  • Automatic tool discovery from MCP servers
  • Multi-format content support (text, images, embedded resources)
  • Namespace management for tool organization
  • Robust error handling with detailed logging
  • Both synchronous and asynchronous operation modes

Transport Support

The integration supports multiple transport mechanisms:

  • HTTP/HTTPS endpoints (streamable HTTP, SSE)
  • WebSocket connections
  • Local file paths (Python scripts, JavaScript files)
  • Dict-based stdio configurations

API Reference

MCPToolWrapper

Wrapper class providing both async and sync versions of MCP tool calls.

toolregistry.integrations.mcp.integration.MCPToolWrapper 

MCPToolWrapper(connection: MCPConnectionManager, name: str, params: list[str] | None)

Bases: BaseToolWrapper

Wrapper class providing both async and sync versions of MCP tool calls.

Attributes:

Name Type Description
name str

Name of the tool/operation.
params Optional[List[str]]

List of parameter names.

Initialize MCP tool wrapper.

Parameters:

Name Type Description Default
connection MCPConnectionManager

Shared connection manager for the MCP server.

 required
name str

Name of the tool/operation.

 required
params Optional[List[str]]

List of parameter names.

 required

transport property 

transport: str | dict | Path

Transport source, for backward compatibility.

call_async async 

call_async(*args: Any, **kwargs: Any) -> Any

Async implementation of MCP tool call.

Parameters:

Name Type Description Default
args Any

Positional arguments to pass to the tool.

 ()
kwargs Any

Keyword arguments to pass to the tool.

 {}

Returns:

Name Type Description
Any Any

Result from tool execution.

Raises:

Type Description
ValueError

If name not set.
Exception

If tool execution fails.

call_sync 

call_sync(*args: Any, **kwargs: Any) -> Any

Synchronous implementation of MCP tool call.

Delegates to the connection manager's call_tool_sync which runs the coroutine on a persistent background event loop, keeping the MCP transport alive across calls.

Parameters:

Name Type Description Default
args Any

Positional arguments to pass to the tool.

 ()
kwargs Any

Keyword arguments to pass to the tool.

 {}

Returns:

Name Type Description
Any Any

Result from tool execution.

Raises:

Type Description
ValueError

If name not set.
Exception

If tool execution fails.

MCPTool

Wrapper class for MCP tools that preserves original function metadata.

toolregistry.integrations.mcp.integration.MCPTool

Bases: Tool

Wrapper class for MCP tools that preserves original function metadata.

Attributes:

Name Type Description
name str

Name of the tool.
description str

Description of the tool.
parameters Dict[str, Any]

Parameter schema definition.
callable Callable[..., Any]

The wrapped callable function.
is_async bool

Whether the tool is async, defaults to False.

from_tool_json classmethod 

from_tool_json(tool_spec: Tool, connection: MCPConnectionManager, namespace: str | None = None) -> MCPTool

Create an MCPTool instance from a JSON representation.

Parameters:

Name Type Description Default
tool_spec Tool

The JSON representation of the tool.

 required
connection MCPConnectionManager

Shared connection manager for the MCP server.

 required
namespace Optional[str]

An optional namespace to prefix the tool name. If provided, the tool name will be formatted as "{namespace}.{name}".

 None

Returns:

Name Type Description
MCPTool MCPTool

A new instance of MCPTool configured with the provided parameters.

MCPIntegration

Handles integration with MCP server for tool registration.

toolregistry.integrations.mcp.integration.MCPIntegration 

MCPIntegration(registry: ToolRegistry)

Handles integration with MCP server for tool registration.

Attributes:

Name Type Description
registry ToolRegistry

Tool registry instance.

close async 

close() -> None

Close all persistent connections (async).

close_sync 

close_sync() -> None

Close all persistent connections (sync).

Shuts down background loop threads without requiring an event loop in the calling thread.

register_mcp_tools 

register_mcp_tools(transport: str | dict[str, Any] | Path, namespace: bool | str = False, persistent: bool = True, headers: dict[str, str] | None = None) -> None

Register all tools from an MCP server (synchronous entry point).

Parameters:

Name Type Description Default
transport Union[str, Dict[str, Any], Path]

Can be: - URL string (http(s)://, ws(s)://) - Path to script file (.py, .js) - Dict with "command", "args", "env" keys for stdio transport

 required
namespace Union[bool, str]

Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the server info name. - If a string is provided, it is used as the namespace. Defaults to False.

 False
persistent bool

If True (default), keep the connection open across tool calls. If False, create a new connection per call.

 True
headers Optional[Dict[str, str]]

HTTP headers for SSE or streamable-http transports (e.g. authentication).

 None

register_mcp_tools_async async 

register_mcp_tools_async(transport: str | dict[str, Any] | Path, namespace: bool | str = False, persistent: bool = True, headers: dict[str, str] | None = None) -> None

Async implementation to register all tools from an MCP server.

Parameters:

Name Type Description Default
transport Union[str, Dict[str, Any], Path]

Can be: - URL string (http(s)://, ws(s)://) - Path to script file (.py, .js) - Dict with "command", "args", "env" keys for stdio transport

 required
namespace Union[bool, str]

Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the server info name. - If a string is provided, it is used as the namespace. Defaults to False.

 False
persistent bool

If True (default), keep the connection open across tool calls. If False, create a new connection per call.

 True
headers Optional[Dict[str, str]]

HTTP headers for SSE or streamable-http transports (e.g. authentication).

 None

Raises:

Type Description
RuntimeError

If connection to server fails.

Module Utilities

MCPClient

Minimal MCP client adapter over the official mcp SDK. Supports stdio, SSE, streamable-http, and websocket transports.

toolregistry.integrations.mcp.client.MCPClient 

MCPClient(source: str | dict | Path, headers: dict[str, str] | None = None)

Thin async context manager around mcp.ClientSession.

Usage

async with MCPClient("http://localhost:8000/mcp") as client: tools = await client.list_tools() result = await client.call_tool("tool_name", {"arg": "value"})

Initialize MCPClient.

Parameters:

Name Type Description Default
source str | dict | Path

Connection target. Can be: - HTTP(S) URL string for SSE or streamable-http transport - WS(S) URL string for websocket transport - Dict with "command", "args", "env" keys for stdio transport - Path or string path to a .py/.js script for stdio transport

 required
headers dict[str, str] | None

Optional HTTP headers to send with SSE or streamable-http requests (e.g. for authentication). Ignored for stdio and websocket transports.

 None

initialize_result property 

initialize_result

Access the full initialization result.

Returns:

Type Description

The InitializeResult from the session, or None.

is_connected property 

is_connected: bool

Whether the client has an active session.

Returns:

Type Description
bool

True if connected with an active session, False otherwise.

server_info property 

server_info

Access server information from the initialization result.

Returns:

Type Description

Server info from the session, or None if not available.

session property 

session: ClientSession | None

Access the underlying ClientSession.

Returns:

Type Description
ClientSession | None

The active ClientSession, or None if not connected.

call_tool async 

call_tool(name: str, arguments: dict[str, Any]) -> CallToolResult

Call a tool on the MCP server.

Parameters:

Name Type Description Default
name str

Name of the tool to call.

 required
arguments dict[str, Any]

Dictionary of arguments to pass to the tool.

 required

Returns:

Type Description
CallToolResult

CallToolResult from the MCP server.

list_tools async 

list_tools() -> list[ToolSpec]

List available tools from the MCP server.

Returns:

Type Description
list[Tool]

List of ToolSpec objects describing available tools.

MCP Module

The main MCP integration module.

toolregistry.integrations.mcp

MCPConnectionManager 

MCPConnectionManager(transport: str | dict | Path, headers: dict[str, str] | None = None, persistent: bool = True)

Persistent connection manager for an MCP server.

All tools registered from the same server share this connection. Supports lazy connect on first call and auto-reconnect on failure.

For sync callers, a background daemon thread with its own event loop is lazily created on the first call_tool_sync() call. This keeps the loop (and therefore the MCP transport) alive across calls. Async callers use the caller's own loop and are unaffected.

Parameters:

Name Type Description Default
transport str | dict | Path

MCP server source (URL, dict, or Path).

 required
headers dict[str, str] | None

Optional HTTP headers for SSE/streamable-http transports.

 None
persistent bool

If True (default), keep the connection open across calls. If False, create a new connection per call (original behavior).

 True

is_connected property 

is_connected: bool

Whether an active session exists.

transport property 

transport: str | dict | Path

The transport source for the MCP server.

__del__ 

__del__() -> None

Best-effort cleanup of the background thread.

__getstate__ 

__getstate__() -> dict

Drop live connection state before pickling.

The config (transport, headers, persistent flag) is preserved. The live _client (holding OS sockets), _lock (event-loop-bound), and sync loop thread are dropped. The worker process will reconnect lazily on first call_tool().

__setstate__ 

__setstate__(state: dict) -> None

Restore config; connection will be re-established lazily.

call_tool async 

call_tool(name: str, arguments: dict[str, Any]) -> CallToolResult

Call a tool on the MCP server.

Uses persistent or per-request connection based on config.

Parameters:

Name Type Description Default
name str

Name of the tool to call.

 required
arguments dict[str, Any]

Arguments to pass to the tool.

 required

Returns:

Type Description
CallToolResult

CallToolResult from the MCP server.

call_tool_sync 

call_tool_sync(name: str, arguments: dict[str, Any]) -> CallToolResult

Call a tool synchronously using a persistent background loop.

The first call lazily starts a daemon thread whose event loop keeps MCP transport resources alive across calls.

Parameters:

Name Type Description Default
name str

Name of the tool to call.

 required
arguments dict[str, Any]

Arguments to pass to the tool.

 required

Returns:

Type Description
CallToolResult

CallToolResult from the MCP server.

close async 

close() -> None

Close the persistent connection and stop the sync loop thread.

close_sync 

close_sync() -> None

Close from sync context — tear down background thread and connection.

list_tools async 

list_tools()

List available tools using a temporary connection.

Returns:

Type Description

List of ToolSpec objects.

MCPIntegration 

MCPIntegration(registry: ToolRegistry)

Handles integration with MCP server for tool registration.

Attributes:

Name Type Description
registry ToolRegistry

Tool registry instance.

close async 

close() -> None

Close all persistent connections (async).

close_sync 

close_sync() -> None

Close all persistent connections (sync).

Shuts down background loop threads without requiring an event loop in the calling thread.

register_mcp_tools 

register_mcp_tools(transport: str | dict[str, Any] | Path, namespace: bool | str = False, persistent: bool = True, headers: dict[str, str] | None = None) -> None

Register all tools from an MCP server (synchronous entry point).

Parameters:

Name Type Description Default
transport Union[str, Dict[str, Any], Path]

Can be: - URL string (http(s)://, ws(s)://) - Path to script file (.py, .js) - Dict with "command", "args", "env" keys for stdio transport

 required
namespace Union[bool, str]

Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the server info name. - If a string is provided, it is used as the namespace. Defaults to False.

 False
persistent bool

If True (default), keep the connection open across tool calls. If False, create a new connection per call.

 True
headers Optional[Dict[str, str]]

HTTP headers for SSE or streamable-http transports (e.g. authentication).

 None

register_mcp_tools_async async 

register_mcp_tools_async(transport: str | dict[str, Any] | Path, namespace: bool | str = False, persistent: bool = True, headers: dict[str, str] | None = None) -> None

Async implementation to register all tools from an MCP server.

Parameters:

Name Type Description Default
transport Union[str, Dict[str, Any], Path]

Can be: - URL string (http(s)://, ws(s)://) - Path to script file (.py, .js) - Dict with "command", "args", "env" keys for stdio transport

 required
namespace Union[bool, str]

Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the server info name. - If a string is provided, it is used as the namespace. Defaults to False.

 False
persistent bool

If True (default), keep the connection open across tool calls. If False, create a new connection per call.

 True
headers Optional[Dict[str, str]]

HTTP headers for SSE or streamable-http transports (e.g. authentication).

 None

Raises:

Type Description
RuntimeError

If connection to server fails.

MCPTool

Bases: Tool

Wrapper class for MCP tools that preserves original function metadata.

Attributes:

Name Type Description
name str

Name of the tool.
description str

Description of the tool.
parameters Dict[str, Any]

Parameter schema definition.
callable Callable[..., Any]

The wrapped callable function.
is_async bool

Whether the tool is async, defaults to False.

from_tool_json classmethod 

from_tool_json(tool_spec: Tool, connection: MCPConnectionManager, namespace: str | None = None) -> MCPTool

Create an MCPTool instance from a JSON representation.

Parameters:

Name Type Description Default
tool_spec Tool

The JSON representation of the tool.

 required
connection MCPConnectionManager

Shared connection manager for the MCP server.

 required
namespace Optional[str]

An optional namespace to prefix the tool name. If provided, the tool name will be formatted as "{namespace}.{name}".

 None

Returns:

Name Type Description
MCPTool MCPTool

A new instance of MCPTool configured with the provided parameters.

MCPToolWrapper 

MCPToolWrapper(connection: MCPConnectionManager, name: str, params: list[str] | None)

Bases: BaseToolWrapper

Wrapper class providing both async and sync versions of MCP tool calls.

Attributes:

Name Type Description
name str

Name of the tool/operation.
params Optional[List[str]]

List of parameter names.

Initialize MCP tool wrapper.

Parameters:

Name Type Description Default
connection MCPConnectionManager

Shared connection manager for the MCP server.

 required
name str

Name of the tool/operation.

 required
params Optional[List[str]]

List of parameter names.

 required

transport property 

transport: str | dict | Path

Transport source, for backward compatibility.

call_async async 

call_async(*args: Any, **kwargs: Any) -> Any

Async implementation of MCP tool call.

Parameters:

Name Type Description Default
args Any

Positional arguments to pass to the tool.

 ()
kwargs Any

Keyword arguments to pass to the tool.

 {}

Returns:

Name Type Description
Any Any

Result from tool execution.

Raises:

Type Description
ValueError

If name not set.
Exception

If tool execution fails.

call_sync 

call_sync(*args: Any, **kwargs: Any) -> Any

Synchronous implementation of MCP tool call.

Delegates to the connection manager's call_tool_sync which runs the coroutine on a persistent background event loop, keeping the MCP transport alive across calls.

Parameters:

Name Type Description Default
args Any

Positional arguments to pass to the tool.

 ()
kwargs Any

Keyword arguments to pass to the tool.

 {}

Returns:

Name Type Description
Any Any

Result from tool execution.

Raises:

Type Description
ValueError

If name not set.
Exception

If tool execution fails.