MCPDeployer - swarms

Overview

MCPDeployer is the server side of MCP in Swarms. MCPManager connects an agent to an MCP server; MCPDeployer turns an agent into one. Give it one target or several. A target is an Agent, any swarm with a run() method (SequentialWorkflow, SwarmRouter, HierarchicalSwarm, …), or a plain Python callable. Each target becomes one MCP tool. Every HTTP request passes through an auth layer before it reaches the MCP transport, and a server with no auth configured refuses to build.

from swarms import Agent, MCPDeployer

analyst = Agent(
    agent_name="Market-Analyst",
    agent_description="Answers questions about markets and companies.",
    model_name="claude-sonnet-5",
    max_loops=1,
)

MCPDeployer(analyst, api_keys=["sk-local-dev"], port=8000).run()

Another agent then uses it like any other MCP server:

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

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

MCPDeployer ships in swarms 15.0.3 and needs mcp>=2.0.0. It builds on mcp.server.mcpserver.MCPServer, which the 1.x releases of mcp do not have.

pip install -U swarms "mcp>=2.0.0"

Import

from swarms import MCPDeployer, deploy_as_mcp

How a target becomes a tool


Tool name	The dict key if you passed a dict, otherwise a snake_case form of the target’s `agent_name`, `name`, or `__name__`. `Market-Analyst` becomes `market_analyst`.
Tool description	The target’s `agent_description`, then `description`, then the first line of its docstring, then `Run <tool name> on a task.` An `Agent` always has an `agent_description`, and the default one is generic, so set it to something the calling model can choose by.
Input schema	Always `task: str` (required) and `img: str \| None` (optional).
How it is called	`target.run(task, img=img)` when `img` is sent and `run` accepts it, otherwise `target.run(task)`. A plain callable is called as `target(task)`, and `img` is ignored.
Result	A string is returned as is. `None` becomes `""`. Anything else is serialized with `json.dumps(..., indent=2)`, falling back to `str()`. The client always receives one text block.

The call runs on a worker thread, so a blocking agent does not stall the server. Set timeout to cap how long one call may take; see Timeouts.

What a client sees

A client that lists the tools of a server serving word_count gets this schema, shown here as MCPManager.get_tools() returns it:

{
  "type": "function",
  "function": {
    "name": "word_count",
    "description": "Count the words in the text.",
    "parameters": {
      "type": "object",
      "title": "word_countArguments",
      "properties": {
        "task": {"title": "Task", "type": "string"},
        "img": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null, "title": "Img"}
      },
      "required": ["task"]
    },
    "strict": false
  }
}

The server also sends MCP instructions on connect: one name: description line per target. Functions passed in extra_tools keep their own signatures as their schemas and are listed before the targets. tool_names, tools, and /health cover the targets only.

Serving several targets

Pass a list to name each tool after its target, or a dict to choose the names yourself. Each entry becomes its own tool on the same server, behind the same auth.

from swarms import Agent, MCPDeployer

researcher = Agent(agent_name="Researcher", model_name="gpt-5.4", max_loops=1, print_on=False)
critic = Agent(agent_name="Critic", model_name="gpt-5.4", max_loops=1, print_on=False)

# Tools: researcher, critic
by_target = MCPDeployer([researcher, critic], api_keys=["sk-local-dev"])

# Tools: research, critique
by_key = MCPDeployer({"research": researcher, "critique": critic}, api_keys=["sk-local-dev"])

by_key.run()

Two targets that resolve to the same name raise ValueError, so use a dict when names could collide. add_tool() registers more targets before the server starts.

Serving swarms

Anything with a run(task) method is served the same way as an agent: SequentialWorkflow, ConcurrentWorkflow, SwarmRouter, HierarchicalSwarm, and the rest. The tool name comes from the swarm’s name, and the description from its description.

from swarms import Agent, MCPDeployer, SequentialWorkflow

pipeline = SequentialWorkflow(
    name="Briefing-Pipeline",
    description="Researches a topic and writes a three-paragraph briefing.",
    agents=[
        Agent(agent_name="Researcher", model_name="gpt-5.4", max_loops=1, print_on=False),
        Agent(agent_name="Writer", model_name="gpt-5.4", max_loops=1, print_on=False),
    ],
    max_loops=1,
)

MCPDeployer(pipeline, tool_name="write_briefing", api_keys=["sk-local-dev"], timeout=300).run()

A swarm that returns a dict or list reaches the client as indented JSON text.

State between calls

One target instance serves every call, from every client. An Agent keeps its conversation between run() calls, so each task lands in the same history, and calls that arrive together run on separate threads against the same object. When calls must not share context, serve a function that builds a fresh agent per call:

from swarms import Agent, MCPDeployer


def research(task: str) -> str:
    """Answer a research question in one short paragraph."""
    agent = Agent(agent_name="Researcher", model_name="gpt-5.4", max_loops=1, print_on=False)
    return agent.run(task)


MCPDeployer(research, api_keys=["sk-local-dev"]).run()

Constructor

MCPDeployer(
    targets,
    name=None,
    description=None,
    tool_name=None,
    host="127.0.0.1",
    port=8000,
    transport="streamable-http",
    path=None,
    api_keys=None,
    api_key_env=None,
    api_key_header="x-api-key",
    auth=None,
    token_verifier=None,
    required_scopes=None,
    allow_anonymous=False,
    public_paths=None,
    extra_tools=None,
    timeout=None,
    json_response=False,
    stateless_http=True,
    verbose=False,
    show_banner=True,
)

Targets and naming

Any | List[Any] | Dict[str, Any]

required

What to serve. One target, a list of targets, or a dict of tool name to target. A target is an Agent, any object with a run(task, ...) method, or a callable taking the task string. Each becomes one tool.

Optional[str]

default:"None"

Server name advertised to MCP clients. Defaults to the first tool’s name.

Optional[str]

default:"None"

Tool name for a single target. With several targets, pass a dict instead. Passing it with a list or dict raises ValueError.

Optional[str]

default:"None"

Tool description for a single target. Same restriction as tool_name.

Optional[Iterable[Callable]]

default:"None"

More plain functions to expose beside the targets. Each function’s signature becomes its schema and its docstring its description, so give every one a docstring.

Transport

str

default:"127.0.0.1"

Bind address. Binding anywhere other than 127.0.0.1 or localhost turns off the mcp package’s DNS-rebinding guard, which would otherwise reject every request whose Host header is not localhost.

int

default:"8000"

Bind port.

str

default:"streamable-http"

"streamable-http", "sse", or "stdio". Anything else raises ValueError. Auth applies to the two HTTP transports only. A swarms client reaching an SSE server must set transport="sse" on its MCPConnection (or mcp_transport="sse" on the agent), because clients default to streamable HTTP.

Optional[str]

default:"None"

URL path of the MCP endpoint. Defaults to /mcp, or /sse for the SSE transport.

bool

default:"False"

Streamable HTTP only. Reply with plain JSON instead of an event stream.

bool

default:"True"

Streamable HTTP only. Keep no per-session state, which is what you want behind a load balancer.

Optional[float]

default:"None"

Seconds one tool call may run before it fails. None or 0 means no limit. See Timeouts.

Auth

Optional[Iterable[str]]

default:"None"

Static keys. Compared in constant time. Blank entries are dropped and duplicates removed.

Optional[str]

default:"None"

Name of an environment variable holding more keys, comma-separated. Read once, at construction.

str

default:"x-api-key"

Header a client may send a raw key in. A Bearer prefix in it is stripped. Authorization: Bearer <key> is always accepted as well.

Callable[[Optional[str], Headers], bool | dict | None]

default:"None"

Your own check, sync or async. It receives the credential (or None) and the request headers. A truthy return admits the request; a dict return is also kept as the request’s claims, with sub/subject and scopes read from it. A falsy return or an exception refuses the request. Takes precedence over every other auth setting.

Optional[TokenVerifier]

default:"None"

An mcp.server.auth.provider.TokenVerifier. Used when auth is not set. A token that fails verification, has expired, or lacks any of required_scopes is refused.

Optional[Iterable[str]]

default:"None"

Scopes a verified token must carry.

bool

default:"False"

Serve with no auth at all. Off by default: with no api_keys, api_key_env keys, auth, or token_verifier, the constructor raises ValueError.

Optional[Iterable[str]]

default:"('/health',)"

Paths that skip auth.

Output

bool

default:"False"

Log every admitted tool call.

bool

default:"True"

Print the startup banner from run() and start().

Auth, in order of precedence

authenticate() checks these in order and stops at the first that applies:

allow_anonymous=True admits everything.
auth decides alone. The static keys and token verifier are not consulted.
No credential in either header: refused.
token_verifier verifies the token, its expiry, and required_scopes. When a verifier is set, the static keys are not consulted.
The credential is compared against the static keys.

A refused request gets:

HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer

{"error": "unauthorized"}

The client can send its key either way:

curl -H "x-api-key: sk-local-dev" ...
curl -H "Authorization: Bearer sk-local-dev" ...

MCPConnection(api_key=...) and MCPManager(api_key=...) handle this for you.

Custom auth

from swarms import Agent, MCPDeployer

TENANTS = {"acme": "sk-acme", "globex": "sk-globex"}


async def check(credential, headers):
    tenant = headers.get("x-tenant")
    if not credential or TENANTS.get(tenant) != credential:
        return False
    return {"sub": tenant, "scopes": ["run"]}


agent = Agent(agent_name="Support", model_name="claude-sonnet-5", max_loops=1)

MCPDeployer(agent, auth=check).run()

A swarms client sends the extra header through MCPConnection(headers={"x-tenant": "acme"}, api_key="sk-acme").

Token verifier

Any object with an async verify_token(token) method that returns an mcp.server.auth.provider.AccessToken, or None, works as a verifier.

from mcp.server.auth.provider import AccessToken

from swarms import Agent, MCPDeployer


class StaticTokenVerifier:
    async def verify_token(self, token: str):
        if token == "tok-admin":
            return AccessToken(token=token, client_id="ops-team", scopes=["agent:run"])
        return None


agent = Agent(agent_name="Summariser", model_name="gpt-5.4", max_loops=1, print_on=False)

MCPDeployer(agent, token_verifier=StaticTokenVerifier(), required_scopes=["agent:run"]).run()

Who made the request

An admitted request carries an AuthResult in its ASGI scope, at scope["state"]["mcp_auth"]:

Field	Value
`method`	`"anonymous"`, `"custom"`, `"token"`, or `"api_key"`
`subject`	`subject` or `sub` from a custom `auth` dict, or the token’s subject or `client_id`
`scopes`	`scopes` from a custom `auth` dict, or the token’s scopes
`claims`	The full dict a custom `auth` returned, or the token’s claims

The module-level helper credential_from_headers(headers, api_key_header="x-api-key") in swarms.structs.mcp_deployer returns the credential exactly as the auth layer reads it, which is handy in tests.

Timeouts

timeout bounds one tool call on the server. When it expires, the client gets a tool result with is_error set instead of an answer, even if the target is still blocked inside a model call. The worker thread running the target is abandoned, not killed: Python cannot stop a thread, so the agent finishes its run in the background and its result is discarded. Budget model spend with that in mind. The client has its own limit. A swarms client waits tool_timeout seconds for one tool call (default 120, set on MCPConnection), so serving a slow swarm with timeout=300 also needs a client that waits that long:

from swarms.schemas.mcp_schemas import MCPConnection

MCPConnection(url="http://127.0.0.1:8000/mcp", api_key="sk-local-dev", tool_timeout=320)

stdio

Desktop MCP hosts launch servers as subprocesses and talk over stdin and stdout. There are no headers, so the host is the security boundary. The constructor still requires an auth setting, so pass allow_anonymous=True.

MCPDeployer(agent, transport="stdio", allow_anonymous=True, show_banner=False).run()

On stdio, stdout carries the protocol. Set show_banner=False, since the banner is printed to stdout, and keep print_on=False on served agents so nothing else is written there.

start() and build_app() need an HTTP transport and raise ValueError on stdio.

Methods

run

deployer.run(log_level="info")

Serve until interrupted. Blocks. For the HTTP transports this runs uvicorn; for stdio it hands stdin/stdout to the MCP server and, unless allow_anonymous=True, logs a warning that auth settings are ignored, since stdio carries no headers.

start

deployer.start(wait=10.0)

Serve on a background daemon thread and return once the socket is accepting connections. Returns the deployer. Raises RuntimeError if the server has not started within wait seconds (a port already in use ends up here), and ValueError for the stdio transport. Calling it again while running is a no-op. Uvicorn logs at warning level here, so per-request access lines are not printed; run() logs at log_level.

stop

deployer.stop(wait=10.0)

Stop a server started with start(), waiting up to wait seconds for the thread to finish.

A deployer serves once. On the streamable HTTP transport, calling start() again after stop() fails with RuntimeError: MCPDeployer did not start, because the underlying mcp session manager can only run once per instance. Build a new MCPDeployer to serve again.

Context manager

with MCPDeployer(...) as deployer: calls start() on entry and stop() on exit. This is the easiest way to run a server and a client in one script or test.

from swarms import MCPDeployer
from swarms.tools.mcp_manager import MCPManager


def word_count(task: str) -> int:
    """Count the words in the text."""
    return len(task.split())


with MCPDeployer(word_count, api_keys=["sk-local-dev"], port=8000) as deployer:
    client = MCPManager(mcp_url=deployer.url, api_key="sk-local-dev")
    print(client.list_tool_names())
    print(client.call_tool("word_count", {"task": "the quick brown fox"})["result"])

['word_count']
4

add_tool

deployer.add_tool(target, name=None, description=None)

Register one more target as a tool. Call it before run() or start(); after start() it raises RuntimeError. A name that is already registered raises ValueError. Returns the ServedTool.

authenticate

result = await deployer.authenticate(headers)

The check the auth layer runs on every request. Returns an AuthResult when the request is admitted and None when it is refused. Useful for testing an auth setup without starting a server.

build_app

app = deployer.build_app()

The ASGI app: the MCP transport wrapped in the auth layer. Serve it with your own ASGI server when run() and start() are not what you need, for example uvicorn.run(deployer.app, host="0.0.0.0", port=8000). Only HTTP requests are checked; lifespan events pass straight through, so the MCP session manager starts normally. If you mount it inside another app, that app must run this app’s lifespan. The app property builds it once and caches it. Raises ValueError for stdio.

print_banner

deployer.print_banner(console=None)

Print the startup banner: server name, tools, endpoint, transport, and auth mode. Pass a rich.console.Console to print somewhere other than stdout. run() and start() call it unless show_banner=False.

Properties

str

http://{host}:{port}{path}, the address to give a client. With host="0.0.0.0" this shows the bind address, so give remote clients the machine’s own hostname instead.

List[str]

Every registered tool name, in registration order.

Dict[str, ServedTool]

Tool name to ServedTool (name, target, description, target_type).

str

The first registered tool’s name.

Any

The first registered target.

str

The first registered tool’s description.

List[str]

The resolved static keys, from api_keys and api_key_env together.

mcp.server.mcpserver.MCPServer

The underlying mcp server, with every tool already registered.

ASGI app

The cached result of build_app().

Health check

GET /health is always public and returns:

{
  "status": "ok",
  "name": "market_analyst",
  "tool": "market_analyst",
  "tools": ["market_analyst"],
  "transport": "streamable-http"
}

deploy_as_mcp

from swarms import deploy_as_mcp

deploy_as_mcp(analyst, api_keys=["sk-local-dev"], port=8000)

Builds an MCPDeployer from target and the same keyword arguments, then calls run(). Blocks until the server stops, then returns the deployer. target takes the same forms as targets: one target, a list, or a dict.

Errors

Raised	When
`ValueError("MCPDeployer needs a target to serve.")`	`targets` is `None`
`StopIteration`	`targets` is an empty list or dict
`ValueError`	`transport` is not `streamable-http`, `sse`, or `stdio`
`ValueError("No auth configured. ...")`	No keys, `auth`, or `token_verifier`, and `allow_anonymous` is `False`. This applies to `stdio` too
`ValueError`	A target is neither callable nor has a `run()` method
`ValueError`	Two targets resolve to the same tool name. Pass a dict to name them
`ValueError`	`tool_name` or `description` passed with a list or dict
`ValueError`	`start()` or `build_app()` on the `stdio` transport
`RuntimeError`	`add_tool()` after `start()`, or the server did not start within `wait`
`ModuleNotFoundError` (`mcp.server.mcpserver`)	Construction with `mcp` 1.x installed. Upgrade to `mcp>=2.0.0`

Errors inside a tool call never reach the client as exceptions. A target that raises, or a call that passes timeout, comes back as a tool result with is_error set and the error text as its content.

Examples

Serve an agent over MCP

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

Serve a team from one server

Several agents, a workflow, and a function as separate tools.

MCPManager

The client side: how agents connect to servers like this one.

The framework repository has more, covering every auth mode and transport: examples/mcp/mcp_deployer/.

Source

swarms/structs/mcp_deployer.py

​Overview

​Import

​How a target becomes a tool

​What a client sees

​Serving several targets

​Serving swarms

​State between calls

​Constructor

​Targets and naming

​Transport

​Auth

​Output

​Auth, in order of precedence

​Custom auth

​Token verifier

​Who made the request

​Timeouts

​stdio

​Methods

​run

​start

​stop

​Context manager

​add_tool

​authenticate

​build_app

​print_banner

​Properties

​Health check

​deploy_as_mcp

​Errors

​Examples

Serve an agent over MCP

Serve a team from one server

MCPManager

​Source

Overview

Import

How a target becomes a tool

What a client sees

Serving several targets

Serving swarms

State between calls

Constructor

Targets and naming

Transport

Auth

Output

Auth, in order of precedence

Custom auth

Token verifier

Who made the request

Timeouts

stdio

Methods

run

start

stop

Context manager

add_tool

authenticate

build_app

print_banner

Properties

Health check

deploy_as_mcp

Errors

Examples

Source