Swarms API MCP Server

🚀 Swarm API Documentation

This API documentation provides detailed information on how to use the Swarm API, including code examples, cost explanations, error handling, and more.

🚀 Quick Start

You can use the following code to call the Swarm API:

import os
from typing import Any, Dict, List, Optional
import requests

# Get API_KEY from environment variables
API_KEY = os.getenv("SWARMS_API_KEY")

# Create an example payload
payload = {
    "agents": [
        {
            "agent_name": "Example Agent 1",
            "description": "This is a test agent.",
            "system_prompt": "You are an example agent.",
            "model_name": "gpt-4o",
            "role": "worker",
            "max_loops": 1,
            "temperature": 0.5
        }
    ],
    "max_loops": 1,
    "swarm_type": "ConcurrentWorkflow",
    "task": "This is a test task.",
    "return_history": True
}

# Send a POST request
response = requests.post(
    "https://swarms-api-285321057562.us-east1.run.app/v1/swarm/completions",
    headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
    json=payload
)

# Print the response content
print(response.json())

💻 Usage Examples

Basic Usage

import os
from typing import Any, Dict, List, Optional
import requests

# Get API_KEY from environment variables
API_KEY = os.getenv("SWARMS_API_KEY")

# Create an example payload
payload = {
    "agents": [
        {
            "agent_name": "Example Agent 1",
            "description": "This is a test agent.",
            "system_prompt": "You are an example agent.",
            "model_name": "gpt-4o",
            "role": "worker",
            "max_loops": 1,
            "temperature": 0.5
        }
    ],
    "max_loops": 1,
    "swarm_type": "ConcurrentWorkflow",
    "task": "This is a test task.",
    "return_history": True
}

# Send a POST request
response = requests.post(
    "https://swarms-api-285321057562.us-east1.run.app/v1/swarm/completions",
    headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
    json=payload
)

# Print the response content
print(response.json())

Advanced Usage

import os
from datetime import datetime
from typing import Any, Dict, List, Optional

import requests
import httpx
from fastmcp import FastMCP
from pydantic import BaseModel, Field
from swarms import SwarmType
from dotenv import load_dotenv

load_dotenv()

BASE_URL = "https://swarms-api-285321057562.us-east1.run.app"

# Create an MCP server
mcp = FastMCP("swarms-api")

# Add an additional tool
@mcp.tool(name="swarm_completion", description="Run swarm to complete tasks.")
def swarm_completion(swarm: SwarmSpec) -> Dict[str, Any]:
    api_key = os.getenv("SWARMS_API_KEY")
    headers = {"x-api-key": api_key, "Content-Type": "application/json"}

    payload = swarm.model_dump()

    response = requests.post(f"{BASE_URL}/v1/swarm/completions", json=payload, headers=headers)
    
    return response.json()

@mcp.tool(name="swarms_available", description="Get a list of available swarms.")
async def swarms_available() -> Any:
    """
    Get a list of available swarms.
    """
    headers = {"Content-Type": "application/json"}

    async with httpx.AsyncClient() as client:
        response = await client.get(f"{BASE_URL}/v1/models/available", headers=headers)
        response.raise_for_status()  # Handle bad responses
        return response.json()

if __name__ == "__main__":
    mcp.run(transport="sse")

📚 Documentation

Cost Explanation

The cost of using the API is based on the following factors:

1. Number of agents used
2. Number of input/output tokens
3. Model selection
4. Time period (discounts available during peak hours)

For detailed pricing information, please visit the API Pricing Page.

Error Handling

Additional Information

For more information, please refer to the Official Documentation.