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

错误处理

其他信息

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