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 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 料金ページを参照してください。

エラーハンドリング

その他の情報

詳細情報については、公式ドキュメントを参照してください。