MCP Integration¶

Mamba Agents supports the Model Context Protocol (MCP) for connecting to external tool servers.

Overview¶

MCP allows your agent to use tools provided by external servers:

Stdio transport - Run MCP servers as subprocesses
SSE transport - Connect to HTTP-based MCP servers (Server-Sent Events)
Streamable HTTP transport - Connect to modern HTTP-based MCP servers (v0.1.3+)
Authentication - API key support for secure servers
Tool prefixing - Avoid name conflicts between servers
File-based config - Load from .mcp.json files (Claude Desktop compatible)
Connection testing - Verify server connectivity before use (v0.1.3+)

Quick Start¶

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

# Configure MCP servers
configs = [
    MCPServerConfig(
        name="filesystem",
        transport="stdio",
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/project"],
        tool_prefix="fs",  # Tools become: fs_read, fs_write, etc.
    ),
]

# Create manager and get toolsets
manager = MCPClientManager(configs)

# Pass toolsets to Agent (pydantic-ai handles server lifecycle)
agent = Agent("gpt-4o", toolsets=manager.as_toolsets())

# Use the agent - MCP servers connect automatically
result = await agent.run("List files in the project")

Loading from .mcp.json Files¶

Mamba Agents can load MCP server configurations from .mcp.json files, which is the standard format used by Claude Desktop, VS Code extensions, and other MCP tools.

Basic Usage¶

from mamba_agents import Agent
from mamba_agents.mcp import MCPClientManager

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

File Format¶

The .mcp.json file uses a standard format compatible with Claude Desktop:

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/project"],
      "env": {"NODE_ENV": "production"}
    },
    "web-search": {
      "url": "http://localhost:8080/sse"
    }
  }
}

Extended Fields¶

Mamba Agents supports additional fields beyond the standard format:

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem"],
      "tool_prefix": "fs",
      "env_file": ".env.local"
    },
    "api-server": {
      "url": "http://localhost:8080/sse",
      "tool_prefix": "api"
    }
  }
}

Field	Description
`command`	Command to run (stdio transport)
`args`	Command arguments
`env`	Environment variables
`url`	Server URL (SSE transport)
`tool_prefix`	Prefix for tool names (mamba-agents extension)
`env_file`	Path to .env file (mamba-agents extension)

Loading Multiple Files¶

Merge configurations from multiple sources:

from mamba_agents.mcp import MCPClientManager

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

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

# Add team shared config
manager.add_from_file("/shared/team.mcp.json")

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

Combining with Programmatic Config¶

Mix file-based and programmatic configuration:

from mamba_agents.mcp import MCPClientManager, MCPServerConfig

# Start with file-based config
manager = MCPClientManager.from_mcp_json(".mcp.json")

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

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

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 server config: {e}")

Using load_mcp_json Directly¶

For more control, use the load_mcp_json function directly:

from mamba_agents.mcp import load_mcp_json, MCPClientManager

# Load and inspect configs before creating manager
configs = load_mcp_json(".mcp.json")

# Filter or modify configs
filtered = [c for c in configs if c.name.startswith("prod-")]

manager = MCPClientManager(filtered)

Transport Types¶

Stdio Transport¶

Run MCP servers as subprocesses:

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

# Or with a local command
MCPServerConfig(
    name="custom",
    transport="stdio",
    command="python",
    args=["-m", "my_mcp_server"],
)

SSE Transport¶

Connect to HTTP-based servers using Server-Sent Events:

MCPServerConfig(
    name="web-tools",
    transport="sse",
    url="http://localhost:8080/sse",
)

Streamable HTTP Transport (v0.1.3+)¶

Connect to modern HTTP-based MCP servers:

MCPServerConfig(
    name="api-server",
    transport="streamable_http",
    url="http://localhost:8080/mcp",
    timeout=60,        # Connection timeout
    read_timeout=300,  # Read timeout for long operations
)

Transport Auto-Detection¶

When loading from .mcp.json files, transport is auto-detected from URLs:

URL Pattern	Detected Transport
Ends with `/sse`	SSE
Other URLs	Streamable HTTP
Has `command` field	Stdio

{
  "mcpServers": {
    "sse-server": {
      "url": "http://localhost:8080/sse"
    },
    "http-server": {
      "url": "http://localhost:8080/mcp"
    }
  }
}

Authentication¶

API Key Authentication¶

from mamba_agents.mcp import MCPServerConfig, MCPAuthConfig

MCPServerConfig(
    name="secure-server",
    transport="sse",
    url="https://api.example.com/mcp",
    auth=MCPAuthConfig(
        type="api_key",
        key="my-api-key",  # Direct key
        header="Authorization",  # Header name (default)
    ),
)

Using Environment Variables¶

MCPAuthConfig(
    type="api_key",
    key_env="MY_API_KEY",  # Read from env var
)

# Or with ${} syntax
MCPAuthConfig(
    type="api_key",
    key="${MY_API_KEY}",  # Expanded at runtime
)

Connection Testing (v0.1.3+)¶

Test MCP server connectivity before using them with agents.

Test Individual Server¶

from mamba_agents.mcp import MCPClientManager

manager = MCPClientManager(configs)

# Async
result = await manager.test_connection("filesystem")

# Sync
result = manager.test_connection_sync("filesystem")

if result.success:
    print(f"Connected to {result.server_name}")
    print(f"Server running: {result.is_running}")
    print(f"Tools available: {result.tool_count}")
    for tool in result.tools:
        print(f"  - {tool.name}: {tool.description}")
else:
    print(f"Connection failed: {result.error}")
    print(f"Error type: {result.error_type}")

Test All Servers¶

# Async
results = await manager.test_all_connections()

# Sync
results = manager.test_all_connections_sync()

for name, result in results.items():
    if result.success:
        print(f"{name}: OK ({result.tool_count} tools)")
    else:
        print(f"{name}: FAILED - {result.error}")

MCPConnectionResult¶

Field	Type	Description
`server_name`	str	Server that was tested
`success`	bool	Whether connection succeeded
`is_running`	bool	Whether server is running
`tools`	list[MCPToolInfo]	Available tools
`tool_count`	int	Number of tools
`error`	str	Error message if failed
`error_type`	str	Error class name

MCPToolInfo¶

Field	Type	Description
`name`	str	Tool name
`description`	str	Tool description
`input_schema`	dict	JSON schema for inputs

MCPClientManager¶

Recommended Usage¶

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

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

# Create manager
manager = MCPClientManager(configs)

# Get toolsets and pass to Agent
agent = Agent("gpt-4o", toolsets=manager.as_toolsets())

# pydantic-ai handles MCP server lifecycle automatically
result = await agent.run("Use the MCP tools")

Adding Servers Dynamically¶

manager = MCPClientManager()

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

manager.add_server(MCPServerConfig(
    name="server2",
    transport="sse",
    url="http://localhost:8080/sse",
))

# Get all toolsets
toolsets = manager.as_toolsets()
agent = Agent("gpt-4o", toolsets=toolsets)

Tool Prefixing¶

Avoid name conflicts with tool prefixes:

MCPServerConfig(
    name="server1",
    transport="stdio",
    command="server1",
    tool_prefix="s1",  # Tools become: s1_read, s1_write, etc.
)

MCPServerConfig(
    name="server2",
    transport="stdio",
    command="server2",
    tool_prefix="s2",  # Tools become: s2_read, s2_write, etc.
)

Common MCP Servers¶

Filesystem Server¶

MCPServerConfig(
    name="filesystem",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"],
    tool_prefix="fs",
)

GitHub Server¶

MCPServerConfig(
    name="github",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-github"],
    env_vars={"GITHUB_TOKEN": "${GITHUB_TOKEN}"},  # Pass token to subprocess
    tool_prefix="gh",
)

Brave Search Server¶

MCPServerConfig(
    name="search",
    transport="stdio",
    command="npx",
    args=["-y", "@modelcontextprotocol/server-brave-search"],
    env_vars={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},  # Pass API key to subprocess
    tool_prefix="search",
)

Configuration Reference¶

MCPServerConfig¶

Option	Type	Description
`name`	str	Unique server identifier
`transport`	str	`"stdio"`, `"sse"`, or `"streamable_http"`
`command`	str	Command to run (stdio only)
`args`	list	Command arguments (stdio only)
`url`	str	Server URL (SSE and Streamable HTTP)
`auth`	MCPAuthConfig	Authentication config
`tool_prefix`	str	Prefix for tool names
`env_file`	str	Path to .env file for environment
`env_vars`	dict	Environment variables for server
`timeout`	float	Connection timeout in seconds (default: 30)
`read_timeout`	float	Read timeout in seconds (default: 300)

MCPAuthConfig¶

Option	Type	Description
`type`	str	Auth type (`"api_key"`)
`key`	str	Direct API key
`key_env`	str	Environment variable name
`header`	str	HTTP header (default: `"Authorization"`)

Error Handling¶

Configuration Errors¶

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

# File-based config errors
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 server config: {e}")

# Programmatic config errors
configs = [
    MCPServerConfig(
        name="my-server",
        transport="stdio",
        command="my-mcp-server",
    ),
]

try:
    manager = MCPClientManager(configs)
    toolsets = manager.as_toolsets()
    agent = Agent("gpt-4o", toolsets=toolsets)
    result = await agent.run("Use the tools")
except ValueError as e:
    print(f"Configuration error: {e}")
except Exception as e:
    print(f"Runtime error: {e}")

Error Hierarchy¶

Exception	Description
`MCPConfigError`	Base exception for all MCP config errors
`MCPFileNotFoundError`	`.mcp.json` file not found
`MCPFileParseError`	Invalid JSON in config file
`MCPServerValidationError`	Invalid server entry in config
`MCPConnectionError`	Server connection failure (v0.1.3+)
`MCPConnectionTimeoutError`	Connection timeout (v0.1.3+)
`MCPServerNotFoundError`	Server not found in manager (v0.1.3+)

Best Practices¶

1. Use `as_toolsets()` (Recommended)¶

# Good - pydantic-ai handles lifecycle
manager = MCPClientManager(configs)
agent = Agent("gpt-4o", toolsets=manager.as_toolsets())
result = await agent.run("Use tools")

2. Use Tool Prefixes¶

# Prevent name conflicts between servers
server1 = MCPServerConfig(name="s1", tool_prefix="s1", ...)
server2 = MCPServerConfig(name="s2", tool_prefix="s2", ...)

3. Use Environment Variables for Secrets¶

# Don't hardcode API keys
auth = MCPAuthConfig(key_env="MY_API_KEY")  # Good
auth = MCPAuthConfig(key="sk-...")  # Avoid in production

4. Validate Configurations Early¶

# Check for configuration errors before runtime
manager = MCPClientManager(configs)
try:
    toolsets = manager.as_toolsets()
except ValueError as e:
    print(f"Invalid config: {e}")

Next Steps¶

Model Backends - Connect to local models
MCPClientManager API - Full reference
MCPServerConfig API - Configuration reference
load_mcp_json API - File loader reference
MCP Errors API - Error classes reference

MCP Integration¶

Overview¶

Quick Start¶

Loading from .mcp.json Files¶

Basic Usage¶

File Format¶

Extended Fields¶

Loading Multiple Files¶

Combining with Programmatic Config¶

Error Handling¶

Using load_mcp_json Directly¶

Transport Types¶

Stdio Transport¶

SSE Transport¶

Streamable HTTP Transport (v0.1.3+)¶

Transport Auto-Detection¶

Authentication¶

API Key Authentication¶

Using Environment Variables¶

Connection Testing (v0.1.3+)¶

Test Individual Server¶

Test All Servers¶

MCPConnectionResult¶

MCPToolInfo¶

MCPClientManager¶

Recommended Usage¶

Adding Servers Dynamically¶

Tool Prefixing¶

Common MCP Servers¶

Filesystem Server¶

GitHub Server¶

Brave Search Server¶

Configuration Reference¶

MCPServerConfig¶

MCPAuthConfig¶

Error Handling¶

Configuration Errors¶

Error Hierarchy¶

Best Practices¶

1. Use as_toolsets() (Recommended)¶

2. Use Tool Prefixes¶

3. Use Environment Variables for Secrets¶

4. Validate Configurations Early¶

Next Steps¶

1. Use `as_toolsets()` (Recommended)¶