Claude 実践テクニック集【後編】— 本番運用・マルチエージェント・収益化の上級テクニック

前編で身につけた基本テクニックを基に、このドキュメントではプロダクション環境での実装、複雑なシステム設計、ビジネス化戦略まで、実務レベルの高度なテクニックを解説します。

マルチエージェントシステムの設計から、本番グレードの API 運用、MCP サーバーの自作、そして SaaS を通じた収益化パイプラインの構築まで——Claude を使って実際にビジネス価値を生み出すための知識を、具体的なコード例とともに詳しく説明します。

取り組みの背景 — 後編で扱うトピック

このセクションでは、前編の基礎知識をさらに深掘りし、実際のプロダクション環境で求められるより複雑な実装パターンを扱います。

後編の目標

複雑なシステムの構築: 単独のエージェントではなく、複数のエージェントを連携させるアーキテクチャの設計
本番運用の知見: スケーリング、エラーハンドリング、コスト削減のための実装パターン
カスタム機能の拡張: MCP サーバーを自作して、プロジェクト固有のツールを実装
ビジネスの収益化: Claude API を活用したビジネスモデルの実装と実例

マルチエージェントシステムの設計と実装

複雑なタスクを効率的に処理するには、複数のエージェントを役割分担させる必要があります。

Agent SDK でのオーケストレーター/ワーカー構成

オーケストレーター（統括）エージェントが、複数のワーカーエージェントを制御し、タスクを分配するパターンです。

from anthropic import Anthropic
 
client = Anthropic()
 
# ワーカーエージェント: コンテンツ執筆
def writer_agent(topic: str) -> str:
    """指定されたトピックについて記事を執筆"""
 
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2048,
        system="あなたはプロのテクニカルライターです。正確で分かりやすい記事を執筆してください。",
        messages=[
            {
                "role": "user",
                "content": f"以下のトピックについて、1500 語の技術記事を執筆してください: {topic}"
            }
        ]
    )
    return response.content[0].text
 
# ワーカーエージェント: コンテンツレビュー
def reviewer_agent(content: str) -> dict:
    """執筆されたコンテンツをレビュー"""
 
    import json
 
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system="あなたは編集者です。技術的正確性、文章品質、構成を評価してください。",
        messages=[
            {
                "role": "user",
                "content": f"""以下の記事をレビューして、JSON 形式で評価を返してください：
{{
  "score": 1-10,
  "issues": ["問題A", "問題B"],
  "suggestions": ["改善案A", "改善案B"]
}}
 
記事:
{content}"""
            }
        ]
    )
 
    return json.loads(response.content[0].text)
 
# ワーカーエージェント: コンテンツ改善
def editor_agent(content: str, feedback: str) -> str:
    """フィードバックに基づいてコンテンツを改善"""
 
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2048,
        system="あなたはプロの編集者です。フィードバックに基づいて記事を改善してください。",
        messages=[
            {
                "role": "user",
                "content": f"""以下のフィードバックに基づいて、記事を改善してください。
 
フィードバック:
{feedback}
 
元の記事:
{content}"""
            }
        ]
    )
    return response.content[0].text
 
# オーケストレーター: 全体の流れを制御
def orchestrator_agent(topic: str, max_iterations: int = 2) -> str:
    """複数のエージェントを調整して、高品質な記事を完成させる"""
 
    print(f"▶ トピック: {topic}")
    print(f"▶ 最大改善ループ: {max_iterations}")
 
    # Step 1: 初稿を執筆
    print("\n[Step 1] ライターエージェントが初稿を作成中...")
    content = writer_agent(topic)
 
    # Step 2-N: レビュー・改善のループ
    for iteration in range(max_iterations):
        print(f"\n[Step {iteration + 2}] レビュアーエージェントが評価中...")
        review = reviewer_agent(content)
 
        if review["score"] >= 8:
            print(f"✓ スコア {review['score']}/10 で目標達成！")
            break
 
        print(f"✗ スコア {review['score']}/10")
        print(f"   問題: {', '.join(review['issues'][:2])}")
 
        print(f"\n[Step {iteration + 3}] エディターエージェントが改善中...")
        feedback = f"""
問題点: {', '.join(review['issues'])}
改善案: {', '.join(review['suggestions'])}
"""
        content = editor_agent(content, feedback)
 
    return content
 
# 実行例
final_article = orchestrator_agent("Claude API での Cost Optimization")
print("\n=" * 50)
print(final_article[:500] + "...")

Claude Code の Task ツールによるサブエージェント並列実行

Claude Code の Task ツールを使用して、複数の独立したタスクを並列実行できます。

import asyncio
from anthropic import Anthropic
 
client = Anthropic()
 
async def parallel_data_processing():
    """複数のデータ処理タスクを並列実行"""
 
    # 処理対象のデータセット
    datasets = [
        {"id": 1, "name": "Sales Q1", "size": 50000},
        {"id": 2, "name": "Sales Q2", "size": 75000},
        {"id": 3, "name": "Sales Q3", "size": 60000},
    ]
 
    # 各データセットを処理するタスク
    async def process_dataset(dataset):
        print(f"▶ {dataset['name']} を処理中...")
 
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            messages=[
                {
                    "role": "user",
                    "content": f"""
{dataset['name']} のデータセット（{dataset['size']} 行）について、
以下の分析を JSON 形式で提供してください：
- 推定売上高
- トップ3 カテゴリー
- 成長率
"""
                }
            ]
        )
 
        import json
        result = json.loads(response.content[0].text)
        result["dataset_id"] = dataset["id"]
 
        print(f"✓ {dataset['name']} の処理完了")
        return result
 
    # 全データセットを並列処理
    tasks = [process_dataset(ds) for ds in datasets]
    results = await asyncio.gather(*tasks)
 
    # 結果を集計
    print("\n[集計結果]")
    total_revenue = sum(r.get("estimated_revenue", 0) for r in results)
    print(f"総売上: ${total_revenue:,}")
 
    return results
 
# 非同期実行（実際の使用時）
# results = asyncio.run(parallel_data_processing())

エラーリカバリーとリトライ戦略

本番環境では、一時的な障害への対応が重要です。

import time
from anthropic import APIError
 
def call_api_with_retry(
    messages: list,
    max_retries: int = 3,
    initial_wait: int = 1
) -> str:
    """エラー処理とリトライロジックを備えた API 呼び出し"""
 
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                messages=messages
            )
            return response.content[0].text
 
        except APIError as e:
            if attempt < max_retries - 1:
                wait_time = initial_wait * (2 ** attempt)  # exponential backoff
                print(f"⚠ エラー: {e.message}")
                print(f"  {wait_time} 秒後に再試行します... (試行 {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                print(f"✗ {max_retries} 回の試行後も失敗しました")
                raise
 
def call_api_with_fallback(messages: list, fallback_model: str = "claude-3-haiku-20250307") -> str:
    """メインモデル失敗時にフォールバックモデルを使用"""
 
    try:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=messages
        )
        return response.content[0].text
 
    except APIError:
        print(f"⚠ メインモデルが利用不可。フォールバックモデルを使用します")
 
        response = client.messages.create(
            model=fallback_model,
            max_tokens=512,
            messages=messages
        )
        return response.content[0].text

本番グレードの API 運用パターン

プロダクション環境では、スケーリング、コスト最適化、信頼性が重要になります。

ストリーミング × ツール使用の並列呼び出し

高速レスポンスとツール統合を同時に実現するパターンです。

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});
 
const tools = [
  {
    name: "get_weather",
    description: "指定された地域の天気を取得",
    input_schema: {
      type: "object" as const,
      properties: {
        location: {
          type: "string",
          description: "地域名",
        },
      },
      required: ["location"],
    },
  },
  {
    name: "get_stock_price",
    description: "株式の現在価格を取得",
    input_schema: {
      type: "object" as const,
      properties: {
        symbol: {
          type: "string",
          description: "ティッカーシンボル（例: AAPL）",
        },
      },
      required: ["symbol"],
    },
  },
];
 
async function executeWithTools(userQuery: string): Promise<void> {
  console.log(`ユーザー: ${userQuery}\n`);
 
  let messages: Anthropic.Messages.MessageParam[] = [
    { role: "user", content: userQuery },
  ];
 
  // Tool use ループ
  while (true) {
    const response = await client.messages.create({
      model: "claude-3-5-sonnet-20241022",
      max_tokens: 1024,
      tools: tools,
      messages: messages,
    });
 
    // ストリーミングテキストを出力
    for (const block of response.content) {
      if (block.type === "text") {
        process.stdout.write(block.text);
      }
    }
 
    // Tool use が必要か確認
    if (response.stop_reason !== "tool_use") {
      console.log("\n");
      break;
    }
 
    // Tool call を処理
    const toolResults: Anthropic.Messages.ToolResultBlockParam[] = [];
 
    for (const block of response.content) {
      if (block.type === "tool_use") {
        console.log(`\n[ツール実行] ${block.name}(${JSON.stringify(block.input)})`);
 
        // 実際のツール実行をシミュレート
        let result = "";
        if (block.name === "get_weather") {
          result = `${block.input.location} の天気: 晴れ、25°C`;
        } else if (block.name === "get_stock_price") {
          result = `${block.input.symbol} の現在価格: $150.25`;
        }
 
        toolResults.push({
          type: "tool_result",
          tool_use_id: block.id,
          content: result,
        });
      }
    }
 
    // Tool 結果をメッセージに追加
    messages.push({ role: "assistant", content: response.content });
    messages.push({
      role: "user",
      content: toolResults,
    });
  }
}
 
// 実行例
executeWithTools(
  "東京と大阪の天気とアップルの株価を教えてください"
);

コスト最適化（キャッシュ・バッチ処理・モデル使い分け）

API コストを削減するための複数のテクニックです。

from anthropic import Anthropic
 
client = Anthropic()
 
# テクニック1: プロンプトキャッシング
def cached_analysis(large_context: str, query: str) -> str:
    """大きなコンテキストをキャッシュして再利用"""
 
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=512,
        system=[
            {
                "type": "text",
                "text": "あなたは優秀な分析者です。",
            },
            {
                "type": "text",
                "text": f"以下の大規模な参考資料を分析対象にします:\n{large_context}",
                "cache_control": {"type": "ephemeral"},  # キャッシュ指定
            },
        ],
        messages=[{"role": "user", "content": query}],
    )
 
    return response.content[0].text
 
# テクニック2: バッチ処理（複数リクエストを一度に送信）
def batch_processing(queries: list[str]) -> list[str]:
    """複数のクエリを効率的に処理"""
 
    results = []
 
    for query in queries:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=256,
            messages=[{"role": "user", "content": query}],
        )
        results.append(response.content[0].text)
 
    return results
 
# テクニック3: モデル選択の最適化
def choose_optimal_model(task_complexity: str) -> str:
    """タスクの複雑度に応じて適切なモデルを選択"""
 
    complexity_to_model = {
        "simple": "claude-3-haiku-20250307",  # 低コスト
        "medium": "claude-3-5-sonnet-20241022",  # バランス型
        "complex": "claude-3-opus-20250219",  # 最高精度
    }
 
    return complexity_to_model.get(task_complexity, "claude-3-5-sonnet-20241022")
 
# テクニック4: コスト推定
def estimate_api_cost(input_tokens: int, output_tokens: int, model: str) -> float:
    """API 呼び出しのコストを推定"""
 
    pricing = {
        "claude-3-5-sonnet-20241022": {
            "input": 0.003,  # $3 per 1M input tokens
            "output": 0.015,  # $15 per 1M output tokens
        },
        "claude-3-opus-20250219": {
            "input": 0.015,
            "output": 0.075,
        },
        "claude-3-haiku-20250307": {
            "input": 0.00080,
            "output": 0.004,
        },
    }
 
    rates = pricing.get(model, pricing["claude-3-5-sonnet-20241022"])
 
    input_cost = (input_tokens / 1_000_000) * rates["input"]
    output_cost = (output_tokens / 1_000_000) * rates["output"]
 
    return input_cost + output_cost
 
# 使用例
print(f"推定コスト: ${estimate_api_cost(5000, 1000, 'claude-3-5-sonnet-20241022'):.6f}")

レート制限とグレースフルデグラデーション

API のレート制限に対応するパターンです。

from anthropic import RateLimitError
import time
from datetime import datetime
 
class RateLimitManager:
    """レート制限を管理するクラス"""
 
    def __init__(self, max_requests_per_minute: int = 60):
        self.max_requests = max_requests_per_minute
        self.request_timestamps = []
 
    def can_make_request(self) -> bool:
        """現在リクエストが可能か確認"""
 
        now = datetime.now()
        # 1 分以上前のリクエストを削除
        self.request_timestamps = [
            ts for ts in self.request_timestamps
            if (now - ts).seconds < 60
        ]
 
        return len(self.request_timestamps) < self.max_requests
 
    def wait_if_needed(self) -> None:
        """必要に応じて待機"""
 
        while not self.can_make_request():
            sleep_time = 0.5
            print(f"⚠ レート制限に達しています。{sleep_time} 秒待機します...")
            time.sleep(sleep_time)
 
        self.request_timestamps.append(datetime.now())
 
def call_with_rate_limit_handling(
    messages: list,
    rate_limiter: RateLimitManager
) -> str:
    """レート制限を考慮した API 呼び出し"""
 
    rate_limiter.wait_if_needed()
 
    try:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            messages=messages,
        )
        return response.content[0].text
 
    except RateLimitError:
        print("⚠ レート制限エラー。30秒待機して再試行します...")
        time.sleep(30)
 
        # フォールバック: より低速モデルを使用
        response = client.messages.create(
            model="claude-3-haiku-20250307",
            max_tokens=256,
            messages=messages,
        )
        return response.content[0].text

MCP サーバーを自作する

MCP（Model Context Protocol）サーバーを自作することで、Claude に対してカスタム機能を提供できます。

サーバーの基本構造（TypeScript）

import {
  StdioServerTransport,
  Server,
} from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  Tool,
  TextContent,
  CallToolRequest,
  CallToolRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
 
// MCP サーバーの初期化
const server = new Server(
  {
    name: "custom-tools-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);
 
// ツール定義
const tools: Tool[] = [
  {
    name: "analyze_csv",
    description: "CSV ファイルを分析して統計情報を返す",
    inputSchema: {
      type: "object",
      properties: {
        filepath: {
          type: "string",
          description: "CSV ファイルのパス",
        },
        operation: {
          type: "string",
          enum: ["summary", "describe", "correlate"],
          description: "実行する操作の種類",
        },
      },
      required: ["filepath", "operation"],
    },
  },
  {
    name: "query_database",
    description: "データベースに SQL クエリを実行",
    inputSchema: {
      type: "object",
      properties: {
        query: {
          type: "string",
          description: "実行する SQL クエリ",
        },
        limit: {
          type: "number",
          description: "返す最大行数",
        },
      },
      required: ["query"],
    },
  },
];
 
// ツール実装
async function handleToolCall(
  request: CallToolRequest
): Promise<TextContent> {
  const { name, arguments: args } = request;
 
  if (name === "analyze_csv") {
    const { filepath, operation } = args as {
      filepath: string;
      operation: string;
    };
 
    // CSV 分析ロジック
    let result = "";
 
    if (operation === "summary") {
      result = `${filepath} のサマリー: 1000 行、50 列`;
    } else if (operation === "describe") {
      result = `列情報: id (int), name (string), score (float)`;
    }
 
    return {
      type: "text",
      text: result,
    };
  }
 
  if (name === "query_database") {
    const { query, limit = 10 } = args as {
      query: string;
      limit?: number;
    };
 
    // データベースクエリロジック
    const result = `クエリ実行: ${query} (最大 ${limit} 行)`;
 
    return {
      type: "text",
      text: result,
    };
  }
 
  throw new Error(`Unknown tool: ${name}`);
}
 
// サーバーセットアップ
server.setRequestHandler(CallToolRequestSchema, handleToolCall);
 
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server started");
}
 
main().catch(console.error);

ツール定義とバリデーション（Zod）

入力値のバリデーションを厳密に行うパターンです。

import { z } from "zod";
 
// Zod スキーマでバリデーション定義
const AnalyzeCSVSchema = z.object({
  filepath: z
    .string()
    .min(1, "ファイルパスは必須です")
    .regex(/\.csv$/, "CSV ファイルのみ対応"),
  operation: z.enum(["summary", "describe", "correlate"]),
  includeHeaders: z.boolean().optional().default(true),
});
 
type AnalyzeCSVInput = z.infer<typeof AnalyzeCSVSchema>;
 
// バリデーション付きハンドラ
function analyzeCSV(input: unknown): string {
  try {
    const validated: AnalyzeCSVInput = AnalyzeCSVSchema.parse(input);
 
    // ここからビジネスロジック
    const { filepath, operation, includeHeaders } = validated;
 
    if (operation === "summary") {
      return `${filepath} をサマリー表示 (ヘッダー: ${includeHeaders})`;
    }
 
    return `処理完了: ${operation}`;
  } catch (error) {
    if (error instanceof z.ZodError) {
      return `バリデーションエラー: ${error.errors[0].message}`;
    }
 
    throw error;
  }
}

本番デプロイのベストプラクティス

# Dockerfile 例
FROM node:20-alpine
 
WORKDIR /app
 
COPY package*.json ./
RUN npm ci --only=production
 
COPY src ./src
COPY dist ./dist
 
# MCP サーバーをバックグラウンド起動
CMD ["node", "dist/server.js"]

#!/bin/bash
# デプロイスクリプト
 
# ビルド
npm run build
 
# テスト実行
npm test
 
# Docker イメージ作成
docker build -t mcp-server:latest .
 
# 本番環境にプッシュ
docker push gcr.io/my-project/mcp-server:latest

Claude Code の高度なカスタマイズ

Claude Code の機能を最大限に活用するための高度な設定です。

HTTP Hooks による外部監視・制御

Claude Code の実行状況を外部システムに通知できます。

{
  "hooks": {
    "http": {
      "onTaskStart": "https://monitoring.example.com/webhook/task-start",
      "onTaskComplete": "https://monitoring.example.com/webhook/task-complete",
      "onError": "https://monitoring.example.com/webhook/error"
    }
  }
}

# webhook.py - 外部監視システム側
from flask import Flask, request
 
app = Flask(__name__)
 
@app.route("/webhook/task-start", methods=["POST"])
def task_start():
    data = request.json
    print(f"▶ タスク開始: {data['taskId']}")
    return {"status": "received"}
 
@app.route("/webhook/task-complete", methods=["POST"])
def task_complete():
    data = request.json
    duration = data["duration"]
    print(f"✓ タスク完了: {data['taskId']} ({duration}ms)")
    return {"status": "received"}
 
@app.route("/webhook/error", methods=["POST"])
def error_handler():
    data = request.json
    print(f"✗ エラー: {data['message']}")
    # アラート送信
    send_alert(f"Claude Code エラー: {data['message']}")
    return {"status": "received"}
 
if __name__ == "__main__":
    app.run(port=5000)

カスタムフック（PreToolUse / PostToolUse）

ツール実行前後に処理を挿入できます。

# CLAUDE.md のカスタムフック設定
"""
## Custom Hooks
 
### PreToolUse Hook
ツール実行前に実行される処理。API キーの検証、リソースの確認など。
 
### PostToolUse Hook
ツール実行後に実行される処理。結果のログ、キャッシュ更新など。
"""
 
# custom_hooks.py
import json
from datetime import datetime
 
class CustomHooks:
    @staticmethod
    def pre_tool_use(tool_name: str, arguments: dict) -> dict:
        """ツール実行前の処理"""
 
        # ロギング
        print(f"[{datetime.now().isoformat()}] ツール実行: {tool_name}")
        print(f"  引数: {json.dumps(arguments, indent=2)}")
 
        # API キーの検証
        if tool_name == "call_external_api":
            if "api_key" not in arguments:
                raise ValueError("api_key が必須です")
 
        # リソース確認
        if tool_name == "write_file":
            disk_usage = get_disk_usage()
            if disk_usage > 90:
                raise RuntimeError("ディスク容量が足りません")
 
        return arguments
 
    @staticmethod
    def post_tool_use(tool_name: str, result: str, execution_time: float) -> None:
        """ツール実行後の処理"""
 
        # 実行時間のログ
        print(f"  実行時間: {execution_time:.2f}s")
 
        # キャッシュに保存（同じツールの再利用を高速化）
        cache_key = f"{tool_name}:{hash(str(result))}"
        save_to_cache(cache_key, result)
 
        # メトリクスの記録
        record_metric(f"tool.{tool_name}.execution_time", execution_time)
 
def get_disk_usage() -> float:
    """ディスク使用率を取得"""
    import shutil
    total, used, free = shutil.disk_usage("/")
    return (used / total) * 100
 
def save_to_cache(key: str, value: str) -> None:
    """キャッシュに保存"""
    pass
 
def record_metric(name: str, value: float) -> None:
    """メトリクスを記録"""
    pass

Git ワークフロー自動化

Claude Code でコードの自動コミット、ブランチ管理が可能です。

# git_automation.py
import subprocess
from datetime import datetime
 
class GitAutomation:
    @staticmethod
    def auto_commit(message: str = None) -> None:
        """変更を自動コミット"""
 
        # 変更を確認
        status = subprocess.run(["git", "status", "--porcelain"], capture_output=True, text=True)
 
        if not status.stdout.strip():
            print("✓ 変更がありません")
            return
 
        # ステージング
        subprocess.run(["git", "add", "-A"])
 
        # コミットメッセージを自動生成
        if not message:
            changed_files = status.stdout.strip().split("\n")[:3]
            message = f"Auto-commit: Update {len(changed_files)} files"
 
        subprocess.run(["git", "commit", "-m", message])
        print(f"✓ コミット: {message}")
 
    @staticmethod
    def create_feature_branch(feature_name: str) -> None:
        """フィーチャーブランチを作成"""
 
        branch_name = f"feature/{feature_name}-{datetime.now().strftime('%Y%m%d')}"
        subprocess.run(["git", "checkout", "-b", branch_name])
        print(f"✓ ブランチ作成: {branch_name}")
 
    @staticmethod
    def auto_push() -> None:
        """現在のブランチを自動プッシュ"""
 
        subprocess.run(["git", "push", "-u", "origin", "HEAD"])
        print("✓ プッシュ完了")
 
# 使用例
git = GitAutomation()
git.create_feature_branch("user-authentication")
# ... コード作成 ...
git.auto_commit("Implement user authentication")
git.auto_push()

SaaS 収益化パイプラインの構築

Claude API を使ったビジネスモデルの実装例を3つ紹介します。

Claude API × Stripe で月額課金サービスを作る

import stripe
from anthropic import Anthropic
 
stripe.api_key = "sk_live_..."
client = Anthropic()
 
# Stripe セットアップ
PRODUCT_ID = "prod_..."
PRICE_ID = "price_..."
 
def create_subscription(customer_email: str) -> str:
    """顧客の月額課金を開始"""
 
    # Stripe 顧客を作成
    customer = stripe.Customer.create(email=customer_email)
 
    # サブスクリプションを作成
    subscription = stripe.Subscription.create(
        customer=customer.id,
        items=[{"price": PRICE_ID}],
    )
 
    return subscription.id
 
def serve_premium_feature(user_id: str, query: str) -> str:
    """プレミアムユーザーに高度な機能を提供"""
 
    # ユーザーのサブスクリプション状態を確認
    subscription = get_user_subscription(user_id)
 
    if subscription and subscription.status == "active":
        # プレミアムモデルを使用
        model = "claude-3-opus-20250219"
        max_tokens = 4096
        premium_system = "あなたはプレミアムの高度な分析を提供します。"
    else:
        # 無料モデルを使用
        model = "claude-3-haiku-20250307"
        max_tokens = 512
        premium_system = "あなたは基本的なサポートを提供します。"
 
    response = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=premium_system,
        messages=[{"role": "user", "content": query}],
    )
 
    return response.content[0].text
 
def get_user_subscription(user_id: str):
    """ユーザーのサブスクリプション情報を取得"""
    # データベースクエリ
    pass

Kindle 出版の AI 自動化ワークフロー

import os
from anthropic import Anthropic
 
client = Anthropic()
 
class KindlePublishingPipeline:
    def __init__(self, topic: str):
        self.topic = topic
        self.chapters = []
        self.cover_image = None
 
    def generate_outline(self) -> list:
        """書籍の構成を生成"""
 
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2048,
            messages=[
                {
                    "role": "user",
                    "content": f"""
{self.topic} についての技術書籍の目次を生成してください。
以下の形式で提供してください：
1. 序章
2. 第1章: ...
3. 第2章: ...
...
 
各章には詳細な説明も含めてください。
"""
                }
            ]
        )
 
        outline_text = response.content[0].text
        self.outline = outline_text
        return outline_text
 
    def write_chapters(self) -> list:
        """各章を執筆"""
 
        chapters = []
 
        for chapter_num in range(1, 4):  # 3章を生成
            print(f"✓ 第{chapter_num}章を執筆中...")
 
            response = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=3000,
                messages=[
                    {
                        "role": "user",
                        "content": f"""
{self.topic} の第{chapter_num}章を執筆してください。
 
目次:
{self.outline}
 
要件:
- 2000～3000 語
- 初心者向けの説明
- 実例とコード例を含む
- Kindle 向けにマークダウン形式で
"""
                    }
                ]
            )
 
            chapter_content = response.content[0].text
            chapters.append({
                "number": chapter_num,
                "content": chapter_content
            })
 
        self.chapters = chapters
        return chapters
 
    def generate_cover(self) -> str:
        """表紙画像を生成（外部 API 連携例）"""
 
        # 表紙デザインの説明を生成
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            messages=[
                {
                    "role": "user",
                    "content": f"""
{self.topic} についての Kindle 本の表紙デザインを説明してください。
レイアウト、色、フォント、画像スタイルを詳しく記述してください。
"""
                }
            ]
        )
 
        cover_description = response.content[0].text
 
        # 外部API（Midjourney など）で画像生成
        # image_url = call_image_generation_api(cover_description)
        image_url = "https://example.com/cover.jpg"
 
        return image_url
 
    def publish_to_kindle(self) -> bool:
        """Kindle Direct Publishing にアップロード"""
 
        # KDP API で出版
        print(f"✓ '{self.topic}' を Kindle Direct Publishing にアップロード中...")
 
        # 実装例
        # kdp_client.publish(
        #     title=f"{self.topic}: 完全ガイド",
        #     content="\n\n".join([ch["content"] for ch in self.chapters]),
        #     cover_image=self.cover_image
        # )
 
        return True
 
# 使用例
pipeline = KindlePublishingPipeline("Python による Web スクレイピング")
pipeline.generate_outline()
pipeline.write_chapters()
pipeline.generate_cover()
pipeline.publish_to_kindle()
 
print("✓ Kindle 出版が完了しました！")

YouTube 動画制作パイプライン（Pollo AI + Suno AI）

from anthropic import Anthropic
 
client = Anthropic()
 
class YouTubeContentPipeline:
    def __init__(self, topic: str, duration_minutes: int = 10):
        self.topic = topic
        self.duration = duration_minutes
        self.script = None
        self.audio_url = None
        self.video_url = None
 
    def generate_script(self) -> str:
        """YouTube 動画のスクリプトを生成"""
 
        word_count = self.duration * 130  # 1分 = 約130語
 
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=3000,
            messages=[
                {
                    "role": "user",
                    "content": f"""
YouTube 動画 "{self.topic}" の {self.duration} 分のスクリプトを生成してください。
 
要件:
- 総語数: 約 {word_count} 語
- 話し口調で、自然な流れ
- [00:00] などのタイムコード付き
- ナレーションと B-roll の指示を含める
 
形式:
[00:00 - イントロ]
内容...
 
[00:30 - メインパート]
内容...
"""
                }
            ]
        )
 
        self.script = response.content[0].text
        return self.script
 
    def generate_audio(self) -> str:
        """Suno AI を使ってナレーション音声を生成"""
 
        print("✓ Suno AI でナレーション音声を生成中...")
 
        # Suno AI API（仮）
        # audio_url = suno_client.generate_speech(
        #     text=self.script,
        #     voice="professional-narrator-ja",
        #     duration_seconds=self.duration * 60
        # )
 
        self.audio_url = "https://example.com/narration.mp3"
        return self.audio_url
 
    def generate_visuals(self) -> str:
        """Pollo AI を使ってビジュアルを生成"""
 
        print("✓ Pollo AI でビジュアルを生成中...")
 
        # スクリプトから場面を抽出
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": f"""
以下のスクリプトから、各場面に適したビジュアルプロンプトを生成してください。
JSON 形式で、[時間] → [ビジュアルプロンプト] の形式で。
 
スクリプト:
{self.script[:1000]}
"""
                }
            ]
        )
 
        visual_prompts = response.content[0].text
 
        # Pollo AI API（仮）
        # video_url = pollo_client.generate_video(
        #     script=self.script,
        #     audio_url=self.audio_url,
        #     visual_prompts=visual_prompts
        # )
 
        self.video_url = "https://example.com/video.mp4"
        return self.video_url
 
    def publish_to_youtube(self, title: str, description: str) -> str:
        """YouTube にアップロード"""
 
        print("✓ YouTube にアップロード中...")
 
        # YouTube API
        # video_id = youtube_client.upload(
        #     video_url=self.video_url,
        #     title=title,
        #     description=description,
        #     tags=["AI", "Tutorial", self.topic]
        # )
 
        return "https://youtube.com/watch?v=dQw4w9WgXcQ"
 
# 使用例
pipeline = YouTubeContentPipeline("Claude API の実践テクニック", duration_minutes=15)
pipeline.generate_script()
pipeline.generate_audio()
pipeline.generate_visuals()
video_url = pipeline.publish_to_youtube(
    title="Claude API で SaaS を構築する方法",
    description="Claude API を使ってビジネス化可能な SaaS アプリケーションを作る実践ガイド"
)
 
print(f"✓ 動画が公開されました: {video_url}")

まとめ — 関連プレミアム記事一覧

このドキュメントで学んだ高度なテクニックを活かし、実際のプロジェクトに適用することで、Claude の力を最大限に引き出すことができます。

さらに学べるコンテンツ

Claude AI 完全ガイド 2026 — Claude の基礎から実践まで
Claude 実践テクニック集【前編】 — API とコーディングの基本パターン

Claude Lab コミュニティ

Claude Lab では、他のユーザーとの情報交換、事例共有、ベストプラクティスの討論が活発に行われています。あなたの実装経験をコミュニティで共有することで、互いに学び合える環境を作ることが目標です。

本番環境での実装、新しい API の活用、ビジネス化の工夫——これらの知識を持つことで、あなたは Claude を使ったイノベーションの最前線に立つことができます。

さらなる高みへ——Claude の無限の可能性を探索してください。