mcp_multi_server package

Submodules

mcp_multi_server.client module

MCP client for managing connections to multiple MCP servers.

class mcp_multi_server.client.MultiServerClient(config_path: str | Path = 'mcp_servers.json')

Bases: object

Manages multiple MCP server connections for a MCP host.

This class handles:

• Connecting to multiple MCP servers

• Discovering and aggregating server capabilities (tools, resources, templates, prompts)

• Routing tool, prompt and resource calls to the correct server

• Managing session lifecycles with AsyncExitStack

The client can be used as an async context manager for automatic cleanup:

Examples:

Basic usage with context manager:
>>> async with MultiServerClient.from_config("mcp_servers.json") as client:
...     tools = client.list_tools()
...     result = await client.call_tool("my_tool", {"arg": "value"})

Manual connection management:
>>> async with AsyncExitStack() as stack:
...     client = MultiServerClient("mcp_servers.json")
...     await client.connect_all(stack)
...     tools = client.list_tools()

Programmatic configuration:
>>> config = MCPServersConfig(mcpServers={
...     "my_server": ServerConfig(command="python", args=["-m", "my_server"])
... })
>>> async with MultiServerClient.from_dict(config.model_dump()) as client:
...     tools = client.list_tools()

Initialize the multi-server client.

Parameters:

config_path – Path to the JSON configuration file containing server definitions. Defaults to “mcp_servers.json” in the current directory.

Note

This constructor only sets up the configuration path. The actual connection to servers happens when connect_all() is called or when using the client as a context manager.

__init__(config_path: str | Path = 'mcp_servers.json') → None

Initialize the multi-server client.

Parameters:

config_path – Path to the JSON configuration file containing server definitions. Defaults to “mcp_servers.json” in the current directory.

Note

This constructor only sets up the configuration path. The actual connection to servers happens when connect_all() is called or when using the client as a context manager.

classmethod from_config(config_path: str | Path) → MultiServerClient

Create a client from a configuration file path.

This is a convenience class method that’s equivalent to the regular constructor.

Parameters:

config_path – Path to the JSON configuration file.

Returns:

A new MultiServerClient instance.

Examples

>>> client = MultiServerClient.from_config("my_servers.json")

classmethod from_dict(config_dict: Dict[str, Any]) → MultiServerClient

Create a client from a configuration dictionary.

This method allows programmatic configuration without needing a JSON file.

Parameters:

config_dict – Dictionary containing server configurations in the same format as the JSON file (with “mcpServers” key).

Returns:

A new MultiServerClient instance with the provided configuration.

Raises:

pydantic.ValidationError – If the config dictionary doesn’t match the schema.

Examples

>>> config = {
...     "mcpServers": {
...         "tool_server": {
...             "command": "python",
...             "args": ["-m", "my_package.tool_server"]
...         }
...     }
... }
>>> client = MultiServerClient.from_dict(config)

async connect_all(stack: AsyncExitStack) → None

Connect to all configured MCP servers and discover their capabilities.

Parameters:

stack – AsyncExitStack for managing async context managers.

Raises:

• FileNotFoundError – If config file doesn’t exist.

• json.JSONDecodeError – If config file is invalid JSON.

• pydantic.ValidationError – If config data doesn’t match schema.

Note

Individual server connection failures are caught and logged as warnings. The method will continue connecting to remaining servers if one fails.

async set_logging_level(level: Literal['debug', 'info', 'notice', 'warning', 'error', 'critical', 'alert', 'emergency']) → EmptyResult

Set the logging level for the multi-server client and the MCP connected servers.

Parameters:

level – Logging level as a string in lower case (e.g., “debug”, “info”, “notice”, “warning”, “error”, “critical”, “alert”, “emergency”) as defined in MCP LoggingLevel.

Note

The following mappings of MCP to Python logging leves are applied: - “notice” -> “WARNING” - “alert” and “emergency” -> “CRITICAL”

Examples

>>> await MultiServerClient.set_logging_level("debug")

list_tools(cursor: str | None = None, *, params: PaginatedRequestParams | None = None) → ListToolsResult

Get combined list of all tools from all servers.

This method mimics the MCP ClientSession.list_tools() signature but aggregates tools from all connected servers. Server attribution is included in each tool’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

Returns:

ListToolsResult containing all tools from all servers with the server name in the serverName meta field. The nextCursor field is always None (pagination not supported).

Raises:

ValueError – If cursor or params is not None (pagination not supported).

Examples

>>> result = client.list_tools()
>>> for tool in result.tools:
...     server = tool.meta.get("serverName") if tool.meta else None
...     print(f"{tool.name} from {server}")

list_prompts(cursor: str | None = None, *, params: PaginatedRequestParams | None = None) → ListPromptsResult

Get combined list of all prompts from all servers.

This method mimics the MCP ClientSession.list_prompts() signature but aggregates prompts from all connected servers. Server attribution is included in each prompt’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

Returns:

ListPromptsResult containing all prompts from all servers with the server name in the serverName meta fieldthe. The nextCursor field is always None (pagination not supported).

Raises:

ValueError – If cursor or params is not None (pagination not supported).

Examples

>>> result = client.list_prompts()
>>> for prompt in result.prompts:
...     server = prompt.meta.get("serverName") if prompt.meta else None
...     print(f"{prompt.name} from {server}")

list_resources(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True) → ListResourcesResult

Get combined list of all resources from all servers.

This method mimics the MCP ClientSession.list_resources() signature but aggregates resources from all connected servers. Resources are returned with namespaced URIs (server:uri format) for auto-routing, and server attribution is included in each resource’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

• use_namespace – Whether to namespace the URIs with the server name.

Returns:

• Namespaced URIs in format “server_name:original_uri” for auto-routing

• the server name in the serverName meta field for explicit server identification

The nextCursor field is always None (pagination not supported).

Return type:

ListResourcesResult containing all resources from all servers with

Raises:

ValueError – If cursor is not None (pagination not supported).

Examples

>>> result = client.list_resources()
>>> for resource in result.resources:
...     server = resource.meta.get("serverName") if resource.meta else None
...     # URI is already namespaced: "filesystem:file:///path"
...     content = await client.read_resource(resource.uri)

list_resource_templates(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True) → ListResourceTemplatesResult

Get combined list of all resource templates from all servers.

This method mimics the MCP ClientSession.list_resource_templates() signature but aggregates resource templates from all connected servers. Templates are returned with namespaced URI templates (server:template format) for auto-routing, and server attribution is included in each template’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

• use_namespace – Whether to namespace the URI templates with the server name.

Returns:

• Namespaced URI templates in format “server_name:original_template”

• the server name in the serverName meta field for explicit server identification

The nextCursor field is always None (pagination not supported).

Return type:

ListResourceTemplatesResult containing all templates from all servers with

Raises:

ValueError – If cursor is not None (pagination not supported).

Examples

>>> result = client.list_resource_templates()
>>> for template in result.resourceTemplates:
...     server = template.meta.get("serverName") if template.meta else None
...     # URI template is already namespaced: "filesystem:file:///{path}"
...     # Needs to be filled in with actual path when used
...     uri = template.uriTemplate.replace("{path}", "example.txt")
...     content = await client.read_resource(uri)

async call_tool(name: str, arguments: Dict[str, Any], read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None, server_name: str | None = None) → CallToolResult

Route a tool call to the appropriate server.

Parameters:

• name – Name of the tool to call.

• arguments – Arguments to pass to the tool.

• read_timeout_seconds – Optional timeout for reading the tool result.

• progress_callback – Optional callback for progress notifications.

• meta – Optional metadata dictionary to pass additional information.

• server_name – Optional server name to explicitly specify which server to use. If not provided, the server will be automatically determined from the tool name.

Returns:

Result from the tool execution. If the tool name is not found or routing fails, returns a CallToolResult with isError=True containing an error message.

Raises:

• McpError – If the tool execution fails or times out (protocol-level errors).

• RuntimeError – If tool result validation fails (invalid structured content or schema).

Note

Routing errors (unknown tool, unknown server) are returned as error results (isError=True) rather than raising exceptions, following MCP protocol conventions. Protocol-level errors from the underlying session are propagated as exceptions.

async read_resource(uri: str | AnyUrl, server_name: str | None = None) → ReadResourceResult

Read a resource with optional auto-routing via namespaced URIs.

Parameters:

• uri – Resource URI. Can be namespaced as “server:uri” for auto-routing. URIs from list_resources() are already namespaced for convenience. Accepts both str and AnyUrl types for MCP library compatibility.

• server_name – Optional explicit server name. If provided, assumes that there is no any namespace in the provided URI.

Returns:

Resource content.

Raises:

McpError – If server name is not found, URI is not namespaced when server_name is not provided, or if the resource read fails or times out.

Examples:

Auto-routing with namespaced URI (from list_resources()):
>>> resources = client.list_resources().resources
>>> result = await client.read_resource(resources[0].uri)

Explicit server (no namespace should be present in URI):
>>> result = await client.read_resource("file:///path", server_name="filesystem")

Manual namespacing:
>>> result = await client.read_resource("filesystem:file:///path")

Note

Raises McpError for both routing errors and protocol-level errors to align with MCP SDK behavior.

async get_prompt(name: str, arguments: Dict[str, str] | None = None, server_name: str | None = None) → GetPromptResult

Get a prompt by automatically routing to the appropriate server.

Parameters:

• name – Name of the prompt to get.

• arguments – Optional arguments for the prompt.

• server_name – Optional server name to explicitly specify which server to use. If not provided, the server will be automatically determined from the prompt name.

Returns:

Prompt result.

Raises:

McpError – If prompt name is not found, server name is not found, or if the prompt retrieval fails or times out.

Note

Raises McpError for both routing errors and protocol-level errors to align with MCP SDK behavior.

mcp_multi_server.config module

Configuration models for MCP multi-server client.

class mcp_multi_server.config.ServerConfig(*, command: str, args: List[str])

Bases: BaseModel

Configuration for a single MCP server.

command

The executable command to start the server (e.g., “python”, “node”).

Type:

str

args

Command-line arguments to pass to the server executable.

Type:

List[str]

Examples

>>> config = ServerConfig(
...     command="python",
...     args=["-m", "mcp_servers.tool_server"]
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

command: str

args: List[str]

model_config: ClassVar[ConfigDict] = {}

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

class mcp_multi_server.config.MCPServersConfig(*, mcpServers: Dict[str, ServerConfig])

Bases: BaseModel

Configuration for all MCP servers.

mcpServers

Dictionary mapping server names to their configurations. Server names are used as identifiers throughout the client.

Type:

Dict[str, mcp_multi_server.config.ServerConfig]

Examples

>>> config = MCPServersConfig(mcpServers={
...     "tool_server": ServerConfig(
...         command="python",
...         args=["-m", "mcp_servers.tool_server"]
...     ),
...     "resource_server": ServerConfig(
...         command="python",
...         args=["-m", "mcp_servers.resource_server"]
...     )
... })

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

mcpServers: Dict[str, ServerConfig]

model_config: ClassVar[ConfigDict] = {}

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

mcp_multi_server.sync_client module

Synchronous wrapper for MCP MultiServerClient with background event loop.

This module provides SyncMultiServerClient, a context manager that wraps the async MultiServerClient from mcp_multi_server in a synchronous interface using a background thread with persistent event loop.

class mcp_multi_server.sync_client.SyncMultiServerClient(config_path: str | Path | None = None, config_dict: Dict[str, Any] | None = None)

Bases: object

Manages MCP multi server client in a background thread with persistent event loop.

This class provides a synchronous interface to the async MultiServerClient, making it easier to integrate with synchronous code while maintaining the efficiency of async operations.

Usage:

# Context manager (recommended) with SyncMultiServerClient(config_path) as client:

tools = client.list_tools() result = client.call_tool(“tool_name”, {“arg”: “value”})

# Manual lifecycle management client = SyncMultiServerClient(config_path) tools = client.list_tools() # … use client … client.shutdown()

Thread Safety:

All public methods are thread-safe, using asyncio.run_coroutine_threadsafe() to schedule operations on the background event loop.

Initialize SyncMultiServerClient.

Starts background thread and initializes MCP client during construction. Automatically registers cleanup handler to ensure proper shutdown on program exit.

Parameters:

• config_path – Path to MCP configuration file.

• config_dict – Configuration dictionary (alternative to config_path).

Raises:

• ValueError – If neither or both config_path and config_dict are provided.

• Exception – If MCP client initialization fails.

__init__(config_path: str | Path | None = None, config_dict: Dict[str, Any] | None = None)

Initialize SyncMultiServerClient.

Starts background thread and initializes MCP client during construction. Automatically registers cleanup handler to ensure proper shutdown on program exit.

Parameters:

• config_path – Path to MCP configuration file.

• config_dict – Configuration dictionary (alternative to config_path).

Raises:

• ValueError – If neither or both config_path and config_dict are provided.

• Exception – If MCP client initialization fails.

classmethod from_config(config_path: str | Path) → SyncMultiServerClient

Create a client from a configuration file path.

This is a convenience class method that’s equivalent to calling the constructor with a config_path argument.

Parameters:

config_path – Path to the JSON configuration file.

Returns:

A new SyncMultiServerClient instance.

Examples

>>> client = SyncMultiServerClient.from_config("mcp_config.json")

classmethod from_dict(config_dict: Dict[str, Any]) → SyncMultiServerClient

Create a client from a configuration dictionary.

This method allows programmatic configuration without needing a JSON file.

Parameters:

config_dict – Dictionary containing server configurations in the same format as the JSON file (with “mcpServers” key).

Returns:

A new SyncMultiServerClient instance with the provided configuration.

Examples

>>> config = {
...     "mcpServers": {
...         "tool_server": {
...             "command": "python",
...             "args": ["-m", "my_package.tool_server"]
...         }
...     }
... }
>>> client = SyncMultiServerClient.from_dict(config)

shutdown() → None

Shutdown background thread and cleanup MCP client.

Safe to call multiple times. Waits up to 10 seconds for graceful cleanup.

set_logging_level(level: Literal['debug', 'info', 'notice', 'warning', 'error', 'critical', 'alert', 'emergency']) → EmptyResult

Set the logging level for the multi-server client and the MCP connected servers.

Parameters:

level – Logging level as a string in lower case (e.g., “debug”, “info”, “notice”, “warning”, “error”, “critical”, “alert”, “emergency”) as defined in MCP LoggingLevel.

Note

The following mappings of MCP to Python logging leves are applied: - “notice” -> “WARNING” - “alert” and “emergency” -> “CRITICAL”

Examples

>>> SyncMultiServerClient.set_logging_level("debug")

property capabilities: Dict[str, Any]

Get combined capabilities from all connected MCP servers.

Returns:

Dictionary containing combined capabilities from all servers. Returns empty dict if client not initialized or error occurs.

list_tools(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, **kwargs: Any) → ListToolsResult

List available MCP tools in raw MCP format.

Returns:

List of MCP Tool objects (not converted to OpenAI format). Returns empty list if client not initialized or error occurs.

list_prompts(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, **kwargs: Any) → ListPromptsResult

Get combined list of all prompts from all servers.

Returns:

ListPromptsResult containing all prompts from all servers with server attribution in the serverName meta field. Returns empty list if client not initialized or error occurs.

list_resources(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True, **kwargs: Any) → ListResourcesResult

Get combined list of all resources from all servers.

Parameters:

use_namespace – Whether to namespace URIs with server name (default True). When True, URIs are in format “server_name:original_uri” for auto-routing.

Returns:

ListResourcesResult containing all resources from all servers with server attribution in the serverName meta field. Returns empty list if client not initialized or error occurs.

list_resource_templates(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True, **kwargs: Any) → ListResourceTemplatesResult

Get combined list of all resource templates from all servers.

Parameters:

use_namespace – Whether to namespace URI templates with server name (default True). When True, templates are in format “server_name:original_template”.

Returns:

ListResourceTemplatesResult containing all templates from all servers with server attribution in the serverName meta field. Returns empty list if client not initialized or error occurs.

call_tool(name: str, arguments: Dict, read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, timeout: float | None = None, *, meta: dict[str, Any] | None = None, server_name: str | None = None, **kwargs: Any) → CallToolResult

Call MCP tool synchronously with optional timeout.

Parameters:

• name – Name of the tool to call

• arguments – Tool arguments as dictionary

• timeout – Maximum seconds to wait. None means wait forever.

Returns:

CallToolResult object. If timeout occurs, returns an error result.

read_resource(uri: str | AnyUrl, server_name: str | None = None, timeout: float | None = None, **kwargs: Any) → ReadResourceResult

Read a resource with optional auto-routing via namespaced URIs.

Parameters:

• uri – Resource URI. Can be namespaced as “server:uri” for auto-routing. URIs from list_resources() are already namespaced for convenience.

• server_name – Optional explicit server name. If provided, assumes that there is no namespace in the provided URI.

• timeout – Maximum seconds to wait. None means wait forever.

Returns:

ReadResourceResult containing the resource content. Returns empty result if client not initialized or timeout occurs.

Examples

>>> # Auto-routing with namespaced URI (from list_resources())
>>> resources = client.list_resources().resources
>>> result = client.read_resource(resources[0].uri)

>>> # Explicit server (no namespace in URI)
>>> result = client.read_resource("file:///path", server_name="filesystem")

>>> # Manual namespacing
>>> result = client.read_resource("filesystem:file:///path")

get_prompt(name: str, arguments: Dict[str, str] | None = None, server_name: str | None = None, timeout: float | None = None, **kwargs: Any) → GetPromptResult

Get a prompt by automatically routing to the appropriate server.

Parameters:

• name – Name of the prompt to get.

• arguments – Optional arguments for the prompt.

• server_name – Optional server name to explicitly specify which server to use. If not provided, the server will be automatically determined from the prompt name.

• timeout – Maximum seconds to wait. None means wait forever.

Returns:

GetPromptResult containing the prompt messages. Returns empty result if client not initialized or timeout occurs.

Examples

>>> # Auto-routing (prompt name determines server)
>>> result = client.get_prompt("greeting", arguments={"name": "World"})

>>> # Explicit server
>>> result = client.get_prompt("greeting", server_name="prompts_server")

mcp_multi_server.types module

Type definitions for MCP multi-server client.

class mcp_multi_server.types.ServerCapabilities(*, name: str, tools: ListToolsResult | None = None, resources: ListResourcesResult | None = None, resource_templates: ListResourceTemplatesResult | None = None, prompts: ListPromptsResult | None = None)

Bases: BaseModel

Capabilities discovered from an MCP server.

This class stores all the capabilities (tools, resources, templates, prompts) that were discovered during server initialization. It’s used internally to track what each server can do and to aggregate capabilities across all servers.

name

The unique identifier for the server.

Type:

str

tools

List of tools provided by the server, if any.

Type:

mcp.types.ListToolsResult | None

resources

List of resources provided by the server, if any.

Type:

mcp.types.ListResourcesResult | None

resource_templates

List of resource templates provided by the server, if any.

Type:

mcp.types.ListResourceTemplatesResult | None

prompts

List of prompts provided by the server, if any.

Type:

mcp.types.ListPromptsResult | None

Note

All capability fields (tools, resources, etc.) are optional because: - A server may not implement all capability types - Capability discovery may fail for some types while succeeding for others - Empty capability lists are represented as None rather than empty lists

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

name: str

tools: ListToolsResult | None

resources: ListResourcesResult | None

resource_templates: ListResourceTemplatesResult | None

prompts: ListPromptsResult | None

model_config: ClassVar[ConfigDict] = {}

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

mcp_multi_server.utils module

Utility functions for MCP multi-server client.

mcp_multi_server.utils.configure_logging(name: str = 'mcp_multi_server', level: str = 'INFO', format: str | None = None, datefmt: str | None = None) → None

Configure logging for the mcp_multi_server library but not for the MCP servers see MultiServerClient.set_logging_level().

This function provides a convenient way to configure logging for the library. It ensures a handler is configured and sets the log level.

Note

For more control, users can configure logging directly using Python’s logging module in their application code.

Parameters:

• level – Log level as a string (DEBUG, INFO, WARNING, ERROR, CRITICAL). Defaults to “INFO”.

• format – Optional custom format string for log messages. If not provided, uses a default format with timestamp and level.

• datefmt – Optional custom date format string.

Examples:

Basic usage - set log level to DEBUG:
>>> from mcp_multi_server import configure_logging
>>> configure_logging(level="DEBUG")

Custom format:
>>> configure_logging(
...     level="INFO",
...     format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
...     datefmt="%Y-%m-%d %H:%M:%S"
... )

Using standard logging module for more control:
>>> import logging
>>> logging.getLogger("mcp_multi_server").setLevel(logging.DEBUG)
>>> # Or configure entire app:
>>> logging.basicConfig(level=logging.DEBUG)

mcp_multi_server.utils.print_capabilities_summary(client: MultiServerClient | SyncMultiServerClient) → None

Utility function to print a summary of all discovered capabilities in a MultiServerClient object.

mcp_multi_server.utils.mcp_tools_to_openai_format(tools: List[Tool]) → List[Dict[str, Any]]

Convert MCP tools to OpenAI function calling format.

This function transforms MCP tool definitions into the format expected by OpenAI’s function calling API, enabling seamless integration between MCP servers and OpenAI language models.

Parameters:

tools – List of MCP Tool objects to convert.

Returns:

• type: Always “function”

• function: Dict containing name, description, and parameters (JSON schema)

Return type:

List of tool definitions in OpenAI format, where each tool is a dict with

Example

>>> from mcp_multi_server import MultiServerClient
>>> from mcp_multi_server.utils import mcp_tools_to_openai_format
>>> from openai import OpenAI
>>>
>>> async with MultiServerClient.from_config("mcp_servers.json") as client:
>>>     tools_result = client.list_tools().tools or []
>>>     openai_tools = mcp_tools_to_openai_format(tools_result)
>>>     openai_client = OpenAI()
>>>     messages = [
...         {"role": "user", "content": "Find the weather in New York City."}
...     ]
>>>     response = openai_client.chat.completions.create(
...                    model="gpt-4-0613",
...                    messages=messages,
...                    tools=openai_tools if openai_tools else None,
...                    tool_choice="auto" if openai_tools else None,
...                ).choices[0]

Note

The inputSchema from MCP tools is used directly as the parameters field in OpenAI format, as both follow JSON Schema specifications.

mcp_multi_server.utils.format_namespace_uri(server_name: str, uri: str | AnyUrl) → str

Format a URI with a server namespace prefix.

Parameters:

• server_name – Name of the server providing the resource.

• uri – Original URI of the resource.

Returns:

uri”.

Return type:

Namespaced URI in the format “server_name

Examples

>>> format_namespace_uri("filesystem", "file:///path/to/file.txt")
'filesystem:file:///path/to/file.txt'
>>> format_namespace_uri("db", "records://users/123")
'db:records://users/123'

Note

This function is used internally by the client to namespace resource URIs for auto-routing. Users typically don’t need to call this directly.

mcp_multi_server.utils.parse_namespace_uri(uri: str | AnyUrl) → tuple[str | None, str]

Parse a namespaced URI to extract server name and original URI.

Parameters:

uri – URI that may contain a server namespace prefix.

Returns:

Tuple of (server_name, uri). If no namespace is present, server_name is None and uri is the original input.

Examples

>>> parse_namespace_uri("filesystem:file:///path/to/file.txt")
('filesystem', 'file:///path/to/file.txt')
>>> parse_namespace_uri("file:///path/to/file.txt")
(None, 'file:///path/to/file.txt')
>>> parse_namespace_uri("db:records://users/123")
('db', 'records://users/123')

Note

This function distinguishes between protocol schemes (scheme://) and namespace prefixes (namespace:). Protocol schemes are not treated as namespaces.

mcp_multi_server.utils.extract_template_variables(uri_template: str | AnyUrl) → List[str]

Extract variable names from a URI template.

URI templates use curly braces to denote variables that should be substituted. Duplicate variables are automatically deduplicated while preserving order.

Parameters:

uri_template – URI template string with variables in {variable} format.

Returns:

List of unique variable names found in the template (without braces), in order of first appearance.

Examples

>>> extract_template_variables("file:///{path}/to/{filename}")
['path', 'filename']
>>> extract_template_variables("users/{id}/posts/{post_id}")
['id', 'post_id']
>>> extract_template_variables("users/{id}/posts/{id}")
['id']
>>> extract_template_variables("no/variables/here")
[]

mcp_multi_server.utils.substitute_template_variables(uri_template: str | AnyUrl, variables: Dict[str, str]) → str

Substitute variables in URI template with provided values.

Variable values are URL-encoded to handle spaces and special characters properly.

Parameters:

• uri_template – URI template string with variables in {variable} format.

• variables – Dictionary mapping variable names to their replacement values.

Returns:

URI with all variables replaced by their encoded values. Special characters in values are percent-encoded to ensure valid URIs.

Examples

>>> substitute_template_variables(
...     "file:///{path}/{filename}",
...     {"path": "my documents", "filename": "report.txt"}
... )
'file:///my%20documents/report.txt'
>>> substitute_template_variables(
...     "users/{id}",
...     {"id": "123"}
... )
'users/123'

Note

Values are URL-encoded to ensure proper handling of special characters in URIs.

Module contents

MCP Multi-Server Client Library.

A Python library for managing connections to multiple MCP (Model Context Protocol) servers. This library provides a unified interface for discovering, aggregating, and routing capabilities (tools, resources, prompts) across multiple MCP servers.

Key Features:

• Connect to multiple MCP servers simultaneously

• Automatic capability discovery and aggregation

• Intelligent routing of tool calls, resource reads, and prompt retrievals

• Namespace-based URI routing for resources and resource templates

• Collision detection for tools and prompts

Examples:

Basic usage with context manager:
>>> async with MultiServerClient.from_config("mcp_servers.json") as client:
...     tools = client.list_tools()
...     result = await client.call_tool("my_tool", {"arg": "value"})

Programmatic configuration:
>>> from mcp_multi_server import MultiServerClient, MCPServersConfig, ServerConfig
>>> config = MCPServersConfig(mcpServers={
...     "my_server": ServerConfig(command="python", args=["-m", "my_server"])
... })
>>> async with MultiServerClient.from_dict(config.model_dump()) as client:
...     tools = client.list_tools()

Synchronous client (for non-async code):
>>> from mcp_multi_server import SyncMultiServerClient
>>> with SyncMultiServerClient.from_config("mcp_servers.json") as client:
...     tools = client.list_tools()
...     result = client.call_tool("my_tool", {"arg": "value"})

Sync client with programmatic configuration:
>>> config = {"mcpServers": {"my_server": {"command": "python", "args": ["-m", "my_server"]}}}
>>> with SyncMultiServerClient.from_dict(config) as client:
...     resources = client.list_resources()

OpenAI integration:
>>> from mcp_multi_server import mcp_tools_to_openai_format
>>> tools = client.list_tools()
>>> openai_tools = mcp_tools_to_openai_format(tools.tools)

Configuring logging:
>>> from mcp_multi_server import configure_logging
>>> configure_logging(level="DEBUG")  # Enable debug logging
>>> # Or use Python's logging module directly:
>>> import logging
>>> logging.getLogger("mcp_multi_server").setLevel(logging.DEBUG)

See also

• MCP Protocol Documentation: https://modelcontextprotocol.io

• GitHub Repository: https://github.com/apisani1/mcp-multi-server

• Examples: https://github.com/apisani1/mcp-multi-server/tree/main/examples

class mcp_multi_server.MultiServerClient(config_path: str | Path = 'mcp_servers.json')

Bases: object

Manages multiple MCP server connections for a MCP host.

This class handles:

• Connecting to multiple MCP servers

• Discovering and aggregating server capabilities (tools, resources, templates, prompts)

• Routing tool, prompt and resource calls to the correct server

• Managing session lifecycles with AsyncExitStack

The client can be used as an async context manager for automatic cleanup:

Examples:

Basic usage with context manager:
>>> async with MultiServerClient.from_config("mcp_servers.json") as client:
...     tools = client.list_tools()
...     result = await client.call_tool("my_tool", {"arg": "value"})

Manual connection management:
>>> async with AsyncExitStack() as stack:
...     client = MultiServerClient("mcp_servers.json")
...     await client.connect_all(stack)
...     tools = client.list_tools()

Programmatic configuration:
>>> config = MCPServersConfig(mcpServers={
...     "my_server": ServerConfig(command="python", args=["-m", "my_server"])
... })
>>> async with MultiServerClient.from_dict(config.model_dump()) as client:
...     tools = client.list_tools()

Initialize the multi-server client.

Parameters:

config_path – Path to the JSON configuration file containing server definitions. Defaults to “mcp_servers.json” in the current directory.

Note

This constructor only sets up the configuration path. The actual connection to servers happens when connect_all() is called or when using the client as a context manager.

__init__(config_path: str | Path = 'mcp_servers.json') → None

Initialize the multi-server client.

Parameters:

config_path – Path to the JSON configuration file containing server definitions. Defaults to “mcp_servers.json” in the current directory.

Note

This constructor only sets up the configuration path. The actual connection to servers happens when connect_all() is called or when using the client as a context manager.

classmethod from_config(config_path: str | Path) → MultiServerClient

Create a client from a configuration file path.

This is a convenience class method that’s equivalent to the regular constructor.

Parameters:

config_path – Path to the JSON configuration file.

Returns:

A new MultiServerClient instance.

Examples

>>> client = MultiServerClient.from_config("my_servers.json")

classmethod from_dict(config_dict: Dict[str, Any]) → MultiServerClient

Create a client from a configuration dictionary.

This method allows programmatic configuration without needing a JSON file.

Parameters:

config_dict – Dictionary containing server configurations in the same format as the JSON file (with “mcpServers” key).

Returns:

A new MultiServerClient instance with the provided configuration.

Raises:

pydantic.ValidationError – If the config dictionary doesn’t match the schema.

Examples

>>> config = {
...     "mcpServers": {
...         "tool_server": {
...             "command": "python",
...             "args": ["-m", "my_package.tool_server"]
...         }
...     }
... }
>>> client = MultiServerClient.from_dict(config)

async connect_all(stack: AsyncExitStack) → None

Connect to all configured MCP servers and discover their capabilities.

Parameters:

stack – AsyncExitStack for managing async context managers.

Raises:

• FileNotFoundError – If config file doesn’t exist.

• json.JSONDecodeError – If config file is invalid JSON.

• pydantic.ValidationError – If config data doesn’t match schema.

Note

Individual server connection failures are caught and logged as warnings. The method will continue connecting to remaining servers if one fails.

async set_logging_level(level: Literal['debug', 'info', 'notice', 'warning', 'error', 'critical', 'alert', 'emergency']) → EmptyResult

Set the logging level for the multi-server client and the MCP connected servers.

Parameters:

level – Logging level as a string in lower case (e.g., “debug”, “info”, “notice”, “warning”, “error”, “critical”, “alert”, “emergency”) as defined in MCP LoggingLevel.

Note

The following mappings of MCP to Python logging leves are applied: - “notice” -> “WARNING” - “alert” and “emergency” -> “CRITICAL”

Examples

>>> await MultiServerClient.set_logging_level("debug")

list_tools(cursor: str | None = None, *, params: PaginatedRequestParams | None = None) → ListToolsResult

Get combined list of all tools from all servers.

This method mimics the MCP ClientSession.list_tools() signature but aggregates tools from all connected servers. Server attribution is included in each tool’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

Returns:

ListToolsResult containing all tools from all servers with the server name in the serverName meta field. The nextCursor field is always None (pagination not supported).

Raises:

ValueError – If cursor or params is not None (pagination not supported).

Examples

>>> result = client.list_tools()
>>> for tool in result.tools:
...     server = tool.meta.get("serverName") if tool.meta else None
...     print(f"{tool.name} from {server}")

list_prompts(cursor: str | None = None, *, params: PaginatedRequestParams | None = None) → ListPromptsResult

Get combined list of all prompts from all servers.

This method mimics the MCP ClientSession.list_prompts() signature but aggregates prompts from all connected servers. Server attribution is included in each prompt’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

Returns:

ListPromptsResult containing all prompts from all servers with the server name in the serverName meta fieldthe. The nextCursor field is always None (pagination not supported).

Raises:

ValueError – If cursor or params is not None (pagination not supported).

Examples

>>> result = client.list_prompts()
>>> for prompt in result.prompts:
...     server = prompt.meta.get("serverName") if prompt.meta else None
...     print(f"{prompt.name} from {server}")

list_resources(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True) → ListResourcesResult

Get combined list of all resources from all servers.

This method mimics the MCP ClientSession.list_resources() signature but aggregates resources from all connected servers. Resources are returned with namespaced URIs (server:uri format) for auto-routing, and server attribution is included in each resource’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

• use_namespace – Whether to namespace the URIs with the server name.

Returns:

• Namespaced URIs in format “server_name:original_uri” for auto-routing

• the server name in the serverName meta field for explicit server identification

The nextCursor field is always None (pagination not supported).

Return type:

ListResourcesResult containing all resources from all servers with

Raises:

ValueError – If cursor is not None (pagination not supported).

Examples

>>> result = client.list_resources()
>>> for resource in result.resources:
...     server = resource.meta.get("serverName") if resource.meta else None
...     # URI is already namespaced: "filesystem:file:///path"
...     content = await client.read_resource(resource.uri)

list_resource_templates(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True) → ListResourceTemplatesResult

Get combined list of all resource templates from all servers.

This method mimics the MCP ClientSession.list_resource_templates() signature but aggregates resource templates from all connected servers. Templates are returned with namespaced URI templates (server:template format) for auto-routing, and server attribution is included in each template’s meta field.

Parameters:

• cursor – Optional pagination cursor. Not supported for multi-server aggregation, must be None if provided.

• params – Optional PaginatedRequestParams. Not supported for multi-server aggregation, must be None if provided.

• use_namespace – Whether to namespace the URI templates with the server name.

Returns:

• Namespaced URI templates in format “server_name:original_template”

• the server name in the serverName meta field for explicit server identification

The nextCursor field is always None (pagination not supported).

Return type:

ListResourceTemplatesResult containing all templates from all servers with

Raises:

ValueError – If cursor is not None (pagination not supported).

Examples

>>> result = client.list_resource_templates()
>>> for template in result.resourceTemplates:
...     server = template.meta.get("serverName") if template.meta else None
...     # URI template is already namespaced: "filesystem:file:///{path}"
...     # Needs to be filled in with actual path when used
...     uri = template.uriTemplate.replace("{path}", "example.txt")
...     content = await client.read_resource(uri)

async call_tool(name: str, arguments: Dict[str, Any], read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None, server_name: str | None = None) → CallToolResult

Route a tool call to the appropriate server.

Parameters:

• name – Name of the tool to call.

• arguments – Arguments to pass to the tool.

• read_timeout_seconds – Optional timeout for reading the tool result.

• progress_callback – Optional callback for progress notifications.

• meta – Optional metadata dictionary to pass additional information.

• server_name – Optional server name to explicitly specify which server to use. If not provided, the server will be automatically determined from the tool name.

Returns:

Result from the tool execution. If the tool name is not found or routing fails, returns a CallToolResult with isError=True containing an error message.

Raises:

• McpError – If the tool execution fails or times out (protocol-level errors).

• RuntimeError – If tool result validation fails (invalid structured content or schema).

Note

Routing errors (unknown tool, unknown server) are returned as error results (isError=True) rather than raising exceptions, following MCP protocol conventions. Protocol-level errors from the underlying session are propagated as exceptions.

async read_resource(uri: str | AnyUrl, server_name: str | None = None) → ReadResourceResult

Read a resource with optional auto-routing via namespaced URIs.

Parameters:

• uri – Resource URI. Can be namespaced as “server:uri” for auto-routing. URIs from list_resources() are already namespaced for convenience. Accepts both str and AnyUrl types for MCP library compatibility.

• server_name – Optional explicit server name. If provided, assumes that there is no any namespace in the provided URI.

Returns:

Resource content.

Raises:

McpError – If server name is not found, URI is not namespaced when server_name is not provided, or if the resource read fails or times out.

Examples:

Auto-routing with namespaced URI (from list_resources()):
>>> resources = client.list_resources().resources
>>> result = await client.read_resource(resources[0].uri)

Explicit server (no namespace should be present in URI):
>>> result = await client.read_resource("file:///path", server_name="filesystem")

Manual namespacing:
>>> result = await client.read_resource("filesystem:file:///path")

Note

Raises McpError for both routing errors and protocol-level errors to align with MCP SDK behavior.

async get_prompt(name: str, arguments: Dict[str, str] | None = None, server_name: str | None = None) → GetPromptResult

Get a prompt by automatically routing to the appropriate server.

Parameters:

• name – Name of the prompt to get.

• arguments – Optional arguments for the prompt.

• server_name – Optional server name to explicitly specify which server to use. If not provided, the server will be automatically determined from the prompt name.

Returns:

Prompt result.

Raises:

McpError – If prompt name is not found, server name is not found, or if the prompt retrieval fails or times out.

Note

Raises McpError for both routing errors and protocol-level errors to align with MCP SDK behavior.

class mcp_multi_server.SyncMultiServerClient(config_path: str | Path | None = None, config_dict: Dict[str, Any] | None = None)

Bases: object

Manages MCP multi server client in a background thread with persistent event loop.

This class provides a synchronous interface to the async MultiServerClient, making it easier to integrate with synchronous code while maintaining the efficiency of async operations.

Usage:

# Context manager (recommended) with SyncMultiServerClient(config_path) as client:

tools = client.list_tools() result = client.call_tool(“tool_name”, {“arg”: “value”})

# Manual lifecycle management client = SyncMultiServerClient(config_path) tools = client.list_tools() # … use client … client.shutdown()

Thread Safety:

All public methods are thread-safe, using asyncio.run_coroutine_threadsafe() to schedule operations on the background event loop.

Initialize SyncMultiServerClient.

Starts background thread and initializes MCP client during construction. Automatically registers cleanup handler to ensure proper shutdown on program exit.

Parameters:

• config_path – Path to MCP configuration file.

• config_dict – Configuration dictionary (alternative to config_path).

Raises:

• ValueError – If neither or both config_path and config_dict are provided.

• Exception – If MCP client initialization fails.

__init__(config_path: str | Path | None = None, config_dict: Dict[str, Any] | None = None)

Initialize SyncMultiServerClient.

Starts background thread and initializes MCP client during construction. Automatically registers cleanup handler to ensure proper shutdown on program exit.

Parameters:

• config_path – Path to MCP configuration file.

• config_dict – Configuration dictionary (alternative to config_path).

Raises:

• ValueError – If neither or both config_path and config_dict are provided.

• Exception – If MCP client initialization fails.

classmethod from_config(config_path: str | Path) → SyncMultiServerClient

Create a client from a configuration file path.

This is a convenience class method that’s equivalent to calling the constructor with a config_path argument.

Parameters:

config_path – Path to the JSON configuration file.

Returns:

A new SyncMultiServerClient instance.

Examples

>>> client = SyncMultiServerClient.from_config("mcp_config.json")

classmethod from_dict(config_dict: Dict[str, Any]) → SyncMultiServerClient

Create a client from a configuration dictionary.

This method allows programmatic configuration without needing a JSON file.

Parameters:

config_dict – Dictionary containing server configurations in the same format as the JSON file (with “mcpServers” key).

Returns:

A new SyncMultiServerClient instance with the provided configuration.

Examples

>>> config = {
...     "mcpServers": {
...         "tool_server": {
...             "command": "python",
...             "args": ["-m", "my_package.tool_server"]
...         }
...     }
... }
>>> client = SyncMultiServerClient.from_dict(config)

shutdown() → None

Shutdown background thread and cleanup MCP client.

Safe to call multiple times. Waits up to 10 seconds for graceful cleanup.

set_logging_level(level: Literal['debug', 'info', 'notice', 'warning', 'error', 'critical', 'alert', 'emergency']) → EmptyResult

Set the logging level for the multi-server client and the MCP connected servers.

Parameters:

level – Logging level as a string in lower case (e.g., “debug”, “info”, “notice”, “warning”, “error”, “critical”, “alert”, “emergency”) as defined in MCP LoggingLevel.

Note

The following mappings of MCP to Python logging leves are applied: - “notice” -> “WARNING” - “alert” and “emergency” -> “CRITICAL”

Examples

>>> SyncMultiServerClient.set_logging_level("debug")

property capabilities: Dict[str, Any]

Get combined capabilities from all connected MCP servers.

Returns:

Dictionary containing combined capabilities from all servers. Returns empty dict if client not initialized or error occurs.

list_tools(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, **kwargs: Any) → ListToolsResult

List available MCP tools in raw MCP format.

Returns:

List of MCP Tool objects (not converted to OpenAI format). Returns empty list if client not initialized or error occurs.

list_prompts(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, **kwargs: Any) → ListPromptsResult

Get combined list of all prompts from all servers.

Returns:

ListPromptsResult containing all prompts from all servers with server attribution in the serverName meta field. Returns empty list if client not initialized or error occurs.

list_resources(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True, **kwargs: Any) → ListResourcesResult

Get combined list of all resources from all servers.

Parameters:

use_namespace – Whether to namespace URIs with server name (default True). When True, URIs are in format “server_name:original_uri” for auto-routing.

Returns:

ListResourcesResult containing all resources from all servers with server attribution in the serverName meta field. Returns empty list if client not initialized or error occurs.

list_resource_templates(cursor: str | None = None, *, params: PaginatedRequestParams | None = None, use_namespace: bool = True, **kwargs: Any) → ListResourceTemplatesResult

Get combined list of all resource templates from all servers.

Parameters:

use_namespace – Whether to namespace URI templates with server name (default True). When True, templates are in format “server_name:original_template”.

Returns:

ListResourceTemplatesResult containing all templates from all servers with server attribution in the serverName meta field. Returns empty list if client not initialized or error occurs.

call_tool(name: str, arguments: Dict, read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, timeout: float | None = None, *, meta: dict[str, Any] | None = None, server_name: str | None = None, **kwargs: Any) → CallToolResult

Call MCP tool synchronously with optional timeout.

Parameters:

• name – Name of the tool to call

• arguments – Tool arguments as dictionary

• timeout – Maximum seconds to wait. None means wait forever.

Returns:

CallToolResult object. If timeout occurs, returns an error result.

read_resource(uri: str | AnyUrl, server_name: str | None = None, timeout: float | None = None, **kwargs: Any) → ReadResourceResult

Read a resource with optional auto-routing via namespaced URIs.

Parameters:

• uri – Resource URI. Can be namespaced as “server:uri” for auto-routing. URIs from list_resources() are already namespaced for convenience.

• server_name – Optional explicit server name. If provided, assumes that there is no namespace in the provided URI.

• timeout – Maximum seconds to wait. None means wait forever.

Returns:

ReadResourceResult containing the resource content. Returns empty result if client not initialized or timeout occurs.

Examples

>>> # Auto-routing with namespaced URI (from list_resources())
>>> resources = client.list_resources().resources
>>> result = client.read_resource(resources[0].uri)

>>> # Explicit server (no namespace in URI)
>>> result = client.read_resource("file:///path", server_name="filesystem")

>>> # Manual namespacing
>>> result = client.read_resource("filesystem:file:///path")

get_prompt(name: str, arguments: Dict[str, str] | None = None, server_name: str | None = None, timeout: float | None = None, **kwargs: Any) → GetPromptResult

Get a prompt by automatically routing to the appropriate server.

Parameters:

• name – Name of the prompt to get.

• arguments – Optional arguments for the prompt.

• server_name – Optional server name to explicitly specify which server to use. If not provided, the server will be automatically determined from the prompt name.

• timeout – Maximum seconds to wait. None means wait forever.

Returns:

GetPromptResult containing the prompt messages. Returns empty result if client not initialized or timeout occurs.

Examples

>>> # Auto-routing (prompt name determines server)
>>> result = client.get_prompt("greeting", arguments={"name": "World"})

>>> # Explicit server
>>> result = client.get_prompt("greeting", server_name="prompts_server")

class mcp_multi_server.ServerConfig(*, command: str, args: List[str])

Bases: BaseModel

Configuration for a single MCP server.

command

The executable command to start the server (e.g., “python”, “node”).

Type:

str

args

Command-line arguments to pass to the server executable.

Type:

List[str]

Examples

>>> config = ServerConfig(
...     command="python",
...     args=["-m", "mcp_servers.tool_server"]
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

command: str

args: List[str]

model_config: ClassVar[ConfigDict] = {}

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

class mcp_multi_server.MCPServersConfig(*, mcpServers: Dict[str, ServerConfig])

Bases: BaseModel

Configuration for all MCP servers.

mcpServers

Dictionary mapping server names to their configurations. Server names are used as identifiers throughout the client.

Type:

Dict[str, mcp_multi_server.config.ServerConfig]

Examples

>>> config = MCPServersConfig(mcpServers={
...     "tool_server": ServerConfig(
...         command="python",
...         args=["-m", "mcp_servers.tool_server"]
...     ),
...     "resource_server": ServerConfig(
...         command="python",
...         args=["-m", "mcp_servers.resource_server"]
...     )
... })

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

mcpServers: Dict[str, ServerConfig]

model_config: ClassVar[ConfigDict] = {}

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

class mcp_multi_server.ServerCapabilities(*, name: str, tools: ListToolsResult | None = None, resources: ListResourcesResult | None = None, resource_templates: ListResourceTemplatesResult | None = None, prompts: ListPromptsResult | None = None)

Bases: BaseModel

Capabilities discovered from an MCP server.

This class stores all the capabilities (tools, resources, templates, prompts) that were discovered during server initialization. It’s used internally to track what each server can do and to aggregate capabilities across all servers.

name

The unique identifier for the server.

Type:

str

tools

List of tools provided by the server, if any.

Type:

mcp.types.ListToolsResult | None

resources

List of resources provided by the server, if any.

Type:

mcp.types.ListResourcesResult | None

resource_templates

List of resource templates provided by the server, if any.

Type:

mcp.types.ListResourceTemplatesResult | None

prompts

List of prompts provided by the server, if any.

Type:

mcp.types.ListPromptsResult | None

Note

All capability fields (tools, resources, etc.) are optional because: - A server may not implement all capability types - Capability discovery may fail for some types while succeeding for others - Empty capability lists are represented as None rather than empty lists

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

name: str

tools: ListToolsResult | None

resources: ListResourcesResult | None

resource_templates: ListResourceTemplatesResult | None

prompts: ListPromptsResult | None

model_config: ClassVar[ConfigDict] = {}

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

mcp_multi_server.configure_logging(name: str = 'mcp_multi_server', level: str = 'INFO', format: str | None = None, datefmt: str | None = None) → None

Configure logging for the mcp_multi_server library but not for the MCP servers see MultiServerClient.set_logging_level().

This function provides a convenient way to configure logging for the library. It ensures a handler is configured and sets the log level.

Note

For more control, users can configure logging directly using Python’s logging module in their application code.

Parameters:

• level – Log level as a string (DEBUG, INFO, WARNING, ERROR, CRITICAL). Defaults to “INFO”.

• format – Optional custom format string for log messages. If not provided, uses a default format with timestamp and level.

• datefmt – Optional custom date format string.

Examples:

Basic usage - set log level to DEBUG:
>>> from mcp_multi_server import configure_logging
>>> configure_logging(level="DEBUG")

Custom format:
>>> configure_logging(
...     level="INFO",
...     format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
...     datefmt="%Y-%m-%d %H:%M:%S"
... )

Using standard logging module for more control:
>>> import logging
>>> logging.getLogger("mcp_multi_server").setLevel(logging.DEBUG)
>>> # Or configure entire app:
>>> logging.basicConfig(level=logging.DEBUG)

mcp_multi_server.mcp_tools_to_openai_format(tools: List[Tool]) → List[Dict[str, Any]]

Convert MCP tools to OpenAI function calling format.

This function transforms MCP tool definitions into the format expected by OpenAI’s function calling API, enabling seamless integration between MCP servers and OpenAI language models.

Parameters:

tools – List of MCP Tool objects to convert.

Returns:

• type: Always “function”

• function: Dict containing name, description, and parameters (JSON schema)

Return type:

List of tool definitions in OpenAI format, where each tool is a dict with

Example

>>> from mcp_multi_server import MultiServerClient
>>> from mcp_multi_server.utils import mcp_tools_to_openai_format
>>> from openai import OpenAI
>>>
>>> async with MultiServerClient.from_config("mcp_servers.json") as client:
>>>     tools_result = client.list_tools().tools or []
>>>     openai_tools = mcp_tools_to_openai_format(tools_result)
>>>     openai_client = OpenAI()
>>>     messages = [
...         {"role": "user", "content": "Find the weather in New York City."}
...     ]
>>>     response = openai_client.chat.completions.create(
...                    model="gpt-4-0613",
...                    messages=messages,
...                    tools=openai_tools if openai_tools else None,
...                    tool_choice="auto" if openai_tools else None,
...                ).choices[0]

Note

The inputSchema from MCP tools is used directly as the parameters field in OpenAI format, as both follow JSON Schema specifications.

mcp_multi_server.format_namespace_uri(server_name: str, uri: str | AnyUrl) → str

Format a URI with a server namespace prefix.

Parameters:

• server_name – Name of the server providing the resource.

• uri – Original URI of the resource.

Returns:

uri”.

Return type:

Namespaced URI in the format “server_name

Examples

>>> format_namespace_uri("filesystem", "file:///path/to/file.txt")
'filesystem:file:///path/to/file.txt'
>>> format_namespace_uri("db", "records://users/123")
'db:records://users/123'

Note

This function is used internally by the client to namespace resource URIs for auto-routing. Users typically don’t need to call this directly.

mcp_multi_server.parse_namespace_uri(uri: str | AnyUrl) → tuple[str | None, str]

Parse a namespaced URI to extract server name and original URI.

Parameters:

uri – URI that may contain a server namespace prefix.

Returns:

Tuple of (server_name, uri). If no namespace is present, server_name is None and uri is the original input.

Examples

>>> parse_namespace_uri("filesystem:file:///path/to/file.txt")
('filesystem', 'file:///path/to/file.txt')
>>> parse_namespace_uri("file:///path/to/file.txt")
(None, 'file:///path/to/file.txt')
>>> parse_namespace_uri("db:records://users/123")
('db', 'records://users/123')

Note

This function distinguishes between protocol schemes (scheme://) and namespace prefixes (namespace:). Protocol schemes are not treated as namespaces.

mcp_multi_server.extract_template_variables(uri_template: str | AnyUrl) → List[str]

Extract variable names from a URI template.

URI templates use curly braces to denote variables that should be substituted. Duplicate variables are automatically deduplicated while preserving order.

Parameters:

uri_template – URI template string with variables in {variable} format.

Returns:

List of unique variable names found in the template (without braces), in order of first appearance.

Examples

>>> extract_template_variables("file:///{path}/to/{filename}")
['path', 'filename']
>>> extract_template_variables("users/{id}/posts/{post_id}")
['id', 'post_id']
>>> extract_template_variables("users/{id}/posts/{id}")
['id']
>>> extract_template_variables("no/variables/here")
[]

mcp_multi_server.substitute_template_variables(uri_template: str | AnyUrl, variables: Dict[str, str]) → str

Substitute variables in URI template with provided values.

Variable values are URL-encoded to handle spaces and special characters properly.

Parameters:

• uri_template – URI template string with variables in {variable} format.

• variables – Dictionary mapping variable names to their replacement values.

Returns:

URI with all variables replaced by their encoded values. Special characters in values are percent-encoded to ensure valid URIs.

Examples

>>> substitute_template_variables(
...     "file:///{path}/{filename}",
...     {"path": "my documents", "filename": "report.txt"}
... )
'file:///my%20documents/report.txt'
>>> substitute_template_variables(
...     "users/{id}",
...     {"id": "123"}
... )
'users/123'

Note

Values are URL-encoded to ensure proper handling of special characters in URIs.