Model Context Protocol (MCP)

The Model Context Protocol (MCP) is a standardized protocol that enables AI agents to interact with external tools and services through MCP servers. Swarms provides first-class support for MCP integration, allowing your agents to dynamically discover and execute tools.

What is MCP?

MCP (Model Context Protocol) provides:

Standardized Tool Interface: Unified protocol for tool integration
Dynamic Discovery: Automatically discover available tools from MCP servers
Multi-Server Support: Connect to multiple MCP servers simultaneously
Type Safety: Automatic schema validation for tool calls
Flexible Transport: Support for HTTP, WebSocket, and stdio transports

Quick Start

Let the agent do it

The simplest integration: give the agent a URL and it discovers and calls the tools itself.

from swarms import Agent

agent = Agent(
    agent_name="MCP-Agent",
    model_name="claude-sonnet-4-6",
    mcp_url="https://mcp.deepwiki.com/mcp",   # free, no API key
    max_loops=1,
)

result = agent.run("What is the swarms framework? Use the deepwiki tools.")

That is the whole integration. Behind it, the agent builds an MCPManager — reachable as agent.mcp_manager — which handles transport, auth, discovery, and routing.

Several servers at once

agent = Agent(
    agent_name="Multi-MCP-Agent",
    model_name="claude-sonnet-4-6",
    mcp_urls=[
        "https://mcp.deepwiki.com/mcp",
        "https://learn.microsoft.com/api/mcp",
    ],
    max_loops=1,
)

The agent sees the union of every server’s tools and each call is routed back to the server that owns it.

Without an agent

MCPManager works standalone when you want tools, not autonomy:

from swarms.tools.mcp_manager import MCPManager

manager = MCPManager(mcp_url="https://api.example.com/mcp")

manager.list_tool_names()                       # ['get_weather', 'send_email']
tools = manager.get_tools()                     # OpenAI schemas
result = manager.call_tool("get_weather", {"location": "San Francisco"})

Every method has an async twin: aget_tools, acall_tool, aexecute_tool_calls.

Connection Configuration

Agent-level settings

Authentication and transport can be set directly on the agent and apply to every server it uses:

agent = Agent(
    agent_name="Secure-MCP-Agent",
    model_name="claude-sonnet-4-6",
    mcp_url="https://api.example.com/mcp",
    mcp_api_key="sk-...",                    # or mcp_authorization_token
    mcp_headers={"X-Tenant": "acme"},
    mcp_transport="streamable_http",
    mcp_timeout=30,
)

Per-server settings with MCPConnection

For different credentials per server, pass MCPConnection objects:

from swarms import Agent
from swarms.schemas.mcp_schemas import MCPConnection

agent = Agent(
    agent_name="Mixed-Auth-Agent",
    model_name="claude-sonnet-4-6",
    mcp_urls=[
        "http://localhost:8000/mcp",                     # local, no auth
        MCPConnection(
            url="https://api.example.com/mcp",
            authorization_token="your-token",
            transport="streamable_http",
            timeout=30,
        ),
    ],
)

Secrets from the environment

Keep keys out of source — both env:NAME and ${NAME} are resolved when the connection is made:

MCPConnection(url="https://api.example.com/mcp", api_key="env:EXAMPLE_MCP_KEY")

Transport

Transport is auto-detected from the URL. Force it when you need to:

MCPManager(mcp_url="https://api.example.com/mcp", transport="sse")

Valid values are streamable_http, sse, and stdio. Hyphenated forms such as streamable-http are normalized automatically.

OAuth 2.1

from swarms.schemas.mcp_schemas import MCPConnection, MCPOAuthConfig

connection = MCPConnection(
    url="https://api.example.com/mcp",
    oauth=MCPOAuthConfig(
        grant_type="client_credentials",
        client_id="example-client",
        client_secret="env:EXAMPLE_CLIENT_SECRET",
        token_url="https://api.example.com/oauth/token",
        scopes=["tools.read"],
    ),
)

Full OAuth 2.1 is supported, including PKCE authorization-code flow with RFC 7591 dynamic client registration, headless client credentials, and pre-issued tokens.

Multi-Server Integration

One manager, many servers, automatic routing:

from swarms.tools.mcp_manager import MCPManager

manager = MCPManager(mcp_urls=[
    "https://api.example.com/mcp",
    "https://tools.example.com/mcp",
])

# The union of every server's tools
print(manager.list_tool_names())

# Calls go to whichever server advertised the tool — the call site is identical
manager.call_tool("database_query", {"query": "SELECT * FROM users"})
manager.call_tool("send_email", {"to": "user@example.com", "subject": "Hello"})

Servers can be added later; doing so invalidates the tool cache so the next fetch picks them up:

manager.add_server("https://services.example.com/mcp")

Tool Execution

Executing what a model asked for

When an LLM replies with tool calls, hand the response straight to the manager. Each call is routed and the results come back in order — this is the step an Agent performs between turns.

response = {
    "tool_calls": [
        {"function": {"name": "get_weather", "arguments": {"location": "SF"}}},
        {"function": {"name": "send_email", "arguments": {"to": "a@b.com"}}},
    ]
}

results = manager.execute_tool_calls(response)                     # list of dicts
as_json = manager.execute_tool_calls(response, output_type="json") # JSON string
as_text = manager.execute_tool_calls(response, output_type="str")  # plain text

Each result is an envelope:

{
    "tool": "get_weather",
    "server": "https://api.example.com/mcp",
    "arguments": {"location": "SF"},
    "is_error": False,
    "result": "18°C, partly cloudy",
}

When a tool returns structured data, its payload arrives as a JSON string in result:

import json
payload = json.loads(results[0]["result"])

Calling one tool directly

result = manager.call_tool("get_weather", {"location": "San Francisco"})

Async

import asyncio

async def main():
    manager = MCPManager(mcp_urls=["https://api.example.com/mcp"])
    return await manager.aexecute_tool_calls(response, output_type="dict")

results = asyncio.run(main())

Real-World Example

from swarms import Agent
from swarms.schemas.mcp_schemas import MCPConnection

agent = Agent(
    agent_name="MCP-Enabled-Agent",
    system_prompt="You are an AI assistant with access to tools via MCP.",
    model_name="claude-sonnet-4-6",
    max_loops=3,
    mcp_config=MCPConnection(
        url="https://mcp.example.com/api",
        transport="streamable_http",
        authorization_token="your-mcp-api-token",
        timeout=30,
    ),
    streaming_on=True,
)

result = agent.run(
    "Use the available tools to analyze the database and send a summary email"
)

The agent discovers the server’s tools on startup, decides which to call, executes them, and feeds results back into its own loop.

Error Handling

Failures raise the agent MCP exceptions, and operations retry with exponential backoff up to retry_attempts (default 3) before raising:

from swarms.schemas.agent_mcp_errors import (
    AgentMCPConnectionError,
    AgentMCPError,
    AgentMCPToolError,
)
from swarms.tools.mcp_manager import MCPManager

try:
    tools = MCPManager(
        mcp_url="https://api.example.com/mcp",
        timeout=120,          # slow servers
        retry_attempts=5,
    ).get_tools()
except AgentMCPConnectionError as e:
    print(f"Could not reach the server or authentication failed: {e}")
except AgentMCPToolError as e:
    print(f"A tool call failed on the server: {e}")
except AgentMCPError as e:
    print(f"Any other MCP failure: {e}")

Per-result failures do not raise — check the envelope instead:

for result in manager.execute_tool_calls(response):
    if result["is_error"]:
        print(f"{result['tool']} failed: {result['result']}")

Inspecting Configuration

to_dict() gives a serializable, secret-redacted view — safe to log:

manager.to_dict()
# {'enabled': True,
#  'servers': [{'name': 'https://api.example.com/mcp',
#               'url': 'https://api.example.com/mcp',
#               'transport': 'streamable_http',
#               'auth_type': 'api_key',
#               'timeout': 30}]}

Caching

Tool schemas are cached per manager after the first fetch:

manager.get_tools()                      # fetches
manager.get_tools()                      # cached
manager.get_tools(force_refresh=True)    # re-fetches
manager.clear_cache()                    # drop schemas and routing
manager.clear_auth_cache()               # forget OAuth providers and tokens

Build one manager and reuse it rather than constructing one per call.

Best Practices

Connection Pooling

Reuse MCP connections when fetching tools multiple times

Timeout Configuration

Set appropriate timeouts based on server response times

Error Recovery

Implement fallback strategies for MCP server failures

Verbose Logging

Enable verbose mode during development for debugging

Troubleshooting

Common Issues

Connection Timeouts

# Increase timeout for slow servers
connection = MCPConnection(
    url="https://slow-server.com/mcp",
    timeout=120,  # 2 minutes
)

Authentication Failures

# Ensure authorization token is set
connection = MCPConnection(
    url="https://api.example.com/mcp",
    authorization_token="Bearer your-token",
)

Tool Not Found

# Verify what the server actually exposes
manager = MCPManager(mcp_url=url, verbose=True)
print(f"Available tools: {manager.list_tool_names()}")

Agent Not Using the Tools

# Confirm MCP is wired up and the agent can see the schemas
print(agent.mcp_enabled)                       # True when a server is configured
print(agent.mcp_manager.list_tool_names())     # what the agent will be offered

Getting Started

Core Concepts

Agent Development

Reasoning Agents

Multi-Agent Architectures

Deployment & Operations

Integrations

CLI Reference

Changelog

Model Context Protocol (MCP)

What is MCP?

Quick Start

Let the agent do it

Several servers at once

Without an agent

Connection Configuration

Agent-level settings

Per-server settings with MCPConnection

Secrets from the environment

Transport

OAuth 2.1

Multi-Server Integration

Tool Execution

Executing what a model asked for

Calling one tool directly

Async

Real-World Example

Error Handling

Inspecting Configuration

Caching

Best Practices

Connection Pooling

Timeout Configuration

Error Recovery

Verbose Logging

Troubleshooting

Common Issues

Next Steps

Model Providers

Custom Tools

​What is MCP?

​Quick Start

​Let the agent do it

​Several servers at once

​Without an agent

​Connection Configuration

​Agent-level settings

​Per-server settings with MCPConnection

​Secrets from the environment

​Transport

​OAuth 2.1

​Multi-Server Integration

​Tool Execution

​Executing what a model asked for

​Calling one tool directly

​Async

​Real-World Example

​Error Handling

​Inspecting Configuration

​Caching

​Best Practices

Connection Pooling

Timeout Configuration

Error Recovery

Verbose Logging

​Troubleshooting

​Common Issues

​Next Steps

Model Providers

Custom Tools

What is MCP?

Quick Start

Let the agent do it

Several servers at once

Without an agent

Connection Configuration

Agent-level settings

Per-server settings with MCPConnection

Secrets from the environment

Transport

OAuth 2.1

Multi-Server Integration

Tool Execution

Executing what a model asked for

Calling one tool directly

Async

Real-World Example

Error Handling

Inspecting Configuration

Caching

Best Practices

Troubleshooting

Common Issues

Next Steps