toolregistry.mcp_integration module

class toolregistry.mcp_integration.MCPToolWrapper(url: str, name: str, params: List[str] | None)[source]

Bases: object

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

url

URL of the MCP server.

Type:

str

name

Name of the tool/operation.

Type:

str

params

List of parameter names.

Type:

Optional[List[str]]

__init__(url: str, name: str, params: List[str] | None) → None[source]
call_sync(*args: Any, **kwargs: Any) → Any[source]

Synchronous implementation of MCP tool call.

Parameters:

  • args (Any) – Positional arguments to pass to the tool.

  • kwargs (Any) – Keyword arguments to pass to the tool.

Returns:

Result from tool execution.

Return type:

Any

Raises:

  • ValueError – If URL or name not set.

  • Exception – If tool execution fails.

async call_async(*args: Any, **kwargs: Any) → Any[source]

Async implementation of MCP tool call.

Parameters:

  • args (Any) – Positional arguments to pass to the tool.

  • kwargs (Any) – Keyword arguments to pass to the tool.

Returns:

Result from tool execution.

Return type:

Any

Raises:

  • ValueError – If URL or name not set.

  • Exception – If tool execution fails.

class toolregistry.mcp_integration.MCPTool(*, name: str, description: str, parameters: Dict[str, Any], callable: Callable[[...], Any], is_async: bool = False, parameters_model: Any | None = None)[source]

Bases: Tool

Wrapper class for MCP tools that preserves original function metadata.

name

Name of the tool.

Type:

str

description

Description of the tool.

Type:

str

parameters

Parameter schema definition.

Type:

Dict[str, Any]

callable

The wrapped callable function.

Type:

Callable[…, Any]

is_async

Whether the tool is async, defaults to False.

Type:

bool

classmethod from_tool_json(name: str, description: str, input_schema: Dict[str, Any], url: str, namespace: str | None = None) → MCPTool[source]

Create an MCPTool instance from a JSON representation.

Parameters:

  • name (str) – The name of the tool.

  • description (str) – The description of the tool.

  • input_schema (Dict[str, Any]) – The input schema definition for the tool.

  • url (str) – The URL endpoint for the tool.

  • namespace (Optional[str]) – An optional namespace to prefix the tool name. If provided, the tool name will be formatted as “{namespace}.{name}”.

Returns:

A new instance of MCPTool configured with the provided parameters.

Return type:

MCPTool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class toolregistry.mcp_integration.MCPIntegration(registry: ToolRegistry)[source]

Bases: object

Handles integration with MCP server for tool registration.

registry

Tool registry instance.

Type:

ToolRegistry

__init__(registry: ToolRegistry)[source]
async register_mcp_tools_async(server_url: str, with_namespace: bool | str = False) → None[source]

Async implementation to register all tools from an MCP server.

Parameters:

  • server_url (str) – URL of the MCP server (e.g. “http://localhost:8000/mcp/sse”).

  • with_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 OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Raises:

RuntimeError – If connection to server fails.

register_mcp_tools(server_url: str, with_namespace: bool | str = False) → None[source]

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

Parameters:

  • server_url (str) – URL of the MCP server.

  • with_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 OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.