【Claude-Agent-SDK／FastAPI】API実行できるAIエージェントを作る
１．はじめに
本記事では、Claude Agent SDK と FastAPI と組み合わせて 「API経由で実行できるAIエージェント」 を構築してみました。
Claude Agent SDK とFastAPI を組み合わせることで、下記のようなことが実現できます。
- HTTP経由でエージェントにタスクをリクエストできる
- エージェントの実行状態やキャンセル操作をAPIで管理できる
- 他システムやフロントエンドからも利用しやすくなる
今回作ったものは GitHub で公開しています。
また、記事の内容は下記を前提に記述しています。
■ FastAPI
FastAPI は、PythonでAPIサーバを構築するのによく使用されています。
今回のようにAIエージェントを非同期で実行する場合に有効的なフレームワークです。
■ Claude Agent SDK
Claude Agent SDK は、Anthropic社が開発したAIエージェントを構築するためのソフトウェア開発キットです。
Claude CodeではCLI上でAIエージェントを実行しているのに対して、
Claude Agent SDKではプログラム上でAIエージェントを定義して実行することができます。
２．今回実装したAPI
今回実装したAPIは下記の通りです。
| エンドポイント | メソッド | 説明 |
|---|---|---|
/execute/ |
POST | 新規エージェントセッションを開始／再開 |
/status/{session_id} |
GET | セッションの状態、メッセージ履歴、結果を取得 |
/cancel/{session_id} |
POST | 実行中のセッションを中断 |
/sessions/ |
GET | 全セッションの詳細情報を取得 |
/sessions/cleanup |
DELETE | 古いセッションを削除 |
例えば /execute/ エンドポイントを実行する場合は、
次のようなリクエストを送ると新規エージェントが実行されます。
curl -X POST "http://localhost:8000/execute/" \
-H "Content-Type: application/json" \
-d '{
"prompt": "私はis0383kkです。自己紹介をしてください。",
"model": "us.anthropic.claude-3-5-sonnet-20241022-v1:0",
"max_turns": 10
}'
返却されるレスポンスは下記のようになります。
※非同期実行なので、この時点でレスポンスに実行結果のメッセージは含まれません。
{
"session_id": "XXXXXX",
"status": "running",
"message": "Session XXXXXX started successfully"
}
実行中の状況を取得する場合は、下記APIを実行することで取得できます。
curl "http://localhost:8000/status/XXXXXX"
■ セッション管理の仕組み
エージェント実行をAPI化する上で、各エージェントの実行を「セッション」として管理しています。
今回はエージェントのセッションを以下ステータスで管理しています。
-
pending: エージェント初期化中 -
running: エージェントが実行中 -
completed: 正常終了 -
error: 例外発生時 -
cancelled: ユーザーがキャンセルした場合
■ エージェント実行の基本方針
公式が提供しているサンプルをAPI実行できるように拡張しました。
Claude-Agent-SDKでは、下記のようにプロンプトとClaudeAgentOptionsで指定できるオプションをqueryメソッドの引数として使用することでAIエージェントを実行できます。
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
options = ClaudeAgentOptions(
system_prompt="You are an expert Python developer",
permission_mode='acceptEdits',
cwd="/home/user/project"
)
async for message in query(
prompt="Create a Python web server",
options=options
):
print(message)
asyncio.run(main())
上記に対して FastAPI の非同期機能を活用して、エージェントをバックグラウンドで実行しています。
エージェントを実行する際は、uuidをセッションIDとして発行しています。
@app.post("/execute/", response_model=ExecuteResponse)
async def execute_agent(request: ExecuteRequest):
session_id = str(uuid.uuid4())
# セッション情報を作成
session_info = SessionInfo(
session_id=session_id,
status="pending",
prompt=request.prompt,
options=request.to_claude_options()
)
# バックグラウンドタスクとして実行
task = asyncio.create_task(run_agent_session(session_info))
session_info.task = task
# セッション管理に登録
await session_manager.add_session(session_info)
return ExecuteResponse(
session_id=session_id,
status="pending",
message=f"Session {session_id} started successfully"
)
セッションを保持したまま再開する場合、
ClaudeAgentOptionsにresumeというオプションがありますので、そこで保持したいセッションID（uuid）を指定することで会話履歴を保持したままエージェント実行が可能です。
３．フロントエンドから実行してみる
それでは、実装した下記APIを画面から実行してみます。
-
/execute/:新規エージェントセッションを開始／再開 -
/status/{session_id}:セッションの状態、メッセージ履歴、結果を取得 -
/cancel/{session_id}:実行中のセッションを中断
■ 事前準備
本記事では下記前提とします。
- Python 3.8 以上
- Claude Codeが利用できる状態
requirements.txt を用意しているのでライブラリをインストールします。
pip install -r requirements.txt
準備が整ったら FastAPI サーバを下記コマンドで起動します。
$ python main.py
INFO: Started server process [17844]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
（Claude Codeに）動作確認用の画面を用意してもらいました。
ブラウザ上でindex.htmlを開けば準備完了です。
■ エージェントを実行
FastAPIサーバが起動している状態で、
プロンプトを入力して「エージェントを実行」ボタンから実行することができます。
実行の状況は「状態を確認」ボタンを押すことでレスポンスを確認できます。
■ セッションを保持したままエージェントを再実行
プロンプト＋セッションIDを指定して実行することでセッションを保持したままエージェントの実行ができます。
■ セッションを中断する
「セッションをキャンセル」ボタンから実行中のエージェント処理を中断できます。
４．おわりに
最近はAIエージェントを使ってAI駆動開発する流れから、AIエージェントを自分たちで作り運用／提供する流れにシフトしているように感じます。
従来は 「モデルをどう使うか」に注目が集まっていましたが、これからは「どんなエージェントをどう組み込むか」 という部分が重要になるのかなと思いました。
今回作ったリポジトリは比較的簡単に動かすことができるので、良かったら手元で動かしてみてください！
Discussion