Agent SDKでカスタムエージェントを構築する

import { tool, createSdkMcpServer, query } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";
 
// 1. ツールを定義する
const getTemperature = tool(
  "get_temperature",
  "指定した緯度経度の現在の気温を取得する",
  {
    latitude: z.number().describe("緯度"),
    longitude: z.number().describe("経度")
  },
  async (args) => {
    const response = await fetch(
      `https://api.open-meteo.com/v1/forecast?latitude=${args.latitude}&longitude=${args.longitude}&current=temperature_2m`
    );
    const data: any = await response.json();
    return {
      content: [{ type: "text", text: `気温: ${data.current.temperature_2m}°C` }]
    };
  }
);
 
// 2. ツールをMCPサーバーにまとめる
const weatherServer = createSdkMcpServer({
  name: "weather",
  version: "1.0.0",
  tools: [getTemperature]
});
 
// 3. query()に渡して呼び出す
for await (const message of query({
  prompt: "東京の気温を教えて（緯度35.68, 経度139.77）",
  options: {
    mcpServers: { weather: weatherServer },
    allowedTools: ["mcp__weather__get_temperature"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

import asyncio
from typing import Any
import httpx
from claude_agent_sdk import tool, create_sdk_mcp_server, query, ClaudeAgentOptions, ResultMessage
 
 
# 1. ツールを定義する
@tool(
    "get_temperature",
    "指定した緯度経度の現在の気温を取得する",
    {"latitude": float, "longitude": float},
)
async def get_temperature(args: dict[str, Any]) -> dict[str, Any]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": args["latitude"], "longitude": args["longitude"], "current": "temperature_2m"},
        )
        data = response.json()
    return {"content": [{"type": "text", "text": f"気温: {data['current']['temperature_2m']}°C"}]}
 
 
# 2. ツールをMCPサーバーにまとめる
weather_server = create_sdk_mcp_server(name="weather", version="1.0.0", tools=[get_temperature])
 
 
async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_temperature"],
    )
    # 3. query()に渡して呼び出す
    async for message in query(prompt="東京の気温を教えて（緯度35.68, 経度139.77）", options=options):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)
 
 
asyncio.run(main())

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock
 
 
async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
 
    async with ClaudeSDKClient(options=options) as client:
        # 1回目: 調査を依頼
        await client.query("認証モジュールの実装を分析して")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
        # 2回目: 同じクライアントなので、1回目の文脈をそのまま引き継ぐ
        await client.query("その分析を踏まえて、JWTを使う形にリファクタリングして")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
 
asyncio.run(main())

let sessionId: string | undefined;
 
for await (const message of query({
  prompt: "認証モジュールを分析して",
  options: { allowedTools: ["Read", "Glob", "Grep"] }
})) {
  if (message.type === "result") {
    sessionId = message.session_id;
  }
}
 
// 後から、保存しておいたsession_idで明示的に再開する
for await (const message of query({
  prompt: "提案したリファクタリングを実装して",
  options: { resume: sessionId, allowedTools: ["Read", "Edit", "Write"] }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

// 簡単な状態管理の例: 処理済みタスクIDをメモリ上で記録する
const processedTaskIds = new Set<string>();
 
const processTask = tool(
  "process_task",
  "指定されたタスクIDのタスクを処理する。同じIDは2回処理しない",
  { taskId: z.string() },
  async (args) => {
    if (processedTaskIds.has(args.taskId)) {
      return {
        content: [{ type: "text", text: `タスク ${args.taskId} は処理済みです。スキップしました。` }]
      };
    }
    // ここで実際の処理を行う
    processedTaskIds.add(args.taskId);
    return {
      content: [{ type: "text", text: `タスク ${args.taskId} を処理しました。` }]
    };
  }
);

const fetchData = tool(
  "fetch_data",
  "指定したエンドポイントからデータを取得する",
  { endpoint: z.string().url() },
  async (args) => {
    try {
      const response = await fetch(args.endpoint);
 
      if (!response.ok) {
        // HTTPエラーはClaudeに「失敗」として伝える。スローはしない
        return {
          content: [{ type: "text", text: `APIエラー: ${response.status} ${response.statusText}` }],
          isError: true
        };
      }
 
      const data = await response.json();
      return { content: [{ type: "text", text: JSON.stringify(data, null, 2) }] };
    } catch (error) {
      // ネットワークエラーなどもここでキャッチし、ループを止めない
      return {
        content: [{ type: "text", text: `データ取得に失敗しました: ${error instanceof Error ? error.message : String(error)}` }],
        isError: true
      };
    }
  }
);

@tool("fetch_data", "指定したエンドポイントからデータを取得する", {"endpoint": str})
async def fetch_data(args: dict[str, Any]) -> dict[str, Any]:
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(args["endpoint"])
            if response.status_code != 200:
                # HTTPエラーはClaudeに「失敗」として伝える。例外は投げない
                return {
                    "content": [{"type": "text", "text": f"APIエラー: {response.status_code} {response.reason_phrase}"}],
                    "is_error": True,
                }
            data = response.json()
            return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}]}
    except Exception as e:
        # ネットワークエラーなどもここでキャッチし、ループを止めない
        return {
            "content": [{"type": "text", "text": f"データ取得に失敗しました: {str(e)}"}],
            "is_error": True,
        }

import { tool, createSdkMcpServer, query } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";
 
// 登録済みのTODOを記録し、重複登録を防ぐ簡易的な状態管理
const registeredTodos = new Set<string>();
 
const createIssue = tool(
  "create_issue",
  "TODOコメントの内容からGitHub Issueを作成する。同じ内容は重複登録しない",
  {
    title: z.string().describe("Issueのタイトル"),
    body: z.string().describe("Issueの本文。ファイルパスと行番号を含める")
  },
  async (args) => {
    const key = `${args.title}::${args.body}`;
    if (registeredTodos.has(key)) {
      return {
        content: [{ type: "text", text: `「${args.title}」はすでに登録済みのためスキップしました。` }]
      };
    }
 
    try {
      // 実際の運用では gh CLI や Octokit などで Issue を作成する
      // ここではダミーの成功レスポンスを返す
      registeredTodos.add(key);
      return {
        content: [{ type: "text", text: `Issueを作成しました: 「${args.title}」` }]
      };
    } catch (error) {
      // Issue作成に失敗してもループを止めず、Claudeに状況を伝える
      return {
        content: [{ type: "text", text: `Issue作成に失敗しました: ${error instanceof Error ? error.message : String(error)}` }],
        isError: true
      };
    }
  }
);
 
const issueServer = createSdkMcpServer({
  name: "issue-tracker",
  version: "1.0.0",
  tools: [createIssue]
});
 
async function main() {
  for await (const message of query({
    prompt: "src/配下のTODOコメントをすべて探し、それぞれについてcreate_issueツールでIssueを作成して。タイトルはTODOの要約、本文にはファイルパスと行番号を含めること。",
    options: {
      mcpServers: { "issue-tracker": issueServer },
      allowedTools: ["Read", "Glob", "Grep", "mcp__issue-tracker__create_issue"]
    }
  })) {
    if (message.type === "result" && message.subtype === "success") {
      console.log(message.result);
    }
  }
}
 
main();