# Claude Agent SDKの便利だけど知られていない機能 🌍 エージェントの実行中にメッセージを送りたいと思ったことはありませんか？ストリーミング入力で対話的なセッションが実現します！ ストリーミング入力モードを使えば、エージェント実行中にメッセージの追加、中断、画像送信が可能です。 📌 タイトル：ストリーミング入力モード 🔗 URL： 🧩 概要 ストリーミング入力モード（推奨）は、`AsyncGenerator` を使った永続的なインタラクティブセッションを提供します。通常のシングルメッセージ入力（ワンショット）と異なり、セッション中にメッセージのキューイング、画像の添付、実行の中断が可能です。`yield` でメッセージを順次送信し、条件に応じてフォローアップメッセージを動的に生成できるため、チャットインターフェースや複雑なワークフローに最適です。 🛠 使い方 ```python # Python - ストリーミング入力でインタラクティブセッション from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock import base64 async def message_generator(): # 最初のメッセージ yield { "type": "user", "message": {"role": "user", "content": "セキュリティの問題を分析してください"}, } # フォローアップ（画像付き） with open("diagram.png", "rb") as f: image_data = base64.b64encode( yield { "type": "user", "message": { "role": "user", "content": [ {"type": "text", "text": "このアーキテクチャ図もレビューしてください"}, {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": image_data}}, ], }, } options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"]) async with ClaudeSDKClient(options) as client: await client.query(message_generator()) async for message in client.receive_response(): if isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, TextBlock): print(block.text) ``` ```typescript // TypeScript - AsyncGenerator でメッセージを逐次送信 import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk"; async function* generateMessages(): AsyncGenerator { yield { type: "user", message: { role: "user", content: "Analyze this codebase for security issues" }, parent_tool_use_id: null, }; // 条件に応じてフォローアップ await new Promise((resolve) => setTimeout(resolve, 2000)); yield { type: "user", message: { role: "user", content: "Now focus on authentication" }, parent_tool_use_id: null, }; } for await (const message of query({ prompt: generateMessages(), options: { maxTurns: 10, allowedTools: ["Read", "Grep"] } })) { if (message.type === "result" && message.subtype === "success") { console.log(message.result); } } ``` 🏗 本番システムへの組み込み方 ・チャットUIのメッセージ送信をジェネレータの `yield` に接続して、ユーザー入力をリアルタイムでエージェントに渡します ・画像アップロードを `base64` エンコードしてメッセージに含めることで、ビジュアル分析が可能になります ・中断シグナルを送信してエージェントの方向転換を行えるため、長時間実行タスクの制御に活用できます ・セッション中のコンテキストが自然に維持されるため、複数ターンの会話を管理する手間が省けます 💡 ユースケース 💬 カスタムチャットボットでユーザーとエージェントの対話的なセッションを構築する 🖼 コードレビュー中にスクリーンショットや設計図を動的に送信して分析させる 🔄 長時間のコード分析中に追加の指示やフォーカスエリアの変更を送る ⚠️ 注意点 ・シングルメッセージ入力では画像添付、動的メッセージキューイング、リアルタイム中断がサポートされません ・ステートレス環境（Lambda関数など）ではシングルメッセージ入力を使用してください ・Python ではストリーミング入力に `ClaudeSDKClient` のコンテキストマネージャが必要です ✨ ワンショットで終わらない対話こそ、AIエージェントの本領発揮です。ストリーミング入力でリッチな体験を構築しましょう！ #ClaudeAgentSDK# #AIAgent#

# OpenAI Agent SDKの便利で実践的な使い方 🌍 トレースの送信先を自由にカスタマイズしましょう。OpenAI バックエンドと並行して Langfuse や Datadog に送ったり、完全に置き換えたり -- 既存の観測基盤にシームレスに統合できます。 カスタムプロセッサで、トレースデータの送信先・加工・フィルタリングを完全にコントロールできます。 📌 タイトル：Tracing -- Custom processors / Ecosystem 🔗 URL： 🧩 概要 OpenAI Agent SDK のトレーシングは、カスタムプロセッサを追加・置換することで柔軟に拡張できます。`add_trace_processor()` で OpenAI バックエンドと並行して独自の処理を追加し、`set_trace_processors()` で送信先を Langfuse・Datadog・MLflow・Arize・Braintrust などに完全置換できます。OpenAI 以外のモデルを使用する場合でも、`set_tracing_export_api_key()` で無料のトレース可視化を利用できます。 🛠 使い方 `agents` から `add_trace_processor` と `TracingProcessor` をインポートします。`TracingProcessor` を継承した `MyProcessor` クラスを定義し、`on_trace_start`、`on_trace_end`、`on_span_start`、`on_span_end`、`shutdown` の各メソッドを実装します。` や `trace.duration_ms`、` などの情報にアクセスできます。`add_trace_processor(MyProcessor())` で登録すると、OpenAI バックエンドと並行してカスタム処理が実行されます。 🏗 実践的な使い方 **add_trace_processor() で OpenAI と並行して外部送信** OpenAI の Traces ダッシュボードはそのまま使いつつ、自社の観測基盤にもデータを送るパターンです。 `TracingProcessor` を継承して `DatadogProcessor` を作成します。`on_trace_start` で `dd_client.start_span(operation_name="agent.trace", resource= を呼び出し、`on_trace_end` で `dd_client.finish_span(metadata={"duration_ms": trace.duration_ms})` で完了を記録します。`on_span_start` では `operation_name=f"agent.{span.type}"` でスパンを開始し、`shutdown` で `dd_client.flush()` を呼びます。`add_trace_processor(DatadogProcessor(dd_client))` で登録すると、OpenAI と並行して Datadog にもトレースが送信されます。 **set_trace_processors() で送信先を完全置換** OpenAI にトレースを送らず、Langfuse や MLflow だけに送る場合はプロセッサを置き換えます。 `agents` から `set_trace_processors` をインポートし、送信先を完全に置換します。`set_trace_processors([LangfuseTracingProcessor()])` で Langfuse に、`[MLflowTracingProcessor()]` で MLflow に、`[ArizeTracingProcessor()]` で Arize に、`[BraintrustTracingProcessor()]` で Braintrust にそれぞれ切り替えられます。OpenAI バックエンドを含む既存のプロセッサはすべて置き換わります。 **set_tracing_export_api_key() で非 OpenAI モデルのトレースを無料で可視化** OpenAI 以外のモデル（Gemini、Claude 等）を使っている場合でも、トレース機能を無料で利用できます。 `agents` から `set_tracing_export_api_key` をインポートし、`set_tracing_export_api_key("your-tracing-export-key")` を呼び出します。OpenAI 以外のモデル（Gemini、Claude 等）を使っている場合でも、この設定でトレース機能を無料で利用できます。 **複数プロセッサの組み合わせ** 本番環境では、複数の観測ツールにトレースを送るのが一般的です。 `set_trace_processors` にリストで複数のプロセッサを渡します。`LangfuseTracingProcessor()` で詳細な LLM 分析、`DatadogProcessor(dd_client)` でインフラ監視との統合、`CustomAuditProcessor()` で社内監査ログを同時に実現できます。 💡 ユースケース 📡 add_trace_processor() で OpenAI ダッシュボードを維持しつつ Datadog にもトレースを並行送信 🔀 set_trace_processors() で Langfuse/MLflow/Arize/Braintrust に送信先を完全切り替え 🆓 set_tracing_export_api_key() で非 OpenAI モデル（Gemini、Claude）のトレースを無料で可視化 🔧 複数プロセッサ（Langfuse + Datadog + 監査ログ）を組み合わせた本番観測パイプラインの構築 ⚠️ 注意点 - `set_trace_processors()` は既存のプロセッサを**すべて置換**します。OpenAI バックエンドも含めて消えるので、残したい場合は `add_trace_processor()` を使ってください。 - カスタムプロセッサの `shutdown()` メソッドは必ず実装し、バッファに残ったデータをフラッシュしてください。 - 複数プロセッサを使う場合、各プロセッサの処理時間がトレース送信のオーバーヘッドになります。非同期送信を検討してください。 ✨ カスタムプロセッサを使えば、トレースデータを既存の観測スタックにそのまま流し込めます。まずは add で並行送信を試し、安定したら set で送信先を最適化していきましょう！ #OpenAIAgentSDK# #AIAgent#

# OpenAI Agent SDKの便利だけど知られていない機能 🌍 トレースデータをOpenAI以外の観測基盤にも送りたいと思ったことはありませんか？ カスタムトレースプロセッサを使えば、W&B、Arize、DataDogなど好みのプラットフォームにトレースを送信できます。 📌 タイトル：カスタムトレースプロセッサ 🔗 URL： 🧩 概要 `add_trace_processor()` を使うと、デフォルトのトレースプロセッサに加えて独自のプロセッサを追加できます。一方、`set_trace_processors()` を使うとデフォルトのプロセッサを完全に置き換えることができます。これにより、Weights & Biases、Arize、DataDog、独自のログ基盤など、任意のバックエンドにトレースデータを送信するカスタム連携が実現できます。 🛠 使い方 ```python from agents.tracing import ( add_trace_processor, set_trace_processors, TracingProcessor, ) # カスタムプロセッサを定義 class DataDogProcessor(TracingProcessor): def on_trace_start(self, trace): dd_tracer.start_span( def on_trace_end(self, trace): dd_tracer.finish_span() def on_span_start(self, span): dd_tracer.start_child_span( def on_span_end(self, span): dd_tracer.finish_child_span() # shutdown と force_flush も抽象メソッドなので実装が必要 def shutdown(self): dd_tracer.shutdown() def force_flush(self): dd_tracer.flush() # デフォルトに追加（OpenAIとDataDog両方に送信） add_trace_processor(DataDogProcessor()) # または、デフォルトを置き換え（DataDogのみに送信） set_trace_processors([DataDogProcessor()]) ``` 🏗 本番システムへの組み込み方 ・`add_trace_processor()` で既存のOpenAIトレースを維持しつつ、社内の観測基盤にも並行送信する ・障害調査用に独自のログストレージにトレースを保存するプロセッサを追加する ・環境変数でプロセッサの切り替えを制御し、開発/本番で送信先を変える ・プロセッサ内でのエラーが本体処理に影響しないよう、例外ハンドリングを適切に行う 💡 ユースケース 📊 Weights & Biasesでエージェント実行のメトリクスを可視化 🔍 Arize AIでLLM出力の品質モニタリング 📈 DataDogでエージェントのパフォーマンスダッシュボードを構築 🗄 社内データレイクへのトレースデータのアーカイブ ⚠️ 注意点 `set_trace_processors()` を使うとデフォルトのOpenAIトレースプロセッサが無効になります。OpenAIダッシュボードでのトレース確認も必要な場合は `add_trace_processor()` を使いましょう。カスタムプロセッサのパフォーマンスがエージェント全体の速度に影響する可能性があるため、軽量な実装を心がけてください。 ✨ カスタムトレースプロセッサで、既存の観測基盤とシームレスに統合しましょう。 #OpenAIAgentSDK# #AIAgent#

# OpenAI Agent SDKの便利で実践的な使い方 🌍 エージェントの「中で何が起きているか」を可視化しましょう。組み込みトレーシングで LLM 呼び出し・ツール実行・ハンドオフ・ガードレールの全フローを追跡できます。 トレーシングはデフォルトで有効。Traces ダッシュボードでエージェントの動作を視覚的に確認し、デバッグと品質改善を加速します。 📌 タイトル：Tracing 🔗 URL： 🧩 概要 OpenAI Agent SDK の組み込みトレーシングは、エージェント実行の全ステップ（LLM 呼び出し、ツール実行、ハンドオフ、ガードレール評価）を自動的に記録します。デフォルトで有効になっており、OpenAI の Traces ダッシュボードでフローを視覚的に確認できます。`with trace("name")` で関連する複数の実行をグループ化し、`flush_traces()` でワーカー環境でのトレース送信を確実にし、`RunConfig(trace_include_sensitive_data=False)` や環境変数でプライバシーを制御できます。 🛠 使い方 `agents` から `Agent`, `Runner`, `trace` をインポートします。`Agent` を作成して ` "AIの最新トレンドは？")` で実行するだけで、トレーシングはデフォルトで自動的に有効になり、LLM 呼び出しやツール実行が記録されます。 🏗 実践的な使い方 **with trace() で関連する実行をグループ化** 複数のエージェント実行をひとつのトレースにまとめて、ワークフロー全体を俯瞰できるようにします。 `agents` から `trace` をインポートし、`with trace("Joke workflow"):` ブロック内で複数の ` 呼び出しを実行します。ジョーク生成、評価、スコアが7未満なら改善という一連のフローが、ひとつのトレースにグループ化されてダッシュボードで俯瞰できるようになります。 **flush_traces() でワーカー環境のトレース送信を確実に** Celery/RQ/FastAPI のバックグラウンドワーカーでは、プロセス終了前にトレースを送信する必要があります。 `agents` から `flush_traces` をインポートします。Celery タスク内では ` message))` の後に `flush_traces()` を呼び出してトレースを送信します。FastAPI では `@app.on_event("shutdown")` フックで `flush_traces()` を呼び出し、プロセス終了前のトレース送信を保証します。 **プライバシー制御で機密データを保護** ZDR（Zero Data Retention）環境や個人情報を扱う場合のトレース制御パターンです。 ` に `run_config=RunConfig(trace_include_sensitive_data=False)` を渡すと（既定値は `True`）、LLM の入出力や関数呼び出しのデータがトレースに含まれなくなります。既定値は環境変数 `OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA` でも切り替えられます。トレーシング自体を完全に無効化するには、環境変数 `OPENAI_AGENTS_DISABLE_TRACING=1` を設定するか、`RunConfig(tracing_disabled=True)` を指定します。 💡 ユースケース 📊 デフォルト有効のトレーシングで、LLM/ツール/ハンドオフ/ガードレールのフローを Traces ダッシュボードで可視化 🔗 with trace("Joke workflow") で複数のエージェント実行をグループ化し、ワークフロー全体を俯瞰 🔄 Celery/RQ/FastAPI ワーカーで flush_traces() を呼び出し、プロセス終了前のトレース送信を保証 🔒 trace_include_sensitive_data=False で医療・金融データのプライバシーを保護 ⚠️ 注意点 - トレーシングはデフォルトで有効のため、意図せず機密データがトレースに含まれる可能性があります。本番環境では `trace_include_sensitive_data` の設定を確認してください。 - `flush_traces()` はブロッキング処理のため、リクエストのクリティカルパスには入れず、シャットダウンフックやタスク終了時に呼び出してください。 - トレースデータは OpenAI に送信されます。コンプライアンス要件がある場合は `OPENAI_AGENTS_DISABLE_TRACING=1` で無効化するか、カスタムプロセッサ（次セクション参照）で送信先を変更してください。 ✨ 組み込みトレーシングは「設定ゼロで動く」のが最大の強みです。まずはそのままダッシュボードを確認し、ワークフローのグループ化やプライバシー設定を必要に応じて追加していきましょう！ #OpenAIAgentSDK# #AIAgent#

# Claude Agent SDKの便利だけど知られていない機能 🌍 エージェントの応答を待つ間、画面が真っ白で不安になったことはありませんか？リアルタイムストリーミングで解決です！ `include_partial_messages` を有効にすれば、テキストやツール呼び出しが生成される様子をリアルタイムで受信できます。 📌 タイトル：リアルタイム部分メッセージ 🔗 URL： 🧩 概要 `include_partial_messages`（Python）/ `includePartialMessages`（TypeScript）を `True` に設定すると、通常の `AssistantMessage` に加えて `StreamEvent` メッセージが送信されます。`StreamEvent` にはClaude APIの生のストリーミングイベント（`content_block_delta` の `text_delta` や `input_json_delta` など）が含まれ、テキストの断片やツール入力のチャンクをリアルタイムで受信できます。これにより、タイピング風のテキスト表示やツール実行中のステータス表示など、応答性の高いUIを構築できます。 🛠 使い方 ```python # Python - テキストチャンクをリアルタイム表示 from claude_agent_sdk import query, ClaudeAgentOptions from claude_agent_sdk.types import StreamEvent options = ClaudeAgentOptions(include_partial_messages=True) async for message in query(prompt="データベースの仕組みを説明してください", options=options): if isinstance(message, StreamEvent): event = message.event if event.get("type") == "content_block_delta": delta = event.get("delta", {}) if delta.get("type") == "text_delta": print(delta.get("text", ""), end="", flush=True) ``` ```typescript // TypeScript - ツール呼び出しのストリーミング表示 import { query } from "@anthropic-ai/claude-agent-sdk"; let currentTool: string | null = null; for await (const message of query({ prompt: "Read the README.md file", options: { includePartialMessages: true } })) { if (message.type === "stream_event") { const event = message.event; if (event.type === "content_block_start" && event.content_block.type === "tool_use") { currentTool = event.content_block.name; process.stdout.write(`\n[Using ${currentTool}...]`); } else if (event.type === "content_block_delta" && === "text_delta") { process.stdout.write( } else if (event.type === "content_block_stop" && currentTool) { console.log(" done"); currentTool = null; } } } ``` 🏗 本番システムへの組み込み方 ・チャットUIで `text_delta` を受信するたびにメッセージバブルにテキストを追加し、タイピング風の表示を実現します ・`content_block_start`（type: `tool_use`）でツール名を検知し、「実行中...」のスピナーを表示します ・`message_start`/`message_stop` イベントでメッセージの開始・完了を追跡します ・テキストデルタは自分で結合する必要があるため、アキュムレータを用意します 💡 ユースケース 💬 ChatGPT風のリアルタイムタイピング表示をカスタムUIに実装する 🔧 エージェントが使用中のツール名とその進捗をステータスバーに表示する 📝 長文の分析結果を、完成を待たずに部分的に表示開始する ⚠️ 注意点 ・`StreamEvent` のイベントは生のClaude APIイベントであり、蓄積済みテキストではありません。テキストデルタを自分で結合する必要があります ・構造化出力（`structured_output`）はストリーミングデルタでは送信されず、最終の `ResultMessage` にのみ含まれます ・ツール実行中（`in_tool` フラグ）はテキストストリーミングを一時停止するなど、UIの状態管理が必要です ✨ 「処理中です、お待ちください...」だけの画面はもう卒業。部分メッセージでユーザーとエージェントの距離を縮めましょう！ #ClaudeAgentSDK# #AIAgent#

# OpenAI Agent SDKの便利で実践的な使い方 🌍 自前の MCP サーバーをエージェントに接続しましょう。HTTP・SSE・stdio の3つのトランスポートと、接続管理・承認制御の実践パターンを紹介します。 ローカル MCP サーバーは自社インフラ上で動作するため、低レイテンシ・高セキュリティ・カスタムツールの自由度が得られます。 📌 タイトル：MCP -- Streamable HTTP / SSE / stdio 🔗 URL： 🧩 概要 ローカル MCP サーバーは、自分でホストする MCP サーバーをエージェントのツールとして接続する仕組みです。`MCPServerStreamableHttp`（HTTP ベース、認証・リトライ・タイムアウト対応）、`MCPServerSse`（Server-Sent Events）、`MCPServerStdio`（ローカル CLI プロセス）の3つのトランスポートから選択できます。`MCPServerManager` で複数サーバーの一括接続管理、`require_approval` でツール別の承認制御、`tool_meta_resolver` でテナント情報の注入が可能です。 🛠 使い方 `agents` から `Agent`, `Runner` を、`agents.mcp` から `MCPServerStreamableHttp`, `MCPServerStdio` をインポートします。HTTP 接続の場合は `Agent` の `mcp_servers` に `MCPServerStreamableHttp` を渡し、接続情報は `params` 辞書にまとめて `params={"url": "http://internal-mcp.corp:8080/mcp", "headers": {"Authorization": "Bearer "}, "timeout": 30}` と指定します。stdio 接続の場合は `MCPServerStdio` を使い、`params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]}` でローカル CLI プロセスを起動します。 🏗 実践的な使い方 **MCPServerStreamableHttp で社内 MCP に認証付き接続** 低レイテンシが求められる社内サービスへの接続パターンです。認証ヘッダ、リトライ、タイムアウトを設定します。 `MCPServerStreamableHttp` の `params` に `"url": "http://internal-mcp.corp:8080/mcp"`、`"headers"` で `Authorization` ヘッダに `os.getenv('MCP_TOKEN')` から取得したトークンを設定し、`"timeout": 30` を指定します。リトライ回数は `params` の外側のキーワード引数 `max_retry_attempts=3` で設定します。これを `Agent` の `mcp_servers` に渡すことで、認証付きの社内 MCP 接続が完成します。 **MCPServerStdio でローカル CLI ツールを接続** ファイルシステムサーバーや自作 CLI ツールを stdio トランスポートで接続します。 `MCPServerStdio` で `params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]}` を指定してファイルシステムサーバーを接続します。自作ツールの場合は `params={"command": "python", "args": ["-m", "my_custom_mcp_server"], "env": {"DB_URL": os.getenv("DB_URL")}}` で環境変数を渡します。両方を `Agent` の `mcp_servers` リストに渡せば、複数の stdio ツールを同時利用できます。 **MCPServerManager で複数サーバーの一括管理** 複数の MCP サーバーを効率的に管理し、障害時の再接続も制御します。 `agents.mcp` から `MCPServerManager` をインポートし、`servers` に複数の MCP サーバーインスタンスを渡します。`await manager.connect_all()` で一括接続し、` でアクティブなサーバーを確認できます。`await manager.reconnect(failed_only=True)` で障害が発生したサーバーのみ再接続します。 **require_approval のツール別マップ + tool_meta_resolver** ツールごとに承認ポリシーを変え、さらにツール実行時にテナント情報を注入するパターンです。 ツールごとの承認ポリシーを辞書で定義します。`"search_docs"` と `"read_file"` は `"never"`、`"write_file"`・`"delete_file"`・`"execute_query"` は `"always"` とマッピングします。`tool_meta_resolver` に渡す関数は `MCPToolMetaContext` を受け取り、` からテナント情報を読んで `{"tenant_id": str(tenant_id), "source": "agents-sdk"}` のようなメタデータ辞書を返します（ツール引数そのものを書き換えるのではなく、呼び出しごとの `_meta` として送られます）。`MCPServerStreamableHttp` の `require_approval` にこの辞書を、`tool_meta_resolver` にこの関数を渡すことで、ツール別承認制御とテナント情報の自動付与を実現します。 💡 ユースケース 🏢 MCPServerStreamableHttp で社内 MCP サーバーに認証・タイムアウト付きで低レイテンシ接続 📁 MCPServerStdio でローカルのファイルシステムサーバーや自作 CLI ツールをエージェントに接続 🔄 MCPServerManager で複数サーバーを一括管理し、failed_only=True で障害サーバーのみ再接続 🔐 ツール別 require_approval マップ + tool_meta_resolver でテナント ID を自動注入しマルチテナント対応 ⚠️ 注意点 - MCPServerStdio はローカルプロセスを起動するため、コンテナ環境では対象コマンドがインストールされている必要があります。 - MCPServerManager の reconnect は接続エラーのリカバリであり、ツール実行中のエラーハンドリングは別途必要です。 - tool_meta_resolver で注入する情報にシークレットを含めないでください。ログに出力される可能性があります。 ✨ 3つのトランスポートと MCPServerManager を使いこなせば、社内サービスからローカルツールまで自在に接続できます。まずは stdio で手元のツールを繋いで、HTTP で本番接続に移行していきましょう！ #OpenAIAgentSDK# #AIAgent#

# OpenAI Agent SDKの便利だけど知られていない機能 🌍 複雑なマルチエージェント構成、頭の中だけで把握できていますか？ draw_graphを使えば、エージェントの構成をGraphvizで視覚的に可視化し、全体像を一目で確認できます。 📌 タイトル：draw_graph によるエージェント可視化 🔗 URL： 🧩 概要 `draw_graph(agent)` を呼び出すだけで、エージェント・ツール・ハンドオフの関係をGraphvizのグラフとして描画できます。エージェントは黄色の四角形、ツールは緑の楕円、MCPサーバーは灰色で表示されます。ハンドオフは実線矢印、ツール接続は点線、MCP接続は破線で区別されます。生成されたグラフは `.view()` でブラウザ表示、`filename=` でファイル保存が可能です。 🛠 使い方 ```python # vizオプション付きでインストール # pip install "openai-agents[viz]" from agents import Agent, function_tool from agents.extensions.visualization import draw_graph @function_tool def search(query: str) -> str: return "results" sub_agent = Agent(name="Researcher", tools=[search]) main_agent = Agent( name="Coordinator", handoffs=[sub_agent], ) # グラフを描画してブラウザで表示 graph = draw_graph(main_agent) graph.view() # ファイルに保存 draw_graph(main_agent, filename="agent_graph") ``` 🏗 本番システムへの組み込み方 ・CI/CDパイプラインでグラフを自動生成し、ドキュメントとして保存する ・コードレビュー時にエージェント構成の変更を視覚的に確認する ・新しいメンバーのオンボーディングに、システム全体像の把握資料として活用する ・構成変更前後のグラフを比較して、意図しないハンドオフの追加・削除を検出する 💡 ユースケース 🗺 マルチエージェントシステムのアーキテクチャドキュメント自動生成 👥 チーム内でのエージェント構成の共有・レビュー 🐛 ハンドオフの循環参照や孤立ツールの検出 📊 プロジェクト提案時のシステム構成図作成 ⚠️ 注意点 利用には `pip install "openai-agents[viz]"` で追加依存（Graphviz）のインストールが必要です。システムにGraphvizバイナリもインストールされている必要があります。非常に大規模なエージェント構成では、グラフが複雑になりすぎる場合があります。 ✨ draw_graphで、エージェント構成を「見える化」してチーム全体の理解を深めましょう。 #OpenAIAgentSDK# #AIAgent#

# Claude Agent SDKの便利だけど知られていない機能 🌍 エージェントがファイルを勝手に削除しようとしたら？実行時承認で安全にコントロールできます！ `canUseTool` コールバックを使えば、ツール実行前にリアルタイムで承認・拒否・入力変更が可能です。 📌 タイトル：canUseTool コールバックによる実行時承認 🔗 URL： 🧩 概要 `canUseTool` はクエリオプションに渡すコールバック関数で、権限評価フローの最終ステップ（6番目）で呼び出されます。先行するステップ（allow ルールや `acceptEdits`・`bypassPermissions` モードなど）で自動承認されたツール呼び出しではコールバックは呼ばれないため、すべてのツール呼び出しに適用したいチェックは `PreToolUse` フックを使ってください。コールバックはツール名と入力パラメータを受け取り、`allow`（許可）または `deny`（拒否）を返します。許可時には入力を変更することも可能で、例えばパスをサンドボックス内に書き換えたり、危険なコマンドをブロックしたりできます。`dontAsk` モードでは呼び出されません。 🛠 使い方 ```python # Python - ツール実行の承認コールバック from claude_agent_sdk import ClaudeAgentOptions, query from claude_agent_sdk.types import PermissionResultAllow, PermissionResultDeny, HookMatcher async def can_use_tool(tool_name, input_data, context): print(f"\nツール: {tool_name}") if tool_name == "Bash": print(f"コマンド: {input_data.get('command')}") # rm コマンドをブロックしてアーカイブを提案 if "rm" in input_data.get("command", ""): return PermissionResultDeny( message="削除の代わりにアーカイブしてください" ) response = input("許可しますか？ (y/n): ") if response.lower() == "y": return PermissionResultAllow(updated_input=input_data) return PermissionResultDeny(message="ユーザーが拒否しました") # ダミーフック（Python のストリーミングモードで必要） async def dummy_hook(input_data, tool_use_id, context): return {"continue_": True} ``` ```typescript // TypeScript - 入力を変更して許可する例 import { query } from "@anthropic-ai/claude-agent-sdk"; for await (const message of query({ prompt: "Clean up temporary files", options: { canUseTool: async (toolName, input) => { if (toolName === "Bash") { // パスをサンドボックス内に書き換え const sandboxed = { ...input, command: input.command.replace("/tmp", "/tmp/sandbox") }; return { behavior: "allow", updatedInput: sandboxed }; } return { behavior: "allow", updatedInput: input }; } } })) { /* ... */ } ``` 🏗 本番システムへの組み込み方 ・Web UIでツール実行の承認ダイアログを表示し、ユーザーの判断を返します ・`context.suggestions` から `PermissionUpdate` を取得し、「今後も許可」の永続化ルールを設定できます ・危険な操作（`rm`、`DROP TABLE` など）は自動的にブロックし、安全な代替案をメッセージで提案します ・入力のサニタイズ（パスの制限、パラメータの上書き）で、エージェントの行動範囲を動的に制御します 💡 ユースケース 🔒 本番データベースへの書き込みコマンドを承認制にする 📁 ファイル操作を特定のディレクトリに制限する（パスの動的書き換え） 👥 マルチユーザー環境で、ユーザーの権限レベルに応じてツール利用を制御する ⚠️ 注意点 ・Python では `canUseTool` を使うにはストリーミングモードと `PreToolUse` フックの設定が必要です ・`dontAsk` モードではコールバックがスキップされるため、セキュリティ上の制限にはフックを使用してください ・コールバックは回答まで無期限に実行を一時停止します。長時間のケースには `defer` フック決定を利用してください ✨ 「全部おまかせ」か「全部手動」かの二択ではなく、canUseTool で「要所だけ人間が判断する」スマートな運用を実現しましょう！ #ClaudeAgentSDK# #AIAgent#

# OpenAI Agent SDKの便利で実践的な使い方 🌍 MCP サーバーのツールを「全部見せる」のではなく、フィルタ・プロンプト・キャッシュで賢く制御しましょう。権限スコープの絞り込みから動的フィルタリングまで、安全で高速な MCP 活用パターンです。 ツールフィルタで権限を最小化し、プロンプトで動的指示を取得し、キャッシュでレイテンシを削減する -- MCP の運用を本格化する3つの機能です。 📌 タイトル：MCP -- Tool filtering / Prompts / Caching 🔗 URL： 🧩 概要 MCP サーバーが公開するツールをそのまますべてエージェントに渡すと、不要なツール呼び出しや権限超過のリスクがあります。`create_static_tool_filter` で許可するツールを静的に制限し、動的フィルタ関数で実行時の条件（ロール・テナント等）に応じてツールを出し分けできます。`get_prompt` で MCP サーバーから動的な指示を取得し、`cache_tools_list=True` でツール一覧のキャッシュによりレイテンシを改善、`invalidate_tools_cache()` で変更時にキャッシュをクリアします。 🛠 使い方 `agents.mcp` から `MCPServerStreamableHttp` と `create_static_tool_filter` をインポートします。`MCPServerStreamableHttp` の `tool_filter` に `create_static_tool_filter(allowed_tool_names=["read_file", "write_file"])` を渡すことで、エージェントが使えるツールを `read_file` と `write_file` のみに制限できます。 🏗 実践的な使い方 **create_static_tool_filter で権限スコープを絞り込む** エージェントの役割に応じて、使えるツールを最小限に制限するパターンです。 読み取り専用エージェント用には `create_static_tool_filter(allowed_tool_names=["read_file", "search_docs", "list_files"])` でフィルタを作成し、`MCPServerStreamableHttp` の `tool_filter` に渡します。フルアクセス用には `allowed_tool_names=["read_file", "write_file", "delete_file", "search_docs"]` と設定します。同じ MCP サーバーでもエージェントの役割に応じて異なるツールセットを公開できます。 **動的フィルタで danger_ ツールを非表示にする** ロールに応じて危険なツールを動的に非表示にするパターンです。 `agents.mcp` から `ToolFilterContext` をインポートし、`role_based_filter` 関数を定義します。` が `"CodeReviewer"` で `"danger_")` の場合に `False` を返してツールを非表示にし、それ以外は `True` を返します。この関数を `MCPServerStreamableHttp` の `tool_filter` に渡すことで、ロールに応じた動的なツールフィルタリングが実現できます。 **get_prompt で MCP サーバーから動的指示を取得** MCP サーバーが提供するプロンプトを取得して、エージェントの指示を動的に構成します。 `MCPServerStreamableHttp` を作成し、`async with mcp:` のコンテキスト内で `prompt_result = await mcp.get_prompt("operation_guidelines")` を呼び出して MCP サーバーからプロンプトを取得します（第2引数に `{"focus": "..."}` のような引数辞書も渡せます）。取得した `prompt_result.messages[0].content.text` を `Agent` の `instructions` に埋め込むことで、エージェントの指示を動的に構成できます。 **cache_tools_list でレイテンシを改善** ツール一覧の取得をキャッシュし、MCP サーバーへの不要なリクエストを減らします。 `MCPServerStreamableHttp` の `cache_tools_list=True` を設定すると、ツール一覧の取得結果がキャッシュされ、MCP サーバーへの不要なリクエストが削減されます。ツールが追加・変更された場合は `await mcp.invalidate_tools_cache()` でキャッシュをクリアします。 💡 ユースケース 🔒 create_static_tool_filter で read_file/write_file のみ許可し、最小権限の原則を実現 🎭 動的フィルタで CodeReviewer には danger_ ツールを非表示にし、ロール別のアクセス制御 📜 get_prompt で MCP サーバーから最新の運用ガイドラインを取得し、エージェントの指示を動的に更新 ⚡ cache_tools_list=True でツール一覧キャッシュを有効化し、繰り返しのツール取得レイテンシを削減 ⚠️ 注意点 - 静的フィルタの `allowed_tool_names` はツール名のタイプミスに注意してください。存在しないツール名を指定してもエラーにならず、単にフィルタされないだけです。 - 動的フィルタは毎回のツール一覧取得時に実行されるため、重い処理を入れるとレイテンシに影響します。 - `cache_tools_list=True` 使用時にツールが追加・変更された場合、`invalidate_tools_cache()` を呼ばないと古い一覧が使われ続けます。 ✨ フィルタ・プロンプト・キャッシュの3つを組み合わせれば、MCP ツールの安全性・柔軟性・パフォーマンスを同時に向上できます。まずは静的フィルタで権限を絞り、必要に応じて動的フィルタやキャッシュを導入していきましょう！ #OpenAIAgentSDK# #AIAgent#