Swarms API MCP Server

🚀 蜂群 API 文档

蜂群 API 为开发者提供了便捷的方式来调用相关服务，可用于运行蜂群完成任务、获取可用蜂群列表等。以下将详细介绍 API 的使用方法、费用说明、错误处理等内容。

🚀 快速开始

您可以使用以下代码调用蜂群 API：

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

# 环境变量获取 API_KEY
API_KEY = os.getenv("SWARMS_API_KEY")

# 创建一个示例负载
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": "这是一个测试任务。",
    "return_history": True
}

# 发送 POST 请求
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(response.json())

💻 使用示例

基础用法

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

# 环境变量获取 API_KEY
API_KEY = os.getenv("SWARMS_API_KEY")

# 创建一个示例负载
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": "这是一个测试任务。",
    "return_history": True
}

# 发送 POST 请求
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(response.json())

高级用法

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"

# 创建一个 MCP 服务器
mcp = FastMCP("swarms-api")

# 添加一个额外工具
@mcp.tool(name="swarm_completion", description="运行蜂群完成任务。")
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="获取可用蜂群列表。")
async def swarms_available() -> Any:
    """
    获取可用蜂群的列表。
    """
    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()  # 对于不良响应进行处理
        return response.json()

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

📄 费用说明

API 使用费用基于以下因素：

1. 使用的代理数量
2. 输入/输出令牌数
3. 模型选择
4. 时间段（高峰期有折扣）

如需了解详细定价信息，请访问 API 价格页面。

🔧 错误处理

📚 其他信息

如需了解更多信息，请参考 官方文档。