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 价格页面。

错误处理

其他信息

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