MCPClientManager

Manage MCP server connections.

Quick Example

from mamba_agents.mcp import MCPClientManager, MCPServerConfig

manager = MCPClientManager()

# Add server
manager.add_server(MCPServerConfig(
    name="server1",
    transport="stdio",
    command="my-server",
))

# Get toolsets for agent
toolsets = manager.as_toolsets()

With Agent

from mamba_agents import Agent
from mamba_agents.mcp import MCPClientManager, MCPServerConfig

configs = [
    MCPServerConfig(
        name="filesystem",
        transport="stdio",
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/project"],
    ),
]

manager = MCPClientManager(configs)
agent = Agent("gpt-4o", toolsets=manager.as_toolsets())
result = await agent.run("List project files")

Loading from .mcp.json

Load configurations from .mcp.json files (Claude Desktop compatible format):

from mamba_agents.mcp import MCPClientManager

# Create manager from file
manager = MCPClientManager.from_mcp_json(".mcp.json")
agent = Agent("gpt-4o", toolsets=manager.as_toolsets())

Merging Multiple Files

# Start with project config
manager = MCPClientManager.from_mcp_json("project/.mcp.json")

# Add user defaults
manager.add_from_file("~/.mcp.json")

Combining File and Programmatic Config

from mamba_agents.mcp import MCPClientManager, MCPServerConfig

# Load from file
manager = MCPClientManager.from_mcp_json(".mcp.json")

# Add additional server
manager.add_server(MCPServerConfig(
    name="custom",
    transport="stdio",
    command="custom-server",
))

Error Handling

from mamba_agents.mcp import (
    MCPClientManager,
    MCPFileNotFoundError,
    MCPFileParseError,
    MCPServerValidationError,
)

try:
    manager = MCPClientManager.from_mcp_json(".mcp.json")
except MCPFileNotFoundError:
    print("Config file not found")
except MCPFileParseError as e:
    print(f"Invalid JSON: {e}")
except MCPServerValidationError as e:
    print(f"Invalid config: {e}")

API Reference

MCPClientManager

MCPClientManager(
    configs: list[MCPServerConfig] | None = None,
)

Manages MCP server configurations and creates toolsets for pydantic-ai Agent.

The recommended pattern is to use as_toolsets() to get MCP servers and pass them directly to the Agent via the toolsets parameter. pydantic-ai handles the server lifecycle (connection/disconnection) automatically.

Example

configs = [ ... MCPServerConfig( ... name="filesystem", ... transport="stdio", ... command="npx", ... args=["-y", "@modelcontextprotocol/server-filesystem", "/project"], ... tool_prefix="fs", ... ), ... ] manager = MCPClientManager(configs) agent = Agent("gpt-4o", toolsets=manager.as_toolsets()) result = await agent.run("List project files")

Initialize the MCP client manager.

PARAMETER	DESCRIPTION
configs	Optional list of server configurations. TYPE: list[MCPServerConfig] | None DEFAULT: None

Source code in src/mamba_agents/mcp/client.py

def __init__(self, configs: list[MCPServerConfig] | None = None) -> None:
    """Initialize the MCP client manager.

    Args:
        configs: Optional list of server configurations.
    """
    self._configs = configs or []

configs property

configs: list[MCPServerConfig]

Get all server configurations.

add_server

add_server(config: MCPServerConfig) -> None

Add a server configuration.

PARAMETER	DESCRIPTION
config	Server configuration to add. TYPE: MCPServerConfig

Source code in src/mamba_agents/mcp/client.py

def add_server(self, config: MCPServerConfig) -> None:
    """Add a server configuration.

    Args:
        config: Server configuration to add.
    """
    self._configs.append(config)

as_toolsets

as_toolsets() -> list[MCPServer]

Get MCP servers as toolsets for pydantic-ai Agent.

Creates MCPServer instances from configurations. These servers should be passed to Agent via the toolsets parameter. pydantic-ai handles server lifecycle automatically (connection on first use, cleanup on exit).

RETURNS	DESCRIPTION
list[MCPServer]	List of MCPServer instances to pass to Agent(toolsets=...).

Example

manager = MCPClientManager(configs) agent = Agent("gpt-4o", toolsets=manager.as_toolsets())

Source code in src/mamba_agents/mcp/client.py

def as_toolsets(self) -> list[MCPServer]:
    """Get MCP servers as toolsets for pydantic-ai Agent.

    Creates MCPServer instances from configurations. These servers should
    be passed to Agent via the `toolsets` parameter. pydantic-ai handles
    server lifecycle automatically (connection on first use, cleanup on exit).

    Returns:
        List of MCPServer instances to pass to Agent(toolsets=...).

    Example:
        >>> manager = MCPClientManager(configs)
        >>> agent = Agent("gpt-4o", toolsets=manager.as_toolsets())
    """
    servers: list[MCPServer] = []
    for config in self._configs:
        server = self._create_server(config)
        servers.append(server)
    return servers

from_mcp_json classmethod

from_mcp_json(path: str | Path) -> MCPClientManager

Create an MCPClientManager from a .mcp.json file.

Parses the .mcp.json file format (compatible with Claude Desktop) and creates a manager with the loaded configurations.

PARAMETER	DESCRIPTION
path	Path to the .mcp.json file. Can be a string or Path object. Supports ~ expansion for user home directory. TYPE: str | Path

RETURNS	DESCRIPTION
MCPClientManager	MCPClientManager instance with loaded configurations.

RAISES	DESCRIPTION
MCPFileNotFoundError	If the file does not exist.
MCPFileParseError	If the file is not valid JSON.
MCPServerValidationError	If a server entry is invalid.

Example

manager = MCPClientManager.from_mcp_json(".mcp.json") agent = Agent("gpt-4o", toolsets=manager.as_toolsets())

Source code in src/mamba_agents/mcp/client.py

@classmethod
def from_mcp_json(cls, path: str | Path) -> MCPClientManager:
    """Create an MCPClientManager from a .mcp.json file.

    Parses the .mcp.json file format (compatible with Claude Desktop) and
    creates a manager with the loaded configurations.

    Args:
        path: Path to the .mcp.json file. Can be a string or Path object.
              Supports ~ expansion for user home directory.

    Returns:
        MCPClientManager instance with loaded configurations.

    Raises:
        MCPFileNotFoundError: If the file does not exist.
        MCPFileParseError: If the file is not valid JSON.
        MCPServerValidationError: If a server entry is invalid.

    Example:
        >>> manager = MCPClientManager.from_mcp_json(".mcp.json")
        >>> agent = Agent("gpt-4o", toolsets=manager.as_toolsets())
    """
    configs = load_mcp_json(path)
    return cls(configs)

add_from_file

add_from_file(path: str | Path) -> None

Add server configurations from a .mcp.json file.

Parses the .mcp.json file and appends configurations to the existing list. Useful for merging multiple configuration sources.

PARAMETER	DESCRIPTION
path	Path to the .mcp.json file. Can be a string or Path object. Supports ~ expansion for user home directory. TYPE: str | Path

RAISES	DESCRIPTION
MCPFileNotFoundError	If the file does not exist.
MCPFileParseError	If the file is not valid JSON.
MCPServerValidationError	If a server entry is invalid.

Example

manager = MCPClientManager(existing_configs) manager.add_from_file("project/.mcp.json") manager.add_from_file("~/.mcp.json") # User defaults

Source code in src/mamba_agents/mcp/client.py

def add_from_file(self, path: str | Path) -> None:
    """Add server configurations from a .mcp.json file.

    Parses the .mcp.json file and appends configurations to the existing
    list. Useful for merging multiple configuration sources.

    Args:
        path: Path to the .mcp.json file. Can be a string or Path object.
              Supports ~ expansion for user home directory.

    Raises:
        MCPFileNotFoundError: If the file does not exist.
        MCPFileParseError: If the file is not valid JSON.
        MCPServerValidationError: If a server entry is invalid.

    Example:
        >>> manager = MCPClientManager(existing_configs)
        >>> manager.add_from_file("project/.mcp.json")
        >>> manager.add_from_file("~/.mcp.json")  # User defaults
    """
    configs = load_mcp_json(path)
    self._configs.extend(configs)

get_server

get_server(name: str) -> MCPServer

Get a single MCP server instance by name.

Creates an MCPServer instance from the configuration with the given name.

PARAMETER	DESCRIPTION
name	Name of the server to retrieve. TYPE: str

RETURNS	DESCRIPTION
MCPServer	MCPServer instance.

RAISES	DESCRIPTION
MCPServerNotFoundError	If no server with the given name exists.

Example

manager = MCPClientManager(configs) server = manager.get_server("filesystem")

Source code in src/mamba_agents/mcp/client.py

def get_server(self, name: str) -> MCPServer:
    """Get a single MCP server instance by name.

    Creates an MCPServer instance from the configuration with the given name.

    Args:
        name: Name of the server to retrieve.

    Returns:
        MCPServer instance.

    Raises:
        MCPServerNotFoundError: If no server with the given name exists.

    Example:
        >>> manager = MCPClientManager(configs)
        >>> server = manager.get_server("filesystem")
    """
    config = self._get_config_by_name(name)
    if config is None:
        raise MCPServerNotFoundError(f"Server not found: {name}")
    return self._create_server(config)

test_connection async

test_connection(server_name: str) -> MCPConnectionResult

Test connection to an MCP server.

Creates a server instance, connects to it, and retrieves the list of available tools to verify the connection is working.

PARAMETER	DESCRIPTION
server_name	Name of the server to test. TYPE: str

RETURNS	DESCRIPTION
MCPConnectionResult	MCPConnectionResult with connection status and tool information.

Example

manager = MCPClientManager(configs) result = await manager.test_connection("filesystem") if result.success: ... print(f"Connected! {result.tool_count} tools available")

Source code in src/mamba_agents/mcp/client.py

async def test_connection(self, server_name: str) -> MCPConnectionResult:
    """Test connection to an MCP server.

    Creates a server instance, connects to it, and retrieves the list of
    available tools to verify the connection is working.

    Args:
        server_name: Name of the server to test.

    Returns:
        MCPConnectionResult with connection status and tool information.

    Example:
        >>> manager = MCPClientManager(configs)
        >>> result = await manager.test_connection("filesystem")
        >>> if result.success:
        ...     print(f"Connected! {result.tool_count} tools available")
    """
    config = self._get_config_by_name(server_name)
    if config is None:
        return MCPConnectionResult(
            server_name=server_name,
            success=False,
            error=f"Server not found: {server_name}",
            error_type="MCPServerNotFoundError",
        )

    try:
        server = self._create_server(config)
        async with server:
            is_running = server.is_running
            tools_raw = await server.list_tools()
            tools = [
                MCPToolInfo(
                    name=t.name,
                    description=t.description,
                    input_schema=t.inputSchema,
                )
                for t in tools_raw
            ]
            return MCPConnectionResult(
                server_name=server_name,
                success=True,
                is_running=is_running,
                tools=tools,
                tool_count=len(tools),
            )
    except TimeoutError as e:
        return MCPConnectionResult(
            server_name=server_name,
            success=False,
            error=f"Connection to {server_name} timed out: {e}",
            error_type="MCPConnectionTimeoutError",
        )
    except Exception as e:
        return MCPConnectionResult(
            server_name=server_name,
            success=False,
            error=str(e),
            error_type=type(e).__name__,
        )

test_all_connections async

test_all_connections() -> dict[str, MCPConnectionResult]

Test connections to all configured MCP servers.

Tests each server concurrently and returns results for all.

RETURNS	DESCRIPTION
dict[str, MCPConnectionResult]	Dictionary mapping server names to their connection results.

Example

manager = MCPClientManager(configs) results = await manager.test_all_connections() for name, result in results.items(): ... status = "OK" if result.success else "FAILED" ... print(f"{name}: {status}")

Source code in src/mamba_agents/mcp/client.py

async def test_all_connections(self) -> dict[str, MCPConnectionResult]:
    """Test connections to all configured MCP servers.

    Tests each server concurrently and returns results for all.

    Returns:
        Dictionary mapping server names to their connection results.

    Example:
        >>> manager = MCPClientManager(configs)
        >>> results = await manager.test_all_connections()
        >>> for name, result in results.items():
        ...     status = "OK" if result.success else "FAILED"
        ...     print(f"{name}: {status}")
    """
    tasks = [self.test_connection(config.name) for config in self._configs]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    return_dict: dict[str, MCPConnectionResult] = {}
    for config, result in zip(self._configs, results, strict=True):
        if isinstance(result, Exception):
            return_dict[config.name] = MCPConnectionResult(
                server_name=config.name,
                success=False,
                error=str(result),
                error_type=type(result).__name__,
            )
        else:
            return_dict[config.name] = result
    return return_dict

test_connection_sync

test_connection_sync(
    server_name: str,
) -> MCPConnectionResult

Synchronous wrapper for test_connection.

PARAMETER	DESCRIPTION
server_name	Name of the server to test. TYPE: str

RETURNS	DESCRIPTION
MCPConnectionResult	MCPConnectionResult with connection status and tool information.

Example

manager = MCPClientManager(configs) result = manager.test_connection_sync("filesystem") print(f"Success: {result.success}")

Source code in src/mamba_agents/mcp/client.py

def test_connection_sync(self, server_name: str) -> MCPConnectionResult:
    """Synchronous wrapper for test_connection.

    Args:
        server_name: Name of the server to test.

    Returns:
        MCPConnectionResult with connection status and tool information.

    Example:
        >>> manager = MCPClientManager(configs)
        >>> result = manager.test_connection_sync("filesystem")
        >>> print(f"Success: {result.success}")
    """
    return asyncio.run(self.test_connection(server_name))

test_all_connections_sync

test_all_connections_sync() -> dict[
    str, MCPConnectionResult
]

Synchronous wrapper for test_all_connections.

RETURNS	DESCRIPTION
dict[str, MCPConnectionResult]	Dictionary mapping server names to their connection results.

Example

manager = MCPClientManager(configs) results = manager.test_all_connections_sync()

Source code in src/mamba_agents/mcp/client.py

def test_all_connections_sync(self) -> dict[str, MCPConnectionResult]:
    """Synchronous wrapper for test_all_connections.

    Returns:
        Dictionary mapping server names to their connection results.

    Example:
        >>> manager = MCPClientManager(configs)
        >>> results = manager.test_all_connections_sync()
    """
    return asyncio.run(self.test_all_connections())