Claude APIでAIエージェントを作る実装ガイド【2026】Tool Use・Tool Runner・MCPをPython/TypeScriptで

# Python（3.10以上を推奨。仮想環境を切る）
python -m venv .venv && source .venv/bin/activate
pip install anthropic

# TypeScript / Node.js
npm install @anthropic-ai/sdk
npm install zod   # Tool RunnerでbetaZodToolを使う場合

# APIキー（.envに置くなら.gitignoreに必ず追加）
export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY を自動で読む

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    system="あなたは社内ヘルプデスクのアシスタントです。簡潔に日本語で答えます。",
    messages=[{"role": "user", "content": "VPNにつながらないときに最初に確認することは？"}],
)

# content はブロックの配列。text 以外（thinking など）も入りうる
for block in response.content:
    if block.type == "text":
        print(block.text)

print(response.stop_reason, response.usage.input_tokens, response.usage.output_tokens)

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic(); // ANTHROPIC_API_KEY を自動で読む

const response = await client.messages.create({
  model: "claude-opus-5",
  max_tokens: 16000,
  system: "あなたは社内ヘルプデスクのアシスタントです。簡潔に日本語で答えます。",
  messages: [{ role: "user", content: "VPNにつながらないときに最初に確認することは？" }],
});

for (const block of response.content) {
  if (block.type === "text") console.log(block.text);
}

import json
import logging
import anthropic

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("helpdesk")

client = anthropic.Anthropic()
MODEL = "claude-opus-5"
MAX_TURNS = 8  # 暴走・コスト青天井を防ぐ上限

SYSTEM = """あなたは株式会社サンプルの社内ヘルプデスク担当です。
- 回答の前に必ず search_faq で社内FAQを確認し、FAQに基づいて答えてください。
- FAQで解決しない、または申請・作業が必要な場合だけ create_ticket を使います。
- 推測で手順を作らないでください。分からないことは分からないと伝えます。
- 回答は日本語で、手順は番号付きで簡潔に。"""

TOOLS = [
    {
        "name": "search_faq",
        "description": (
            "社内FAQ（IT・経理・人事・総務）を全文検索し、関連度の高い順に最大5件を返す。"
            "回答を作る前に必ず使う。0件なら言い換えて1回だけ再検索する。"
            "返り値は {id, title, body, url} のJSON配列。個人の申請状況や人事情報は検索できない。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "検索キーワード。質問文ではなく製品名・エラー文・手続き名など",
                },
                "category": {
                    "type": "string",
                    "enum": ["it", "accounting", "hr", "general"],
                    "description": "分かる場合のみ指定。不明なら省略",
                },
            },
            "required": ["query"],
        },
    },
    {
        "name": "create_ticket",
        "description": (
            "担当部署へのチケットを起票する。FAQで解決しない場合や、権限付与・機器交換など"
            "人の作業が必要な場合のみ使う。起票前に利用者本人の確認が入るため、"
            "summary には利用者が読んで分かる要約を書く。返り値は {ticket_id, status}。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "department": {"type": "string", "enum": ["it", "accounting", "hr", "general"]},
                "summary": {"type": "string", "description": "60字以内の件名"},
                "detail": {"type": "string", "description": "状況・試したこと・エラー文"},
                "priority": {"type": "string", "enum": ["low", "normal", "high"]},
            },
            "required": ["department", "summary", "detail", "priority"],
        },
    },
]


# ---- ツールの実装（ここはあなたの社内システムにつなぐ） ----
def search_faq(query: str, category: str | None = None) -> list[dict]:
    # 実際は社内検索API・全文検索・ベクトル検索などを呼ぶ。読み取り専用の資格情報で
    return [{"id": "faq-102", "title": "VPNに接続できない", "body": "1. 端末を再起動...", "url": "https://intra.example.com/faq/102"}]


def create_ticket(department: str, summary: str, detail: str, priority: str, user_id: str) -> dict:
    # 実際はチケット管理システムのAPIを呼ぶ。起票者は必ずログイン中の本人にする
    return {"ticket_id": "T-2026-0001", "status": "open"}


def ask_user_approval(tool_input: dict) -> bool:
    # 実運用ではUIに確認ダイアログを出す。ここではCLIで代用
    print("\n以下の内容で起票します:\n" + json.dumps(tool_input, ensure_ascii=False, indent=2))
    return input("よろしいですか？ [y/N] ").strip().lower() == "y"


class ToolError(Exception):
    """モデルに伝えてよいエラー（is_error で返す）"""


def run_tool(name: str, tool_input: dict, user_id: str) -> str:
    if name == "search_faq":
        return json.dumps(search_faq(**tool_input), ensure_ascii=False)
    if name == "create_ticket":
        # 人の承認ゲート：モデルの判断だけでは書き込ませない
        if not ask_user_approval(tool_input):
            raise ToolError("利用者が起票を承認しませんでした。起票せずに、他にできることを案内してください。")
        return json.dumps(create_ticket(**tool_input, user_id=user_id), ensure_ascii=False)
    raise ToolError(f"未知のツールです: {name}")


def run_agent(user_message: str, user_id: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL,
            max_tokens=16000,
            system=SYSTEM,
            tools=TOOLS,
            messages=messages,
        )
        log.info("turn=%d stop=%s in=%d out=%d", turn, response.stop_reason,
                 response.usage.input_tokens, response.usage.output_tokens)

        if response.stop_reason == "end_turn":
            return "".join(b.text for b in response.content if b.type == "text")
        if response.stop_reason == "refusal":
            return "この内容にはお答えできません。担当部署に直接お問い合わせください。"
        if response.stop_reason == "max_tokens":
            raise RuntimeError("出力が上限に達しました。max_tokens か指示を見直してください")
        if response.stop_reason != "tool_use":
            raise RuntimeError(f"想定外の stop_reason: {response.stop_reason}")

        # assistant の応答は content ごと（thinking ブロックも含めて）そのまま戻す
        messages.append({"role": "assistant", "content": response.content})

        # 並列呼び出しに備え、全 tool_use の結果を1つの user メッセージにまとめる
        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            log.info("tool=%s input=%s", block.name, json.dumps(block.input, ensure_ascii=False))
            try:
                content = run_tool(block.name, block.input, user_id)
                tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": content})
            except ToolError as e:
                tool_results.append({"type": "tool_result", "tool_use_id": block.id,
                                     "content": str(e), "is_error": True})
            except Exception:
                log.exception("tool failed")
                # 内部の例外メッセージ（接続先・スタックトレース）はモデルに渡さない
                tool_results.append({"type": "tool_result", "tool_use_id": block.id,
                                     "content": "内部エラーで実行できませんでした。", "is_error": True})

        messages.append({"role": "user", "content": tool_results})

    return "処理が長くなったため中断しました。お手数ですがチケットを直接起票してください。"


if __name__ == "__main__":
    print(run_agent("VPNがつながらず、再起動しても直りません。どうすれば？", user_id="u-1234"))

import json
import anthropic
from anthropic import beta_tool

client = anthropic.Anthropic()


@beta_tool
def search_faq(query: str) -> str:
    """社内FAQを全文検索し、関連度の高い順に最大5件をJSONで返す。回答前に必ず使う。

    Args:
        query: 検索キーワード。製品名・エラー文・手続き名など
    """
    results = [{"id": "faq-102", "title": "VPNに接続できない", "body": "1. 端末を再起動..."}]
    return json.dumps(results, ensure_ascii=False)


@beta_tool
def create_ticket(department: str, summary: str, detail: str) -> str:
    """担当部署へのチケットを起票する。FAQで解決しない場合のみ使う。

    Args:
        department: it / accounting / hr / general のいずれか
        summary: 60字以内の件名
        detail: 状況・試したこと・エラー文
    """
    # 承認ゲートはツール関数の中に置く
    if input(f"「{summary}」で起票しますか？ [y/N] ").strip().lower() != "y":
        return "利用者が起票を承認しませんでした。起票せずに案内してください。"
    return json.dumps({"ticket_id": "T-2026-0002", "status": "open"})


runner = client.beta.messages.tool_runner(
    model="claude-opus-5",
    max_tokens=16000,
    system="あなたは社内ヘルプデスク担当です。必ずFAQを確認してから答えます。",
    tools=[search_faq, create_ticket],
    messages=[{"role": "user", "content": "経費精算の締め日を過ぎてしまいました。どうすれば？"}],
)

# 1ターンごとの assistant メッセージが順に得られる（ログや進捗表示に使える）
for message in runner:
    for block in message.content:
        if block.type == "text":
            print(block.text)

import Anthropic from "@anthropic-ai/sdk";
import { betaZodTool } from "@anthropic-ai/sdk/helpers/beta/zod";
import { z } from "zod";

const client = new Anthropic();

const searchFaq = betaZodTool({
  name: "search_faq",
  description: "社内FAQを全文検索し、関連度の高い順に最大5件をJSONで返す。回答前に必ず使う。",
  inputSchema: z.object({
    query: z.string().describe("検索キーワード。製品名・エラー文・手続き名など"),
  }),
  run: async ({ query }) => {
    const results = [{ id: "faq-102", title: "VPNに接続できない", body: "1. 端末を再起動..." }];
    return JSON.stringify(results);
  },
});

const finalMessage = await client.beta.messages.toolRunner({
  model: "claude-opus-5",
  max_tokens: 16000,
  tools: [searchFaq],
  messages: [{ role: "user", content: "VPNにつながりません" }],
});

for (const block of finalMessage.content) {
  if (block.type === "text") console.log(block.text);
}

CREATE_TICKET = {
    "name": "create_ticket",
    "description": "担当部署へのチケットを起票する。FAQで解決しない場合のみ使う。",
    "strict": True,  # tool_use.input がスキーマ通りであることを保証
    "input_schema": {
        "type": "object",
        "properties": {
            "department": {"type": "string", "enum": ["it", "accounting", "hr", "general"]},
            "summary": {"type": "string"},
            "priority": {"type": "string", "enum": ["low", "normal", "high"]},
        },
        "required": ["department", "summary", "priority"],
        "additionalProperties": False,
    },
}

from typing import Literal
import anthropic
from pydantic import BaseModel

client = anthropic.Anthropic()


class Triage(BaseModel):
    department: Literal["it", "accounting", "hr", "general"]
    urgency: Literal["low", "normal", "high"]
    needs_human: bool
    summary: str


response = client.messages.parse(
    model="claude-opus-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "来週入社の新人のPCがまだ届いていません。至急確認したいです。"}],
    output_format=Triage,
)

triage = response.parsed_output  # 検証済みの Triage インスタンス
print(triage.department, triage.urgency, triage.needs_human)

import json
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "来週入社の新人のPCがまだ届いていません。"}],
    output_config={
        "format": {
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "department": {"type": "string", "enum": ["it", "accounting", "hr", "general"]},
                    "urgency": {"type": "string", "enum": ["low", "normal", "high"]},
                },
                "required": ["department", "urgency"],
                "additionalProperties": False,
            },
        }
    },
)
text = next(b.text for b in response.content if b.type == "text")
data = json.loads(text)

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    thinking={"type": "adaptive"},
    output_config={"effort": "medium"},  # low / medium / high / xhigh / max
    tools=TOOLS,
    messages=messages,
)

with client.messages.stream(
    model="claude-opus-5",
    max_tokens=16000,
    system=SYSTEM,
    messages=[{"role": "user", "content": "新しいPCの初期設定手順をまとめて"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()  # usage や stop_reason はここから

print("\n", final.stop_reason, final.usage.output_tokens)

const stream = client.messages
  .stream({
    model: "claude-opus-5",
    max_tokens: 16000,
    messages: [{ role: "user", content: "新しいPCの初期設定手順をまとめて" }],
  })
  .on("text", (text) => process.stdout.write(text));

const final = await stream.finalMessage();
console.log("\n", final.stop_reason);

messages = [{"role": "user", "content": "今月発表された主要なクラウド障害の公式報告をまとめて"}]

while True:
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=16000,
        tools=[{"type": "web_search_20260209", "name": "web_search", "max_uses": 5}],
        messages=messages,
    )
    # サーバー側の処理が長いと pause_turn で一度戻る。そのまま送り返せば続きから再開
    if response.stop_reason == "pause_turn":
        messages.append({"role": "assistant", "content": response.content})
        continue
    break

for block in response.content:
    if block.type == "text":
        print(block.text)

response = client.beta.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    betas=["mcp-client-2025-11-20"],
    mcp_servers=[{"type": "url", "url": "https://example.com/mcp", "name": "my-mcp"}],
    tools=[{"type": "mcp_toolset", "mcp_server_name": "my-mcp"}],  # mcp_servers とセットで必須
    messages=[{"role": "user", "content": "今週の未対応チケットを一覧にして"}],
)

import anthropic

client = anthropic.Anthropic(max_retries=4, timeout=120.0)


def ask(messages: list[dict]) -> str | None:
    try:
        response = client.messages.create(
            model="claude-opus-5", max_tokens=16000, messages=messages,
        )
    except anthropic.RateLimitError:
        # 429：自動リトライ後も超過。キューに戻して後で再実行、利用者には混雑表示
        return None
    except anthropic.BadRequestError as e:
        # 400：リクエストの形が不正。リトライしても直らないのでログを見て修正
        raise RuntimeError(f"不正なリクエスト: {e.message}") from e
    except anthropic.APIStatusError as e:
        # その他のHTTPエラー（5xx の過負荷など）
        raise RuntimeError(f"APIエラー status={e.status_code}") from e
    except anthropic.APIConnectionError as e:
        # ネットワーク到達不可・タイムアウト
        raise RuntimeError("Claude APIに接続できません") from e
    return "".join(b.text for b in response.content if b.type == "text")

# pip install claude-agent-sdk
import anyio
from claude_agent_sdk import ClaudeAgentOptions, query


async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Grep", "Glob"])  # 読み取り系だけ許可
    async for message in query(prompt="docs/ 配下の手順書から、VPN関連の記述を要約して", options=options):
        print(message)


anyio.run(main)