tiangolo/fastapi

fastapi は何をしますか？

非同期WebSocketを介してマルチエージェントのストリーミングJSONインターフェースを安全にデプロイするためのボイラープレート設定。

概要

このプロジェクトテンプレートは、FastAPIを使用して、マルチエージェントシステム向けの安全な非同期JSONストリーミングWebSocketサーバーを迅速にセットアップするために設計されています。複数のクライアントやAIエージェントとの同時通信を処理するために必要なすべての構成とコードを提供し、非常に応答性の高いリアルタイムインタラクションを実現します。

特徴

• 非同期WebSocketサポート: FastAPIのネイティブ非同期機能に基づいた、効率的でノンブロッキングなWebSocketエンドポイント。
• マルチエージェント対応: 異なるエージェントやクライアントからの接続を別々のチャネルで処理し、効率的にメッセージをルーティングできます。
• JSONストリーミング: エージェントからクライアントへのJSONデータのシームレスなストリーミング。チャンクやインクリメンタルな応答に最適です。
• セキュリティ: JWTベースの認証と承認が組み込まれており、悪意のある接続から保護します。機密性の高いエージェント間通信に不可欠です。
• 自動ドキュメント: OpenAPI標準に基づく自動生成ドキュメントにより、セットアップとデバッグが容易になります。
• エラーハンドリング: 堅牢なエラー処理メカニズムにより、WebSocket接続のグレースフルな切断とリカバリを保証します。

クイックスタート

# リポジトリをクローン
git clone https://github.com/tiangolo/fastapi
cd fastapi

# 依存関係をインストール
pip install -r requirements.txt

# サーバーを起動
uvicorn main:app --reload

環境変数

.envファイルを作成し、以下の環境変数を設定します:

SECRET_KEY=your-secret-key
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

プロジェクト構造

.
├── main.py                # アプリケーションのエントリーポイント
├── requirements.txt        # Pythonパッケージの依存関係
├── .env                    # 環境変数
├── routers/                # APIルーター
│   └── websocket.py        # WebSocketルーター
├── models/                 # Pydanticモデル
│   └── schemas.py          # スキーマ定義
└── utils/                  # ユーティリティ関数
    └── auth.py             # 認証ユーティリティ

WebSocketエンドポイント

接続

WebSocketエンドポイントは/ws/{agent_id}で利用可能です。tokenクエリパラメータを指定して接続します:

const ws = new WebSocket("ws://localhost:8000/ws/agent_1?token=YOUR_JWT_TOKEN");

メッセージ形式

クライアントからサーバーへのメッセージ:

{
  "type": "agent.message",
  "data": {
    "instruction": "Summarize the latest news"
  }
}

サーバーからクライアントへのメッセージ:

{
  "type": "agent.response",
  "data": {
    "content": "Here is the summary...",
    "status": "completed"
  }
}

認証

すべてのWebSocket接続には有効なJWTが必要です。/tokenエンドポイントからトークンを取得します:

curl -X POST "http://localhost:8000/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=testuser&password=testpass"

例

複数のエージェントとの簡単なやり取り:

import asyncio
import websockets

async def connect_agent(agent_id, token):
    uri = f"ws://localhost:8000/ws/{agent_id}?token={token}"
    async with websockets.connect(uri) as websocket:
        await websocket.send(json.dumps({"type": "agent.message", "data": {"instruction": "Hello"}}))
        response = await websocket.recv()
        print(f"Agent {agent_id}: {response}")

asyncio.run(connect_agent("agent_1", "your_jwt_token"))

コントリビューション

問題の報告やプルリクエストは大歓迎です。コードベースに貢献する前に、CONTRIBUTING.mdファイルをご確認ください。

ライセンス

このプロジェクトはMITライセンスの下で公開されています。詳細はLICENSEファイルをご覧ください。