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: Streamable HTTP, SSE, 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. If two servers expose a tool with the same name, the first server’s tool is kept and a warning is logged.

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, or mcp_oauth
    mcp_headers={"X-Tenant": "acme"},
    mcp_transport="streamable_http",
    mcp_timeout=30,
)

mcp_api_key, mcp_authorization_token, and mcp_oauth only fill in servers that have no credential of their own. mcp_headers is merged into each server’s headers, with the server’s own winning. mcp_transport and mcp_timeout replace every server’s own value.

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. This works for api_key, authorization_token, the OAuth credential fields, and stdio env values; headers are sent as written:

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

Transport

Every connection uses streamable HTTP unless you say otherwise; the URL alone does not switch it. Set the transport for an SSE server:

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

Valid values are streamable_http, sse, stdio, and auto. With auto, a URL ending in /sse uses SSE and a value with no scheme, such as "python server.py", uses stdio. A connection with command set always uses stdio:

MCPConnection(command="python", args=["server.py"], env={"API_KEY": "env:API_KEY"})

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. See MCPManager for every field.

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",
}

result is a str when the server returned text, which is how most servers send dicts and numbers, so it may hold JSON. It is a dict when the server returned structured content only, or text mixed with images or resources. Check the type before calling json.loads(); see What result holds.

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.

Serving your own agents over MCP

The integration also runs the other way. MCPDeployer turns an agent, any swarm with a run() method, or a plain function into an MCP server with an auth layer in front of it. Each target becomes one tool that takes task and an optional img, so any MCP client, including another swarms agent, can call it.

from swarms import Agent, MCPDeployer

analyst = Agent(
    agent_name="Market-Analyst",
    agent_description="Answers questions about markets and companies.",
    model_name="gpt-5.4",
    max_loops=1,
    print_on=False,
)

# Serves http://127.0.0.1:8000/mcp with one tool, market_analyst.
MCPDeployer(analyst, api_keys=["sk-local-dev"], port=8000).run()

Another agent connects with the same MCPConnection used for any other server:

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

client = Agent(
    agent_name="Portfolio-Manager",
    model_name="gpt-5.4",
    mcp_url=MCPConnection(url="http://127.0.0.1:8000/mcp", api_key="sk-local-dev"),
    max_loops=2,
)

Several agents, one server. Pass a list, or a dict of tool name to target: MCPDeployer({"research": researcher, "write": pipeline}, api_keys=[...]).
Auth. Static api_keys, keys from an environment variable with api_key_env, your own auth callable, or an mcp TokenVerifier with required_scopes. A server with no auth configured refuses to build unless you pass allow_anonymous=True.
Transports. Streamable HTTP (default), SSE, or stdio for desktop MCP hosts.
Requirements. swarms 15.0.3 or later and mcp>=2.0.0.

Serve an agent over MCP

One agent behind an API key, called by a second agent.

MCPDeployer reference

Every parameter, auth mode, transport, and method.

Error Handling

Connection and discovery failures raise the agent MCP exceptions. Tool discovery retries with exponential backoff up to retry_attempts (default 3) before raising; tool calls are not retried:

from swarms.schemas.agent_mcp_errors import (
    AgentMCPConnectionError,
    AgentMCPError,
)
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 AgentMCPError as e:
    print(f"Any other MCP failure: {e}")

AgentMCPToolError is raised only by call_tool(), when no configured server exposes the tool name. Per-call failures do not raise — check the envelope instead. A failure reported by the server carries its text in result; a client-side failure carries it in error:

for result in manager.execute_tool_calls(response):
    if result["is_error"]:
        print(f"{result['tool']} failed: {result.get('error') or result.get('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

Reuse the manager

Tool schemas are cached per manager, so build one and reuse it

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

connection = MCPConnection(
    url="https://slow-server.com/mcp",
    timeout=60,         # connecting and sending requests
    tool_timeout=300,   # how long one tool call may run; default 120
)

Authentication Failures

# Pass the raw token; swarms adds the "Bearer " prefix
connection = MCPConnection(
    url="https://api.example.com/mcp",
    authorization_token="your-token",
)

With mcp 2.x a rejected credential shows up as Server returned an error response, not 401. Check the server’s logs for the reason. 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

Next Steps

MCP tutorials

Step-by-step tutorials against real servers: DeepWiki, Exa, Firecrawl, Hugging Face, Semgrep

MCPManager

The full client reference: connections, results, errors, OAuth

MCPDeployer

Serve your own agents and swarms as MCP servers

Model Providers

Configure different LLM providers

Custom Tools

Create your own tool integrations

​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

​Serving your own agents over MCP

Serve an agent over MCP

MCPDeployer reference

​Error Handling

​Inspecting Configuration

​Caching

​Best Practices

Reuse the manager

Timeout Configuration

Error Recovery

Verbose Logging

​Troubleshooting

​Common Issues

​Next Steps

MCP tutorials

MCPManager

MCPDeployer

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

Serving your own agents over MCP

Error Handling

Inspecting Configuration

Caching

Best Practices

Troubleshooting

Common Issues

Next Steps