Claudeでカスタムコマンドを構築し、ワークフロー自動化を実現する方法

Claude APIを活用して独自のカスタムコマンドを作成することで、繰り返し作業を自動化し、チーム全体の生産性を大幅に向上させられます。本記事では、実装から運用までの実践的なステップを解説します。

Claude APIでカスタムコマンドが必要な理由

Claudeは強力な言語モデルですが、APIをそのまま使用するだけではプロジェクト固有のニーズに対応しきれません。カスタムコマンドを構築することで、以下のメリットが得られます:

  • チーム内で統一された処理フローの実現
  • 複雑なプロンプトエンジニアリングの隠蔽化
  • API呼び出しコストの最適化
  • エラーハンドリングと入力検証の一元管理

カスタムコマンドの基本的な実装パターン

ステップ1: API認証情報の準備

Anthropic ConsoleからAPIキーを取得し、環境変数として設定します。これが全てのカスタムコマンドの基盤になります。

# .env ファイルに以下を記述
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxxxxxxx
CLAUDE_MODEL=claude-3-5-sonnet-20241022

ステップ2: シンプルなカスタムコマンド構造の実装

Pythonを使用した基本的なカスタムコマンドの実装例です。このパターンは、ドキュメント生成やコード審査など、様々なタスクに応用できます。

import anthropic
import os
from typing import Optional

class ClaudeCustomCommand:
    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = os.environ.get("CLAUDE_MODEL", "claude-3-5-sonnet-20241022")
    
    def execute_command(self, command_name: str, user_input: str, system_prompt: Optional[str] = None) -> str:
        """カスタムコマンドを実行し、Claudeからの応答を返す"""
        
        # 入力値の検証
        if not user_input or len(user_input.strip()) == 0:
            raise ValueError("ユーザー入力が空です")
        
        # デフォルトのシステムプロンプト
        if system_prompt is None:
            system_prompt = self._get_system_prompt(command_name)
        
        try:
            message = self.client.messages.create(
                model=self.model,
                max_tokens=2048,
                system=system_prompt,
                messages=[
                    {"role": "user", "content": user_input}
                ]
            )
            
            return message.content[0].text
        
        except anthropic.APIError as e:
            raise RuntimeError(f"API呼び出しエラー: {str(e)}")
    
    def _get_system_prompt(self, command_name: str) -> str:
        """コマンド名に応じて適切なシステムプロンプトを返す"""
        prompts = {
            "code_review": """あなたはシニアソフトウェアエンジニアです。提供されたコードを厳密にレビューし、
品質、セキュリティ、パフォーマンスの観点から改善提案を提供してください。
具体的な改善箇所と修正案をコード例付きで示してください。""",
            
            "doc_generator": """あなたは技術ドキュメント作成の専門家です。提供された情報から
明確で構造化されたドキュメントを生成してください。
マークダウン形式で、見出し、コード例、使用例を含めてください。""",
            
            "bug_analyzer": """あなたはバグ分析の専門家です。エラーログやバグレポートから
根本原因を特定し、解決策を提案してください。
デバッグのステップも含めて説明してください。"""
        }
        
        return prompts.get(command_name, "ユーザーのリクエストに対して正確で有用な回答を提供してください。")

# 使用例
if __name__ == "__main__":
    command = ClaudeCustomCommand()
    
    # コードレビューコマンドの実行
    code_sample = """
def calculate_average(numbers):
    total = 0
    for n in numbers:
        total = total + n
    return total / len(numbers)
"""
    
    result = command.execute_command("code_review", code_sample)
    print("=== コードレビュー結果 ===")
    print(result)

ステップ3: より高度なコマンド設計(複数ステップの処理)

複雑なタスクには、複数のステップを組み合わせたカスタムコマンドが有効です。以下は、仕様書からテストケースを自動生成する例です。

class AdvancedClaudeCommand:
    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-3-5-sonnet-20241022"
        self.conversation_history = []
    
    def generate_test_cases_from_spec(self, specification: str, iterations: int = 2) -> str:
        """仕様書からテストケースを段階的に生成"""
        
        # ステップ1: 仕様書の分析
        analysis_prompt = f"""以下の仕様書を分析し、テストが必要な主要なシナリオを5つ列挙してください:

{specification}

各シナリオについて、テストの目的と期待される結果を簡潔に説明してください。"""
        
        analysis = self._call_claude(analysis_prompt)
        self.conversation_history.append({"role": "user", "content": analysis_prompt})
        self.conversation_history.append({"role": "assistant", "content": analysis})
        
        # ステップ2: テストケースの詳細化
        detail_prompt = """上記のシナリオに基づいて、実行可能なテストケースを生成してください。
各テストケースは以下の形式で提供してください:
- テストID
- 入力データ
- 実行ステップ
- 期待される結果
- エッジケースの考慮"""
        
        detailed_tests = self._call_claude(detail_prompt)
        self.conversation_history.append({"role": "user", "content": detail_prompt})
        self.conversation_history.append({"role": "assistant", "content": detailed_tests})
        
        return detailed_tests
    
    def _call_claude(self, user_message: str) -> str:
        """会話履歴を保持しながらClaudeを呼び出す"""
        self.conversation_history.append({"role": "user", "content": user_message})
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            messages=self.conversation_history
        )
        
        assistant_message = response.content[0].text
        self.conversation_history.append({"role": "assistant", "content": assistant_message})
        
        return assistant_message

カスタムコマンド構築でのハマりやすいポイントと解決策

問題1: トークン数超過によるエラー

長いテキストを処理する際、max_tokensの設定が不適切だとエラーが発生します。入力テキストの長さに応じて動的に調整する実装を推奨します。

def calculate_max_tokens(input_length: int) -> int:
    """入力テキスト長に基づいて max_tokens を計算"""
    # 入力テキストの約1/4がトークン数の目安
    estimated_input_tokens = input_length // 4
    # 出力用に十分なトークンを確保(最小1024、最大4096)
    output_tokens = max(1024, min(4096, 8192 - estimated_input_tokens))
    return output_tokens

# 使用時
max_tokens = calculate_max_tokens(len(user_input))
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=max_tokens,
    messages=[{"role": "user", "content": user_input}]
)

問題2: APIレート制限への対応

複数のリクエストを送信する際、レート制限に引っかかることがあります。リトライ機構を実装することが重要です。

import time
from anthropic import RateLimitError

def call_claude_with_retry(client, messages, max_retries: int = 3, initial_delay: float = 1.0):
    """レート制限対応のリトライ機構付きAPI呼び出し"""
    
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=2048,
                messages=messages
            )
            return response.content[0].text
        
        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = initial_delay * (2 ** attempt)  # 指数バックオフ
                print(f"レート制限に達しました。{wait_time}秒待機します...")
                time.sleep(wait_time)
            else:
                raise
    
    raise RuntimeError("最大リトライ回数に達しました")

問題3: プロンプトインジェクション攻撃への対策

ユーザー入力を含む場合、セキュリティ対策が必要です。入力サニタイズと明確なシステムプロンプトの設定を実施してください。

import re

def sanitize_user_input(user_input: str, max_length: int = 5000) -> str:
    """ユーザー入力のサニタイズ"""
    
    # 長さチェック
    if len(user_input) > max_length:
        raise ValueError(f"入力が長すぎます(最大{max_length}文字)")
    
    # 危険な文字パターンの検出
    dangerous_patterns = [
        r"ignore.*previous.*instruction",
        r"system.*prompt",
        r"system.*message"
    ]
    
    normalized_input = user_input.lower()
    for pattern in dangerous_patterns:
        if re.search(pattern, normalized_input, re.IGNORECASE):
            raise ValueError("入力に不適切なパターンが含まれています")
    
    return user_input.strip()

カスタムコマンドの運用と最適化

コスト削減のためのキャッシング実装

同じリクエストが繰り返される場合、結果をキャッシュすることでAPI呼び出し回数を削減できます。

import hashlib
import json
from functools import wraps

class CachedClaudeCommand:
    def __init__(self, cache_file: str = "claude_cache.json"):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.cache_file = cache_file
        self.cache = self._load_cache()
    
    def _load_cache(self) -> dict:
        """ディスクからキャッシュを読み込む"""
        if os.path.exists(self.cache_file):
            with open(self.cache_file, 'r') as f:
                return json.load(f)
        return {}
    
    def _save_cache(self):
        """キャッシュをディスクに保存"""
        with open(self.cache_file, 'w') as f:
            json.dump(self.cache, f, ensure_ascii=False, indent=2)
    
    def _get_cache_key(self, prompt: str, system_prompt: str) -> str:
        """プロンプトからキャッシュキーを生成"""
        combined = f"{system_prompt}||{prompt}"
        return hashlib.sha256(combined.encode()).hexdigest()
    
    def execute_with_cache(self, prompt: str, system_prompt: str) -> str:
        """キャッシュを活用した実行"""
        
        cache_key = self._get_cache_key(prompt, system_prompt)
        
        # キャッシュヒット
        if cache_key in self.cache:
            print("✓ キャッシュから結果を取得しました")
            return self.cache[cache_key]
        
        # 新規リクエスト
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2048,
            system=system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )
        
        result = response.content[0].text
        self.cache[cache_key] = result
        self._save_cache()
        
        return result

使うべき場面と使うべきでない場面

カスタムコマンドが有効な場面:

  • チーム内で繰り返し実行されるタスク(コード審査、ドキュメント生成など)
  • 複雑なプロンプトを統一的に管理したい場合
  • エラーハンドリングとログが必要な業務プロセス
K
AWS・Python・生成AIを専門とするソフトウェアエンジニア。AI・クラウド・開発ワークフローの実践ガイドを執筆しています。詳しく見る →