MCPManager

Overview

MCPManager is the one class handling the client side of MCP in Swarms. Point it at one or more servers and it takes care of transport selection, authentication, tool discovery, schema caching, and routing each tool call to the server that owns it. Every Agent holds one as agent.mcp_manager. It is enabled when you set mcp_url, mcp_urls, mcp_config, or mcp_configs on the agent. You can also use it directly, with no agent involved.

from swarms.tools.mcp_manager import MCPManager

manager = MCPManager(mcp_url="http://localhost:8000/mcp")

manager.list_tool_names()                                  # what's available
manager.get_tools()                                        # OpenAI schemas for an LLM
manager.call_tool("get_crypto_price", {"coin_id": "btc"})  # call one directly
manager.execute_tool_calls(llm_response)                   # run what a model asked for

For the server side, turning an agent or swarm into an MCP server, see MCPDeployer.

swarms.tools.mcp_client_tools and its standalone functions have been removed. See Migration for direct replacements.

MCPManager works with mcp 1.28.1 and later, including the 2.x line, from swarms 15.0.1 onward. On 15.0.0 with mcp 2.x, failed tool calls were reported as successes; upgrade swarms.

Import

from swarms.tools.mcp_manager import MCPManager

MCPManager is also exported from swarms.tools.

Constructor

Settings passed here are defaults for every server. Credentials and OAuth only fill in servers that do not set their own; transport and timeout replace each server’s own value.

Union[str, MCPConnection, Dict]

A single server — URL string, connection object, or dict.

List[Union[str, MCPConnection, Dict]]

Several servers. Entries may mix forms, so different servers can use different authentication. Two entries with the same URL (or the same stdio command and args) are merged into one.

Union[MCPConnection, Dict]

Full configuration for one server.

List[Union[MCPConnection, Dict]]

Full configuration for several servers.

str

API key applied to every server that does not define its own. Sent as Authorization: Bearer <key> unless the connection changes api_key_header or api_key_prefix.

str

Bearer token applied to every server that does not define its own. Sent as Authorization: Bearer <token> unless an API key already set the Authorization header.

Union[MCPOAuthConfig, Dict]

OAuth 2.1 configuration applied to every server that does not define its own.

Dict[str, str]

Extra headers merged into every request. A server’s own headers win on conflicts.

str

default:"None"

Force streamable_http, sse, stdio, or auto on every server. Hyphenated forms like streamable-http are normalized. When unset, each connection keeps its own transport, which defaults to streamable_http. See Transport.

int

default:"None"

HTTP timeout in seconds, applied to every configured connection and replacing its own value. Left unset by default, so each connection keeps MCPConnection.timeout, which defaults to 30. It does not bound how long a tool may run; that is MCPConnection.tool_timeout.

str

default:"agent"

Name used in log messages.

bool

default:"false"

Log each discovery and tool call.

int

default:"3"

Attempts at listing a server’s tools before giving up, with a 1, 2, 4, … second backoff between them. Values below 1 become 1. Tool calls are not retried. An Agent passes its tool_retry_attempts here.

MCPConnection

swarms.schemas.mcp_schemas.MCPConnection configures one server. Pass it as mcp_url, mcp_config, or an entry in mcp_urls / mcp_configs, on the manager or on an Agent. A dict with the same keys works too.

Field	Default	Purpose
`url`	`"http://localhost:8000/mcp"`	Server endpoint
`name`	`None`	Label used in logs and in each result’s `server` field. Falls back to `url`, then `command`
`transport`	`"streamable_http"`	`streamable_http`, `sse`, `stdio`, or `auto`
`api_key`	`None`	Key sent in `api_key_header` with `api_key_prefix`
`api_key_header`	`"Authorization"`	Header the key is sent in, for example `"x-api-key"`
`api_key_prefix`	`"Bearer"`	Prefix before the key. `None` or `""` sends the raw key
`authorization_token`	`None`	Bearer token, sent as `Authorization: Bearer <token>`
`auth_type`	`None`	`none`, `api_key`, `bearer`, `oauth`, or `custom`. Inferred when unset: `oauth` if `oauth` is set, then `api_key`, `bearer`, `custom` (headers only), `none`
`oauth`	`None`	An `MCPOAuthConfig`. When the auth type is `oauth`, `api_key` and `authorization_token` are not sent
`headers`	`None`	Extra headers for every request
`timeout`	`30`	HTTP timeout in seconds
`sse_read_timeout`	`300`	Seconds to wait for streamed data before giving up
`tool_timeout`	`120`	Seconds one tool call may run before the client gives up
`command`, `args`, `env`	`None`	Launch a local stdio server. Setting `command` selects the stdio transport
`tool_configurations`	`None`	Free-form settings dict, stored with the connection
`type`	`"mcp"`	Connection type label

api_key, authorization_token, the OAuth access_token, client_id, client_secret and token_url, and stdio env values accept "env:NAME" or "${NAME}" and are read from the environment when the connection is made. headers values are sent as written.

Transport

Connections default to streamable_http, whatever the URL looks like. Set transport="sse" for an SSE server, or transport="auto" to choose from the URL: a URL ending in /sse uses SSE and anything else uses streamable HTTP. A URL without http:// or https:// is rejected, never run as a command. Only a connection with command uses stdio. The stdio server process does not inherit your environment: it gets HOME, PATH, SHELL, TERM, USER and LOGNAME, plus the variables in env, so pass any key the server needs there.

from swarms.schemas.mcp_schemas import MCPConnection

MCPConnection(url="http://localhost:8000/sse", transport="sse")
MCPConnection(command="python", args=["server.py"], env={"API_KEY": "env:API_KEY"})

Discovery

get_tools

def get_tools(format="openai", force_refresh=False) -> List[Dict[str, Any]]

Fetch tools from every configured server, in parallel. Returns OpenAI function-calling schemas by default, ready to hand to an LLM. Results are cached; pass force_refresh=True to re-fetch.

tools = manager.get_tools()                    # OpenAI schemas, cached
raw = manager.get_tools(format="mcp")          # mcp.types.Tool objects, not cached
fresh = manager.get_tools(force_refresh=True)  # bypass the cache

When two servers expose a tool with the same name, the first server’s tool is kept and a warning is logged. A server that fails is logged and skipped; AgentMCPConnectionError is raised only when no tools could be loaded and at least one server failed. Async form: await manager.aget_tools(...).

Only the default "openai" format fills the cache and the tool-to-server routing table. call_tool and execute_tool_calls fetch the OpenAI schemas themselves when that table is empty, so routing works whichever format you listed with.

list_tool_names

def list_tool_names() -> List[str]

Every tool name across all configured servers, fetching once if needed — the cheapest way to see what is available.

Execution

call_tool

def call_tool(name: str, arguments: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Call one tool directly, no LLM involved. Returns a result envelope:

{
    "tool": "get_crypto_price",
    "server": "http://localhost:8000/mcp",
    "arguments": {"coin_id": "bitcoin"},
    "is_error": False,
    "result": "Current price of Bitcoin: $64,601.00",
}

is_error is True when the server reports the tool failed; result then holds the server’s error text. When the call itself fails on the client side, the envelope has an error string instead of result. With several servers, an unknown name raises AgentMCPToolError. With one server, every name is sent to it and the server reports unknown tools. A server that cannot be reached raises AgentMCPConnectionError. Async form: await manager.acall_tool(...).

execute_tool_calls

def execute_tool_calls(response, output_type="dict") -> Union[List[Dict], str]

Run the tool calls contained in an LLM response. Each call is routed to the server that advertised the tool; calls to one server share a session, and different servers run in parallel. Results come back in call order. This is the step an Agent performs between LLM turns.

Any

required

An LLM response: a JSON string, one tool call, a list of tool calls, or a whole assistant message with tool_calls, as dicts or SDK objects.

"dict" | "json" | "str"

default:"dict"

Result format: the list of envelopes, that list as a JSON string, or one tool @ server -> result line per call.

Nothing raises per call. An unknown tool name or a failed session becomes an envelope with is_error: True and an error string, so one bad call cannot abort a batch. A response with no tool calls returns [] ("[]" for json). With no servers configured, it raises AgentMCPConnectionError. Async form: await manager.aexecute_tool_calls(...). To re-render results you already hold, use the static MCPManager.format_results(results, output_type).

What `result` holds

result is a native Python value; swarms never encodes it:

The server returned	`result` is
Text blocks only	A `str`, joined with newlines when there are several
Text plus other blocks (images, resources)	`{"text": "...", "content": [...]}`
Only structured content	That `dict`

Most servers, MCPDeployer included, send dicts and numbers as JSON text, so a str result may itself be JSON. Check the type before parsing:

import json

payload = results[0]["result"]
if isinstance(payload, str):
    try:
        payload = json.loads(payload)
    except json.JSONDecodeError:
        pass

Configuration management

Member	Description
`connections`	The normalized `MCPConnection` list
`enabled`	Property — `True` when at least one server is configured. `bool(manager)` and `len(manager)` follow it
`add_server(server)`	Register another server (URL, dict, or `MCPConnection`) and invalidate the tool cache. A duplicate is ignored
`label(connection)`	The name used in logs and result envelopes
`clear_cache()`	Drop cached tool schemas and routing information
`clear_auth_cache()`	Forget in-process OAuth providers and cached client-credentials tokens
`to_dict()`	Serializable, secret-redacted view of the configuration

Authentication

from swarms.schemas.mcp_schemas import MCPConnection, MCPOAuthConfig
from swarms.tools.mcp_manager import MCPManager

# API key, sent as Authorization: Bearer sk-...
MCPManager(mcp_url="https://api.example.com/mcp", api_key="sk-...")

# Bearer token
MCPManager(mcp_url="https://api.example.com/mcp", authorization_token="ey...")

# Raw key in a custom header
MCPManager(mcp_config=MCPConnection(
    url="https://api.example.com/mcp",
    api_key="env:EXAMPLE_MCP_KEY",
    api_key_header="x-api-key",
    api_key_prefix=None,
))

# Headers, transport, timeouts
MCPManager(mcp_config=MCPConnection(
    url="https://api.example.com/mcp",
    headers={"X-Tenant": "acme"},
    transport="streamable_http",
    timeout=20,
    tool_timeout=300,
))

# OAuth 2.1 client credentials
MCPManager(mcp_config=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",
)))

OAuth 2.1

MCPOAuthConfig supports three modes:

Authorization code (grant_type="authorization_code", the default). The interactive browser flow with PKCE, run by the mcp SDK, which discovers the endpoints and registers the client dynamically (RFC 7591). A loopback listener on redirect_uri (default http://127.0.0.1:8765/callback) catches the redirect, and tokens are cached in ~/.swarms/mcp_auth/<server>.json with 0600 permissions, so the browser step runs once.
Client credentials (grant_type="client_credentials"). Headless. Needs client_id, and usually client_secret. The token endpoint is discovered from the server’s well-known metadata unless token_url is set, and a new token is fetched shortly before the old one expires.
Pre-issued token (access_token="..."). No flow runs; the token is sent as a bearer credential.

Other fields: scopes, client_name (default "Swarms Agent") and client_uri for registration, token_storage_path, use_token_cache (default True; False keeps tokens in memory only), open_browser (default True; False logs the URL instead), and callback_timeout (default 300 seconds). MCPFileTokenStorage, exported from swarms.tools, is the on-disk token cache. Call .clear() on one to force a fresh authorization.

Mixed auth across servers

MCPManager(mcp_urls=[
    "http://localhost:8000/mcp",                    # local, no auth
    MCPConnection(url="https://api.example.com/mcp", api_key="sk-..."),
])

Errors

Exception	Raised when
`AgentMCPConnectionError`	A server cannot be reached or rejects the credentials, tool discovery fails on every server, an OAuth flow fails, or a connection is misconfigured
`AgentMCPToolError`	`call_tool()` names a tool that no configured server exposes, with more than one server configured
`AgentMCPError`	Base class for both

All live in swarms.schemas.agent_mcp_errors. Tool discovery retries up to retry_attempts times before raising. Failures during execute_tool_calls are never raised; they come back as is_error: True envelopes. call_tool also returns an envelope when the tool itself fails.

With mcp 2.x, an HTTP error from the server, including 401, reaches the client as Server returned an error response; the status code is not available. Check the server’s own logs to see why a request was refused.

Migration

The standalone functions in swarms.tools.mcp_client_tools were removed once everything moved onto this class.

Removed function	Replacement
`aget_mcp_tools(server_path=URL)`	`await MCPManager(mcp_url=URL).aget_tools()`
`get_mcp_tools_sync(server_path=URL)`	`MCPManager(mcp_url=URL).get_tools()`
`get_tools_for_multiple_mcp_servers(urls=URLS)`	`MCPManager(mcp_urls=URLS).get_tools()`
`execute_tool_call_simple(response=R, server_path=URL)`	`await MCPManager(mcp_url=URL).aexecute_tool_calls(R)`
`execute_multiple_tools_on_multiple_mcp_servers(...)`	`await MCPManager(mcp_urls=URLS).aexecute_tool_calls(R)`
`MCPError`, `MCPConnectionError`, `MCPToolError`, `MCPExecutionError`	`AgentMCPError`, `AgentMCPConnectionError`, `AgentMCPToolError`

Two behavioral differences to be aware of:

output_type on tool fetching is gone. get_tools_for_multiple_mcp_servers accepted the argument but never applied it. Use format="openai" | "mcp" to pick the schema shape.
Execution results are wrapped. The old functions returned the raw MCP CallToolResult dump. execute_tool_calls returns one envelope per call, so results from several servers stay attributable. Read result["result"] for the tool’s own payload.

MCP Integration

Using MCP servers from an agent

Agent

Set mcp_url / mcp_urls to wire this in automatically

MCPDeployer

Serve an agent or swarm as an MCP server

Authentication patterns

Every credential shape, with code

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Overview

Import

Constructor

MCPConnection

Transport

Discovery

get_tools

list_tool_names

Execution

call_tool

execute_tool_calls

What `result` holds

Configuration management

Authentication

OAuth 2.1

Mixed auth across servers

Errors

Migration

MCP Integration

Agent

MCPDeployer

Authentication patterns

​Overview

​Import

​Constructor

​MCPConnection

​Transport

​Discovery

​get_tools

​list_tool_names

​Execution

​call_tool

​execute_tool_calls

​What result holds

​Configuration management

​Authentication

​OAuth 2.1

​Mixed auth across servers

​Errors

​Migration

​Related

MCP Integration

Agent

MCPDeployer

Authentication patterns

Overview

Import

Constructor

MCPConnection

Transport

Discovery

get_tools

list_tool_names

Execution

call_tool

execute_tool_calls

What `result` holds

Configuration management

Authentication

OAuth 2.1

Mixed auth across servers

Errors

Migration

Related