Claude Code Hooksでワークフロー自動化を実装する実践ガイド

本記事では、Claude Code Hooksを活用して、開発ワークフロー内での繰り返しタスクを自動化する具体的な実装方法を解説します。API連携からエラーハンドリングまで、すぐに本番環境で使える実装パターンを紹介します。

Claude Code Hooksとは何か

Claude Code Hooksは、Claudeが外部システムと連携して特定のイベントをトリガーに自動実行するメカニズムです。開発プロセスにおいて、コードレビュー、テスト実行、ビルドパイプラインなどを自動で統合できます。

GithubのWebhookやJenkinsのトリガーと異なり、Claude APIを直接活用することで、より複雑な判定ロジックや複数ステップの処理を自然言語で定義できるのが特徴です。

Claude Code Hooksの基本実装

簡単なHook設定の作成

まずは基本的なコード実行フックを実装してみましょう。以下の例は、コミットがプッシュされたときにコード品質チェックを自動実行するパターンです。

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic()

def execute_code_hook(commit_message: str, changed_files: list[str]) -> dict:
    """
    コミット情報を受け取り、自動チェックを実行するフック
    """
    
    hook_definition = f"""
    あなたはコード品質管理ツールとして動作します。
    
    コミットメッセージ: {commit_message}
    変更ファイル: {', '.join(changed_files)}
    
    以下の項目をチェックしてください:
    1. コミットメッセージが適切な形式か（タイプ(スコープ): 説明）
    2. 変更ファイルが関連性があるか
    3. 大規模なファイル変更（1ファイル500行以上）がないか
    4. 実行すべきテストがあれば、そのコマンドを提示してください
    
    JSON形式で結果を返してください:
    {{
        "is_valid": true/false,
        "issues": ["issue1", "issue2"],
        "required_tests": ["test_command_1"],
        "severity": "info/warning/error"
    }}
    """
    
    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": hook_definition}
        ]
    )
    
    # レスポンスからJSON部分を抽出
    response_text = message.content[0].text
    json_start = response_text.find('{')
    json_end = response_text.rfind('}') + 1
    
    try:
        result = json.loads(response_text[json_start:json_end])
    except json.JSONDecodeError:
        result = {
            "is_valid": False,
            "issues": ["レスポンスパースエラー"],
            "severity": "error"
        }
    
    result["timestamp"] = datetime.now().isoformat()
    return result


# 使用例
if __name__ == "__main__":
    commit_msg = "feat(auth): JWT トークン検証ロジックの実装"
    files_changed = ["src/auth/jwt_validator.py", "tests/auth/test_jwt.py"]
    
    result = execute_code_hook(commit_msg, files_changed)
    print(json.dumps(result, indent=2, ensure_ascii=False))

実行環境の確認

テスト環境: Python 3.11 / Claude API 2025-01 / anthropic==0.25.0

ワークフロー自動化の実装パターン

複数ステップの自動処理チェーン

実際のワークフロー自動化では、複数の処理を連鎖させることが一般的です。コード品質チェック → テスト実行 → デプロイ判定というように段階的に実行する仕組みを作ります。

class WorkflowAutomation:
    def __init__(self, api_key: str = None):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.workflow_state = {}
    
    def run_quality_check(self, code_content: str) -> dict:
        """ステップ1: コード品質チェック"""
        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""
                以下のPythonコードをレビューしてください。
                
                
                {code_content}
                ```
                
                JSON形式で以下を返してください:
                {{
                    "quality_score": 0-100,
                    "issues": ["issue1", "issue2"],
                    "can_proceed": true/false,
                    "required_fixes": ["fix1", "fix2"]
                }}
                """
            }]
        )
        
        response_text = message.content[0].text
        json_start = response_text.find('{')
        json_end = response_text.rfind('}') + 1
        return json.loads(response_text[json_start:json_end])
    
    def generate_test_cases(self, function_signature: str) -> dict:
        """ステップ2: テストケース自動生成"""
        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""
                関数シグネチャ: {function_signature}
                
                Pytestのテストケースを5個生成してください。
                以下の形式で返してください:
                {{
                    "test_cases": [
                        {{
                            "name": "test_case_name",
                            "input": {{}},
                            "expected_output": null,
                            "description": "テストの説明"
                        }}
                    ]
                }}
                """
            }]
        )
        
        response_text = message.content[0].text
        json_start = response_text.find('{')
        json_end = response_text.rfind('}') + 1
        return json.loads(response_text[json_start:json_end])
    
    def evaluate_deployment_readiness(self, checks_results: dict) -> dict:
        """ステップ3: デプロイ可否判定"""
        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""
                以下のチェック結果に基づいて、デプロイ可否を判定してください。
                
                チェック結果: {json.dumps(checks_results, indent=2, ensure_ascii=False)}
                
                JSON形式で返してください:
                {{
                    "can_deploy": true/false,
                    "risk_level": "low/medium/high",
                    "blockers": ["blocker1"],
                    "recommendations": ["recommendation1"],
                    "estimated_rollback_time": "HH:MM"
                }}
                """
            }]
        )
        
        response_text = message.content[0].text
        json_start = response_text.find('{')
        json_end = response_text.rfind('}') + 1
        return json.loads(response_text[json_start:json_end])
    
    def execute_workflow(self, code_content: str, function_sig: str) -> dict:
        """ワークフロー全体を実行"""
        results = {
            "timestamp": datetime.now().isoformat(),
            "stages": {}
        }
        
        # ステップ1
        print("🔍 コード品質チェック実行中...")
        quality_result = self.run_quality_check(code_content)
        results["stages"]["quality_check"] = quality_result
        
        if not quality_result.get("can_proceed", False):
            results["workflow_status"] = "FAILED"
            return results
        
        # ステップ2
        print("✅ テストケース生成中...")
        test_result = self.generate_test_cases(function_sig)
        results["stages"]["test_generation"] = test_result
        
        # ステップ3
        print("🚀 デプロイ判定実行中...")
        deployment_result = self.evaluate_deployment_readiness({
            "quality": quality_result,
            "tests": test_result
        })
        results["stages"]["deployment_evaluation"] = deployment_result
        results["workflow_status"] = "SUCCESS"
        
        return results


# 使用例
if __name__ == "__main__":
    automation = WorkflowAutomation()
    
    sample_code = """
def calculate_average(numbers: list[float]) -> float:
    if not numbers:
        return 0
    return sum(numbers) / len(numbers)
"""
    
    workflow_result = automation.execute_workflow(
        code_content=sample_code,
        function_sig="def calculate_average(numbers: list[float]) -> float:"
    )
    
    print("\n📊 ワークフロー実行結果:")
    print(json.dumps(workflow_result, indent=2, ensure_ascii=False))


  
よくあるハマりポイントと解決策
  
  
JSONパースエラーの回避
  
  
Claude APIからのレスポンスには、JSONの前後に説明文が含まれることがあります。以下のようにロバストなパース関数を用意しましょう。

  
import json
import re

def safe_parse_json(response_text: str) -> dict:
    """
    レスポンステキストから安全にJSONを抽出してパース
    """
    # 複数のJSON抽出パターンを試す
    patterns = [
        r'\{[\s\S]*\}',  # 最初の { から最後の } まで
        r'\[[\s\S]*\]',  # JSONArray
    ]
    
    for pattern in patterns:
        matches = re.findall(pattern, response_text)
        if matches:
            for match in matches:
                try:
                    return json.loads(match)
                except json.JSONDecodeError:
                    continue
    
    # 抽出失敗時のフォールバック
    raise ValueError(f"JSONパースに失敗しました: {response_text[:100]}...")


  
トークン数超過エラーの対策
  
  
大規模なコードベースを処理する場合、トークン数が制限を超える可能性があります。以下のように入力を分割して処理します。

  
def split_large_content(content: str, max_chunk_size: int = 8000) -> list[str]:
    """
    大きなコンテンツを分割
    max_chunk_size: 1チャンク当たりの最大文字数
    """
    chunks = []
    current_chunk = ""
    
    for line in content.split('\n'):
        if len(current_chunk) + len(line) > max_chunk_size:
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = line
        else:
            current_chunk += line + '\n'
    
    if current_chunk:
        chunks.append(current_chunk)
    
    return chunks


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

    
複雑な判定ロジックが必要なコード品質チェック
    
テストケースの自動生成
    
リリースノート、ドキュメントの自動生成
    
複数ステップの条件判定が伴うワークフロー
  
  
  
❌ 使うべきでない場面
  

    
シンプルな正規表現マッチングや単純なフィルタリング
    
ミリ秒単位の高速レスポンスが必須な処理
    
コストが最重要な運用環境での大量実行
  

  
代替手段との比較
  
  
Github Actionsはファイルベースの定義が得意で、複雑なCI/CDパイプラインに向いています。一方、Claude Code Hooksは複雑な判定ロジックが必要な場合や、自然言語による動的な判定が有効です。Jenkins

  
実践的な応用例：PR自動レビューシステム
  
  
Pull Requestが作成されたときに自動的にコードレビューを実施するシステムを構築します。

  
class PRAutoReviewer:
    def __init__(self):
        self.client = anthropic.Anthropic()
    
    def review_pr(self, pr_title: str, pr_description: str, 
                  changed_files: dict[str, str]) -> str:
        """
        PR内容を自動レビュー
        changed_files: {"filename": "code_content"}
        """
        
        file_summary = "\n".join([
            f"### {filename}\n```\n{content[:500]}...\n```"
            for filename, content in changed_files.items()
        ])
        
        review_prompt = f"""
        Pull Request自動レビュアーとして動作してください。
        
        PR タイトル: {pr_title}
        PR 説明: {pr_description}
        
        変更内容: