基盤チームリーダー兼エンジニアリンググループGMの横本(@yokomotod)です。
このブログは基盤チームブログリレー4日目の記事です。3日目の記事は林さんの「BigQuery から高速にデータを持ってきてシームレスに Kotlin data class で使いたい – エムスリーテックブログ」でした。
今回の記事は、Claude Agent SDKの続編です。
前回記事でClaude Codeの公式SDKを使って対話型CLIやWebアプリを作りましたが、いくつか課題がありました。
その後SDKが大きく進化し、Streaming Input Mode でそれらの課題が解決しているのが個人的にビッグニュースだったのでアップデート版をお送りします。
Claude Codeをブラウザからも操作したい同志諸氏の参考になれば幸いです。
ソースコードはこちらでも公開しています。
はじめに
前回記事では、Claude Code SDK（当時の名称）を使って簡単な対話型CLIとWebアプリを実装しました。
その後、SDKやClaude Codeの周辺環境がいろいろと進化しているので、簡単に紹介します。
まず目につくのはSDK名称の変更です。claude-code-sdk が claude-agent-sdk になりました。
さらに、つい先週（2025年10月21日）、Anthropicから Claude Code on the Web が発表されました。
こちらはローカルではなくクラウド上でClaude Codeを動かす環境です*1
言うまでもなく、Claude Code本体も機能拡張が続いています。Hooks、Skillsなど、拡張性を高める機能が追加されました。
そして何より、SDKに Streaming Input Mode が導入されました。（ほとんど気にされてない？）
Claude Agent SDKの進化
v0.1.0: 名前変更とBreaking Changes
まず、名前が変わりました。
- Python:
claude-code-sdk→claude-agent-sdk - TypeScript:
@anthropic-ai/claude-code→@anthropic-ai/claude-agent-sdk
Coding以外の用途、汎用的なAgent作成に使えるSDKだから、という意図のようです。
また名称変更とともに、いくつかの設定の明示化が行われました。
主なBreaking Changes:
-
システムプロンプトのデフォルト廃止
- 以前: Claude Codeのシステムプロンプトがデフォルト
- v0.1.0: デフォルトで空
- 明示的に指定が必要:
system_prompt={"type": "preset", "preset": "claude_code"}
- 明示的に指定が必要:
-
ファイルシステム設定の自動ロード廃止
- 以前:
~/.claude/settings.jsonやCLAUDE.mdを自動読込 - v0.1.0: デフォルトでなにも読み込まない
setting_sources=["user", "project", "local"]で明示的に指定
- 以前:
Claude Code(CLI)との分離が強化されています。
…と、いうので触ってみたら、このシステムプロンプトのデフォルト廃止について、PythonSDKの実装が追いついていないバグを発見。PRを出したらシュッとマージしてもらえました*2。
v0.0.16: Streaming Input Mode導入
v0.1.0はリブランディング、破壊的変更はありますが、大きな機能強化がされたわけではありません。
むしろ、個人的重要アップデートがこちらです。
v0.0.16で ClaudeSDKClient が導入され、Streaming Input Modeが利用可能になりました。
公式ドキュメントでは、Streaming Input Modeが preferred way（推奨方法） とされています。
従来のSingle Message Input の制約:
- 画像の直接添付に非対応
- 動的メッセージキューイングに非対応
- リアルタイム割り込みに非対応
- フック統合に非対応
- 自然な複数ターン会話に非対応
なぜ複数ターン会話ができないのか？
Single Message Input では query() 関数を呼び出すたびに claude -p プロセスが起動されます。
そのままでは前回の状態は保持されません。
resume 引数に直前のsession idを渡すことで無理やり会話履歴を引き継ぐことはできますが、課題が残ります（前回記事参照）。
Streaming Input Modeによる解決:
Streaming Input Modeでは、ClaudeSDKClient を使うことで1つのプロセスで入出力を続けることが出来るようになります。
また次のような特徴が挙げられています。
- 永続的でインタラクティブなセッション
- 複数ターンで自然に会話コンテキストを保持
- リアルタイムフィードバック（生成中のレスポンス確認可能）
- 画像アップロード、ツール統合、フック対応などのリッチな機能
前回記事の最後で期待を持っていた機能が、まさに実装されていたのでした。
対話型CLIを再実装してみる
というわけで！ 今回はStreaming Input Modeを使って、前回作ったCLIやWeb UIを作り直してみます。
まずはまた、通常の claude コマンドのような対話型CLIを作ってみます。
SDKを使って claude コマンド相当の対話型CLIを実装すれば、そこから拡張して「オリジナルClaude Code」を作ることも可能です。
import readline import anyio from claude_agent_sdk import ( AssistantMessage, TextBlock, ClaudeSDKClient, ClaudeAgentOptions, ) async def main(): options = ClaudeAgentOptions( system_prompt={"type": "preset", "preset": "claude_code"}, setting_sources=["user", "project", "local"], permission_mode="bypassPermissions", ) async with ClaudeSDKClient(options=options) as client: while True: user_input = input("> ") if user_input.lower() == "/exit": break if not user_input.strip(): continue await client.query(user_input.strip()) async for message in client.receive_response(): print(f"[debug] {message}\n") if isinstance(message, AssistantMessage): print("● ", end="") for block in message.content: if isinstance(block, TextBlock): print(block.text, end="") print("\n") anyio.run(main)
ClaudeSDKClient を async with で管理し、query() と receive_response() を呼び出すだけです。
進化ポイント:
-
ClaudeSDKClientによるセッション管理async with ClaudeSDKClient(options=options) as client:でセッション開始- プロセスが永続化されていて、毎回
claude -pが起動していない
-
resumeが不要- 毎回プロセス起動ではなくなったので、resumeの必要もなくなりました
-
設定の明示化
system_prompt,setting_sourcesを明示的に指定- これはv0.1.0の対応
前回記事で遭遇していた、プロセス再起動による「Editツールのread済みフラグ消失」問題とはおさらばです。
WebアプリをWebSocketで再実装
続いて、WebからClaude Codeを操作できるWebアプリを作ってみます。
バックエンドはFastAPIを使い、フロントエンドとの通信にはWebSocketを使用します。
フロントエンドは前回同様React + Viteで実装しています。
（なお言うまでもなく、このままアプリ公開したら誰でも遠隔コード実行し放題になるのでダメゼッタイ）
from dataclasses import asdict from claude_agent_sdk import ( AssistantMessage, ClaudeAgentOptions, ClaudeSDKClient, ) from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) class ChatMessage(BaseModel): """Message from client to server.""" prompt: str @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): """WebSocket endpoint for chat communication.""" await websocket.accept() options = ClaudeAgentOptions( system_prompt={"type": "preset", "preset": "claude_code"}, setting_sources=["user", "project", "local"], permission_mode="bypassPermissions", ) try: async with ClaudeSDKClient(options=options) as client: while True: message_data = await websocket.receive_text() message = ChatMessage.model_validate_json(message_data) await client.query(message.prompt) async for response_message in client.receive_response(): data = asdict(response_message) data["type"] = type(response_message).__name__ if isinstance(response_message, AssistantMessage): for i, block in enumerate(response_message.content): block_data = asdict(block) block_data["type"] = type(block).__name__ data["content"][i] = block_data await websocket.send_json(data) except WebSocketDisconnect: print("[WebSocket] Client disconnected") except Exception as e: print(f"[WebSocket] Error: {e}") await websocket.close(code=1011, reason=str(e))
WebSocket通信にしたことで、CLIとほとんど同じスタイルとなり、特に説明することがなくなってしまいました*3。
フロントエンドの実装はGitHubで公開しています（フロントエンドの解説は前回参照）。
さらに今回は、Thinkingメッセージやツール呼び出しメッセージの表示にも対応してみました*4。
ClaudeSDKClientの（マニアックな）注意点
さて実は、前回記事のWeb UIのSSE通信実装から、今回はWebSocketに変わっています。
当初SSE通信のままClaudeSDKClientに切り替えようとして時間が溶けたので、ここで共有とともに供養とします。
結論としては、 ClaudeSDKClient のソースコードに、以下のように注意書きがされているのでした。
you cannot use a ClaudeSDKClient instance across different async runtime contexts (e.g., different trio nurseries or asyncio task groups)
意訳：異なる非同期ランタイムコンテキストを跨いでは使えないよ。
SSEでハマった問題:
次のような感じでsession_idをキーにClientインスタンスを管理して、リクエストを跨いで参照すればとりあえずいいか、と思ってました*5。
clients: dict[str, ClaudeSDKClient] = {} @app.post("/api/chat") async def chat(payload: Payload) -> StreamingResponse: if payload.session_id not in clients: clients[payload.session_id] = ClaudeSDKClient(...) await clients[payload.session_id].connect() client = clients[payload.session_id] async def generate_response(): await client.query(payload.prompt) async for message in client.receive_response(): yield f"data: {json.dumps(...)}\n\n" return StreamingResponse(generate_response(), ...)
が、動かしてみると2回目の呼び出しで謎のハングが発生します。
どうやらSSE接続が終了することでFastAPIのリクエストハンドラのasync contextが終了し、
それが伝搬して ClaudeSDKClient の内部タスクが終了され、状態が破壊されているようでした。時間が溶けた。
というわけで、SSEのままClientの動くcontextを独立させるのは大変だったので、大人しくWebSocketへ変更しました。
おわり
以上、Claude Agent SDKのStreaming Input Modeの紹介と実装でした。
Claude Agent SDK、まだまだ遊べそうなので、引き続き触ってみたいと思います。
SDKを使う際の参考になったり、SDKを使ってなにか作ってみたくなって頂けたら嬉しいです。
We’re Hiring
基盤チームでは、医療の横断的な基盤をつくるエンジニアを募集しています！
気になるツール、OSSの裏側までダイブせずにはいられない、ギークな仲間をお待ちしています！
エンジニア採用ページはこちら
エンジニア新卒採用サイトもオープンしました！
インターンもこちらから。常時募集しています！
