claude-code-sdk

Claude Code SDK の統合

概要

Claude Code SDK (claude-code-sdk) は、Anthropic が公式に提供する Python パッケージで、アプリケーションに Claude AI エージェント機能を統合するために使用します。OAuth 認証（claude login 経由）と API Key 認証をサポートしており、開発者は Claude Max/Pro のサブスクリプションまたは API の従量課金プランを利用できます。この SDK は AI を活用したアプリケーションを構築するための重要なツールであり、特にコード生成、ファイル操作、またはマルチターンの対話が必要なシナリオに適しています。

主要な概念

認証方式

説明: Claude Code SDK は 2 つの認証方式をサポートしています 主な機能:

• OAuth 認証: claude login の実行後に ~/.claude/ に保存される認証情報
• API Key 認証: ANTHROPIC_API_KEY 環境変数経由

ユースケース:

• OAuth: 個人開発、Max/Pro サブスクリプションを持つユーザー
• API Key: 本番環境、従量課金が必要なサービス

SDK のコア API

説明: query() 関数が主要な非同期インターフェースです 主な機能:

• 非同期イテレーターがメッセージを返す
• ストリーミングと完全なレスポンスをサポート
• ツールの権限と実行制限を設定可能

ユースケース: コード生成、ファイル操作、調査タスク、読み取り専用の分析

ツールの種類 (ToolType)

説明: SDK はさまざまな組み込みツールをサポートしています 主な機能:

• READ、WRITE、EDIT - ファイル操作
• BASH - ターミナルコマンドの実行
• GLOB、GREP - ファイル検索
• WEB_FETCH - ウェブページの取得

ベストプラクティス

1. 二重の認証チェック
   • OAuth と API Key の両方をサポート
   • OAuth を優先して使用（Max/Pro のクォータを無料で使用）
   • API Key をフォールバックとして使用

def check_auth_status() -> dict:
    status = {
        'api_key_set': bool(os.environ.get('ANTHROPIC_API_KEY')),
        'oauth_logged_in': False,
    }

    # 檢查 OAuth 登入狀態
    claude_dir = os.path.expanduser('~/.claude')
    if os.path.exists(claude_dir):
        auth_files = ['credentials.json', 'settings.json', '.credentials.json']
        for auth_file in auth_files:
            if os.path.exists(os.path.join(claude_dir, auth_file)):
                status['oauth_logged_in'] = True
                break

    # 只需要其中一種認證
    status['auth_available'] = status['api_key_set'] or status['oauth_logged_in']
    return status

2. バッファサイズの増加
   • デフォルトのバッファは小さすぎて、大きなレスポンスを処理するとエラーになる
   • 50MB に増やすことを推奨

try:
    from claude_code_sdk._internal.transport import subprocess_cli
    subprocess_cli._MAX_BUFFER_SIZE = 50 * 1024 * 1024  # 50MB
except Exception as e:
    print(f"Failed to patch SDK buffer size: {e}")

3. Singleton パターンの使用
   • 重複した初期化を避ける
   • 状態と構成を共有

_claude_service: Optional['ClaudeCodeService'] = None

def get_claude_code() -> 'ClaudeCodeService':
    global _claude_service
    if _claude_service is None:
        _claude_service = ClaudeCodeService()
    return _claude_service

4. デフォルトの構成パターン
   • ユースケースに応じてツールの権限を制限
   • リスクを低減し、安全性を向上

@dataclass
class AgentConfig:
    allowed_tools: List[str]
    max_turns: int = 30
    cwd: Optional[str] = None

    @classmethod
    def coding(cls) -> 'AgentConfig':
        """開發模式：完整權限"""
        return cls(
            allowed_tools=['Read', 'Write', 'Edit', 'Bash', 'Glob', 'Grep'],
            max_turns=50
        )

    @classmethod
    def research(cls) -> 'AgentConfig':
        """研究模式：可上網"""
        return cls(
            allowed_tools=['Read', 'Glob', 'Grep', 'WebFetch'],
            max_turns=20
        )

    @classmethod
    def readonly(cls) -> 'AgentConfig':
        """唯讀模式：最安全"""
        return cls(
            allowed_tools=['Read', 'Glob', 'Grep'],
            max_turns=10
        )

よくある落とし穴

落とし穴 1: フロントエンドで直接 SDK を呼び出す

• 問題: claude-code-sdk は Python パッケージであり、ブラウザーでは動作しません
• 解決策: バックエンド API サービスを構築する必要があります
• 例:

# ❌ 錯誤：嘗試在前端使用
# JavaScript 無法直接使用 Python SDK

# ✅ 正確：建立 FastAPI 後端
from fastapi import FastAPI
from claude_code_sdk import query

app = FastAPI()

@app.post("/api/generate")
async def generate(prompt: str):
    result = []
    async for msg in query(prompt=prompt):
        result.append(msg)
    return {"result": result}

落とし穴 2: 非同期処理を忘れる

• 問題: SDK は非同期であり、await を忘れるとエラーになります
• 解決策: async/await または asyncio を使用する

# ❌ 錯誤
def run_sync():
    for msg in query(prompt="Hello"):  # TypeError
        print(msg)

# ✅ 正確
async def run_async():
    async for msg in query(prompt="Hello"):
        print(msg)

落とし穴 3: Claude API と Claude Code SDK の混同

• 問題:
  • Claude API (anthropic パッケージ) - 直接 HTTP API、API Key が必要
  • Claude Code SDK (claude-code-sdk) - エージェント機能、OAuth をサポート
• 解決策: 要件に応じて正しいパッケージを選択する

特性	Claude API	Claude Code SDK
認証	API Key のみ	OAuth + API Key
機能	純粋な対話	エージェント (ファイル/ターミナル操作)
課金	API 使用量に基づく課金	Max/Pro サブスクリプションで利用可能
パッケージ	anthropic	claude-code-sdk

パターンとアンチパターン

パターン (これをすること)

完全なサービスラッパー:

from dataclasses import dataclass, field
from typing import AsyncIterator, Optional, List, Dict, Any
from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, TextBlock

@dataclass
class ClaudeCodeService:
    """Claude Code 服務封裝"""

    async def run(self, prompt: str, config: Optional[AgentConfig] = None) -> Dict[str, Any]:
        """執行並收集完整結果"""
        options = ClaudeCodeOptions(
            allowed_tools=config.allowed_tools if config else None,
            max_turns=config.max_turns if config else 30,
            cwd=config.cwd
        )

        result_texts = []
        tool_calls = []

        async for message in query(prompt=prompt, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        result_texts.append(block.text)

        return {
            'status': 'success',
            'result': '\n'.join(result_texts),
            'tool_calls': tool_calls
        }

    async def stream(self, prompt: str, config: Optional[AgentConfig] = None) -> AsyncIterator[dict]:
        """串流執行，即時回傳"""
        options = ClaudeCodeOptions(
            allowed_tools=config.allowed_tools if config else None,
            max_turns=config.max_turns if config else 30
        )

        async for message in query(prompt=prompt, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        yield {'type': 'text', 'content': block.text}

FastAPI ストリーミングエンドポイント:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/api/agent/stream")
async def agent_stream(prompt: str):
    async def event_generator():
        service = ClaudeCodeService()
        async for msg in service.stream(prompt):
            yield f"data: {json.dumps(msg)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

アンチパターン (これは避けること)

# ❌ 不要在每次請求都建立新服務
@app.post("/api/generate")
async def bad_generate(prompt: str):
    service = ClaudeCodeService()  # 每次都新建
    return await service.run(prompt)

# ❌ 不要忽略錯誤處理
async def bad_run(prompt: str):
    async for msg in query(prompt=prompt):  # 可能拋出異常
        print(msg)

# ❌ 不要給予過多權限
config = AgentConfig(
    allowed_tools=['Read', 'Write', 'Edit', 'Bash', 'WebFetch'],  # 全開
    max_turns=100  # 太多
)

ツールとリソース

ツール	目的	リンク
claude-code-sdk	公式 Python SDK	pip install claude-code-sdk
Claude Code CLI	コマンドラインツール	npm install -g @anthropic-ai/claude-code
FastAPI	バックエンドフレームワーク	fastapi.tiangolo.com

判断ガイド

Claude Code SDK を使用する場合:

• エージェント機能（ファイル操作、ターミナルコマンド）が必要
• API の従量課金ではなく Max/Pro サブスクリプションを使用したい
• マルチターンの対話が必要な AI アプリケーションを構築する
• ストリーミングによるリアルタイムレスポンスが必要

代替案を検討する場合:

• 簡単な対話のみが必要 → anthropic パッケージを使用
• 純粋なフロントエンドアプリケーション → anthropic + anthropic-dangerous-direct-browser-access を使用
• エージェント機能が不要 → 標準の Claude API を使用

例

例 1: 完全なバックエンドサービス

コンテキスト: Claude Max サブスクリプションを利用できるバックエンド API を構築する

実装:

# backend/main.py
import os
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json

from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, TextBlock

app = FastAPI()

# CORS 設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

class GenerateRequest(BaseModel):
    prompt: str
    max_turns: int = 30

def check_auth():
    """檢查認證狀態"""
    api_key = bool(os.environ.get('ANTHROPIC_API_KEY'))
    oauth = os.path.exists(os.path.expanduser('~/.claude/credentials.json'))
    return api_key or oauth

@app.get("/api/health")
async def health():
    return {
        "status": "ok",
        "auth_available": check_auth()
    }

@app.post("/api/generate")
async def generate(request: GenerateRequest):
    if not check_auth():
        raise HTTPException(401, "請先執行 claude login 或設置 ANTHROPIC_API_KEY")

    options = ClaudeCodeOptions(max_turns=request.max_turns)
    results = []

    async for message in query(prompt=request.prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    results.append(block.text)

    return {"result": "\n".join(results)}

@app.post("/api/stream")
async def stream(request: GenerateRequest):
    if not check_auth():
        raise HTTPException(401, "未認證")

    async def event_generator():
        options = ClaudeCodeOptions(max_turns=request.max_turns)
        async for message in query(prompt=request.prompt, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        yield f"data: {json.dumps({'text': block.text})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

requirements.txt:

fastapi>=0.100.0
uvicorn>=0.23.0
claude-code-sdk>=0.1.0
python-dotenv>=1.0.0

説明: このバックエンドは OAuth または API Key 認証を自動的に検出し、通常のリクエストとストリーミングレスポンスをサポートします。

例 2: フロントエンドの統合

コンテキスト: フロントエンドがバックエンド API を呼び出す

実装:

// 一般請求
async function generate(prompt) {
    const response = await fetch('http://localhost:8000/api/generate', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });
    const data = await response.json();
    return data.result;
}

// 串流請求
async function streamGenerate(prompt, onChunk) {
    const response = await fetch('http://localhost:8000/api/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');

        for (const line of lines) {
            if (line.startsWith('data: ') && line !== 'data: [DONE]') {
                const data = JSON.parse(line.slice(6));
                onChunk(data.text);
            }
        }
    }
}

例 3: Docker デプロイ

コンテキスト: Claude Code CLI を含むコンテナ化されたデプロイ

実装:

# Dockerfile
FROM python:3.11-slim

# 安裝 Node.js (Claude Code CLI 需要)
RUN apt-get update && apt-get install -y curl && \
    curl -fsSL https://deb.nodesource.com/setup_20.x | bash - && \
    apt-get install -y nodejs && \
    rm -rf /var/lib/apt/lists/*

# 安裝 Claude Code CLI
RUN npm install -g @anthropic-ai/claude-code

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'
services:
  backend:
    build: .
    ports:
      - "8000:8000"
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    volumes:
      - ~/.claude:/root/.claude:ro  # 掛載 OAuth 憑證

関連スキル

• [[api-design]] - REST API エンドポイントの設計
• [[backend]] - Python/FastAPI バックエンド開発
• [[devops-cicd]] - Docker デプロイと CI/CD
• [[security-practices]] - API 認証とセキュリティ

参考資料

• Claude Code SDK GitHub
• Anthropic API Documentation
• FastAPI Documentation