Claude Agent SDK完全ガイド：PythonとTypeScriptでClaudeのエージェント機能をアプリに組み込む

pip install claude-agent-sdk

npm install @anthropic-ai/claude-agent-sdk

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        max_turns=5,
    )

    async for message in query(
        prompt="カレントディレクトリのPythonファイルを一覧表示して、最も行数の多いファイルを教えてください",
        options=options,
    ):
        # メッセージタイプ別に処理
        if message.type == "assistant":
            print(message.content)
        elif message.type == "tool_result":
            print(f"[ツール実行結果] {message.tool_name}: {message.output[:100]}...")

asyncio.run(main())

import { query, ClaudeAgentOptions } from "@anthropic-ai/claude-agent-sdk";

async function main() {
  const options: ClaudeAgentOptions = {
    model: "claude-sonnet-4-6",
    maxTurns: 5,
  };

  for await (const message of query({
    prompt: "package.jsonを読み込んで、依存パッケージの一覧を表示してください",
    options,
  })) {
    if (message.type === "assistant") {
      console.log(message.content);
    }
  }
}

main();

from claude_agent_sdk import query, ClaudeAgentOptions, LifecycleHooks

class MyHooks(LifecycleHooks):
    async def on_tool_call(self, tool_name: str, tool_input: dict) -> None:
        # ツール呼び出し前にログ記録
        print(f"[HOOK] ツール呼び出し: {tool_name}, 入力: {tool_input}")

    async def on_tool_result(self, tool_name: str, result: str) -> None:
        # ツール実行後に結果をチェック
        print(f"[HOOK] ツール完了: {tool_name}, 文字数: {len(result)}")

async def main():
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        hooks=MyHooks(),
    )

    async for message in query(
        prompt="src/ディレクトリ内のTypeScriptファイルを検索して",
        options=options,
    ):
        pass

asyncio.run(main())

from claude_agent_sdk import query, ClaudeAgentOptions, MCPServerConfig

async def main():
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        mcp_servers=[
            MCPServerConfig(
                name="postgres",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-postgres"],
                env={"DATABASE_URL": "postgresql://localhost/mydb"},
            )
        ],
    )

    async for message in query(
        prompt="usersテーブルのレコード数を教えてください",
        options=options,
    ):
        if message.type == "assistant":
            print(message.content)

asyncio.run(main())

from claude_agent_sdk import query, ClaudeAgentOptions

ORCHESTRATOR_PROMPT = """
あなたはオーケストレーターです。以下のタスクを並列で実行してください:

1. サブエージェントAに「src/ディレクトリのPythonファイルを解析して品質レポートを作成」を依頼
2. サブエージェントBに「tests/ディレクトリのテストカバレッジを調査してレポートを作成」を依頼

両方の結果を受け取ったら、統合レポートを report.md に書き出してください。
"""

async def run_pipeline():
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        max_turns=20,  # マルチエージェントは多めのターンを確保
    )

    async for message in query(
        prompt=ORCHESTRATOR_PROMPT,
        options=options,
    ):
        if message.type == "assistant":
            print(message.content)

asyncio.run(run_pipeline())

from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(model="claude-sonnet-4-6")

    session = None
    async for message in query(prompt="大きなプロジェクトを解析して", options=options):
        session = message.session  # セッション情報を保持

    # コンテキスト使用量を確認
    if session:
        usage = session.get_context_usage()
        print(f"システムプロンプト: {usage.system_prompt_tokens} tokens")
        print(f"会話履歴: {usage.conversation_tokens} tokens")
        print(f"ツール定義: {usage.tool_tokens} tokens")
        print(f"合計: {usage.total_tokens} / {usage.max_tokens} tokens")

asyncio.run(main())

from typing import Annotated
from claude_agent_sdk import tool

@tool
def calculate_discount(
    price: Annotated[float, "元の価格（円）"],
    rate: Annotated[float, "割引率（0.0〜1.0）"],
) -> float:
    """割引後の価格を計算する"""
    return price * (1 - rate)

from claude_agent_sdk import query, ClaudeAgentOptions, BedrockConfig

async def main():
    options = ClaudeAgentOptions(
        model="anthropic.claude-sonnet-4-6-v1:0",
        bedrock=BedrockConfig(
            region="us-east-1",
            # AWS認証情報は環境変数 AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY を使用
        ),
    )

    async for message in query(
        prompt="AWSのS3バケット一覧を確認して",
        options=options,
    ):
        if message.type == "assistant":
            print(message.content)

asyncio.run(main())