MCP

A hosted Model Context Protocol server for the Swarms API. Point any MCP client at one URL to give it agents, swarms, and batch execution as callable tools. No local process to run.

Endpoint
https://mcp.swarms.world/mcp
Checking

Status

Checking…

Probing the endpoint from our region.

Handshake latency

Measured server-side from our region.

Observed uptime

No observations recorded yet on this server.

Connect

The server speaks streamable HTTP and authenticates with the same x-api-key header as the REST API. Create a key in API keys and export it as SWARMS_API_KEY.

# pip install mcp
import asyncio
import os

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

SWARMS_MCP_URL = "https://mcp.swarms.world/mcp"


async def main() -> None:
    async with streamablehttp_client(
        SWARMS_MCP_URL,
        headers={"x-api-key": os.environ["SWARMS_API_KEY"]},
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Every tool the server exposes, with its input schema.
            tools = await session.list_tools()
            for tool in tools.tools:
                print(tool.name)


asyncio.run(main())

Tools

Reading the tool list from the server…

Run a swarm

Call the swarm completions tool through the session you just opened. This example runs a ConcurrentWorkflow so both analysts work the same task in parallel and return together. Tool names come from list_tools, so run the connect snippet first to confirm what your key can reach.

import asyncio
import json
import os

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

SWARMS_MCP_URL = "https://mcp.swarms.world/mcp"


async def main() -> None:
    async with streamablehttp_client(
        SWARMS_MCP_URL,
        headers={"x-api-key": os.environ["SWARMS_API_KEY"]},
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            result = await session.call_tool(
                "run_swarm_v1_swarm_completions_post",
                {
                    "name": "Market Research Swarm",
                    "description": "Three analysts research the same task in parallel",
                    "swarm_type": "ConcurrentWorkflow",
                    "task": "Analyze the impact of AI agents on modern healthcare",
                    "agents": [
                        {
                            "agent_name": "Market Analyst",
                            "system_prompt": "You analyze market trends and opportunities.",
                            "model_name": "gpt-5.4",
                            "max_loops": 1,
                        },
                        {
                            "agent_name": "Risk Analyst",
                            "system_prompt": "You identify risks and regulatory constraints.",
                            "model_name": "claude-haiku-4-5",
                            "max_loops": 1,
                        },
                    ],
                    "max_loops": 1,
                },
            )

            for block in result.content:
                if block.type == "text":
                    print(json.dumps(json.loads(block.text), indent=2))


asyncio.run(main())

The same swarm without MCP, straight against the REST endpoint:

Swarm completion over HTTP

import os
import requests

payload = {
    "name": "Market Research Swarm",
    "description": "Three analysts research the same task in parallel",
    "swarm_type": "ConcurrentWorkflow",
    "task": "Analyze the impact of AI agents on modern healthcare",
    "agents": [
        {
            "agent_name": "Market Analyst",
            "system_prompt": "You analyze market trends and opportunities.",
            "model_name": "gpt-5.4",
            "max_loops": 1
        },
        {
            "agent_name": "Risk Analyst",
            "system_prompt": "You identify risks and regulatory constraints.",
            "model_name": "claude-haiku-4-5",
            "max_loops": 1
        }
    ],
    "max_loops": 1
}

response = requests.post(
    "https://api.swarms.world/v1/swarm/completions",
    headers={
        "x-api-key": os.environ["SWARMS_API_KEY"],
        "Content-Type": "application/json",
    },
    json=payload,
)
response.raise_for_status()
print(response.json())

Tutorials

Documentation