Claude Code × MCP で構築する個人開発 AI 自動化ハブ — 複数サービス横断の設計と実装

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN_READONLY}"
      }
    },
    "notion": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-notion"],
      "env": {
        "NOTION_API_KEY": "${NOTION_API_KEY}",
        "NOTION_ROOT_PAGE_ID": "${NOTION_ROOT_PAGE_ID}"
      }
    },
    "slack": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-slack"],
      "env": {
        "SLACK_BOT_TOKEN": "${SLACK_BOT_TOKEN}",
        "SLACK_TEAM_ID": "${SLACK_TEAM_ID}"
      }
    }
  }
}

# pr_summary.py — Claude Code から呼び出すスクリプト
import subprocess
import json
import sys
 
def get_pr_context(repo: str, pr_number: int) -> dict:
    """
    GitHub MCP 経由で PR の全コンテキストを収集する。
    diff・コメント・関連 Issue を一度に取得して Claude に渡す。
    """
    try:
        # PR の基本情報取得
        result = subprocess.run(
            ["gh", "pr", "view", str(pr_number), "--repo", repo, "--json",
             "title,body,files,additions,deletions,reviewRequests,labels"],
            capture_output=True, text=True, check=True
        )
        pr_data = json.loads(result.stdout)
        
        # diff の取得（大きすぎる場合は先頭 500 行に制限）
        diff_result = subprocess.run(
            ["gh", "pr", "diff", str(pr_number), "--repo", repo],
            capture_output=True, text=True, check=True
        )
        diff_lines = diff_result.stdout.splitlines()
        if len(diff_lines) > 500:
            diff_content = "\n".join(diff_lines[:500]) + "\n[... 残り省略 ...]"
        else:
            diff_content = diff_result.stdout
            
        return {
            "pr": pr_data,
            "diff": diff_content,
            "error": None
        }
    except subprocess.CalledProcessError as e:
        # GitHub CLI が返したエラーを上位に伝播させる
        return {
            "pr": None,
            "diff": None,
            "error": f"GitHub CLI エラー: {e.stderr.strip()}"
        }
    except json.JSONDecodeError as e:
        return {
            "pr": None,
            "diff": None,
            "error": f"JSON パースエラー: {str(e)}"
        }
 
if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("使い方: python pr_summary.py <owner/repo> <pr_number>")
        sys.exit(1)
    
    repo = sys.argv[1]
    pr_number = int(sys.argv[2])
    context = get_pr_context(repo, pr_number)
    
    if context["error"]:
        print(f"エラー: {context['error']}", file=sys.stderr)
        sys.exit(1)
    
    # Claude Code に渡す JSON を標準出力へ
    print(json.dumps(context, ensure_ascii=False, indent=2))

# issue_triage.py — Issue の全コンテキストを収集して優先度提案を依頼
import subprocess
import json
from datetime import datetime, timezone
 
def fetch_open_issues(repo: str, limit: int = 50) -> list[dict]:
    """
    オープン Issue を取得する。
    'updated' ソートで最近更新されたものを優先する。
    """
    try:
        result = subprocess.run(
            ["gh", "issue", "list", "--repo", repo,
             "--state", "open", "--limit", str(limit),
             "--json", "number,title,body,labels,assignees,milestone,createdAt,updatedAt,comments"],
            capture_output=True, text=True, check=True
        )
        issues = json.loads(result.stdout)
        
        # 各 Issue のコメント数とラベルを整形
        for issue in issues:
            issue["age_days"] = (
                datetime.now(timezone.utc) - 
                datetime.fromisoformat(issue["createdAt"].replace("Z", "+00:00"))
            ).days
            issue["label_names"] = [l["name"] for l in issue.get("labels", [])]
            issue["assignee_names"] = [a["login"] for a in issue.get("assignees", [])]
        
        return issues
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Issue 取得に失敗しました: {e.stderr.strip()}")
 
if __name__ == "__main__":
    import sys
    repo = sys.argv[1] if len(sys.argv) > 1 else "owner/repo"
    
    try:
        issues = fetch_open_issues(repo)
        print(json.dumps(issues, ensure_ascii=False, indent=2))
    except RuntimeError as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

# notion_dev_log.py — 開発ログを Notion に自動記録
import os
import requests
import json
from datetime import datetime
 
NOTION_API_KEY = os.environ["NOTION_API_KEY"]
DEV_LOG_DB_ID = os.environ["NOTION_DEV_LOG_DB_ID"]
NOTION_VERSION = "2022-06-28"
 
HEADERS = {
    "Authorization": f"Bearer {NOTION_API_KEY}",
    "Content-Type": "application/json",
    "Notion-Version": NOTION_VERSION
}
 
def create_dev_log(
    title: str,
    summary: str,
    commits: list[str],
    next_tasks: list[str],
    duration_minutes: int
) -> dict:
    """
    開発セッションのログを Notion データベースに作成する。
    エラー時は詳細なメッセージとともに例外を送出する。
    """
    today = datetime.now().strftime("%Y-%m-%d")
    
    # Notion のブロック形式でコンテンツを構築
    children_blocks = [
        {
            "object": "block",
            "type": "heading_2",
            "heading_2": {
                "rich_text": [{"type": "text", "text": {"content": "セッションサマリー"}}]
            }
        },
        {
            "object": "block",
            "type": "paragraph",
            "paragraph": {
                "rich_text": [{"type": "text", "text": {"content": summary}}]
            }
        },
        {
            "object": "block",
            "type": "heading_2",
            "heading_2": {
                "rich_text": [{"type": "text", "text": {"content": "コミット一覧"}}]
            }
        }
    ]
    
    # コミットをリストブロックに変換
    for commit in commits:
        children_blocks.append({
            "object": "block",
            "type": "bulleted_list_item",
            "bulleted_list_item": {
                "rich_text": [{"type": "text", "text": {"content": commit}}]
            }
        })
    
    # 次のタスクセクションを追加
    children_blocks.append({
        "object": "block",
        "type": "heading_2",
        "heading_2": {
            "rich_text": [{"type": "text", "text": {"content": "次回タスク"}}]
        }
    })
    for task in next_tasks:
        children_blocks.append({
            "object": "block",
            "type": "to_do",
            "to_do": {
                "rich_text": [{"type": "text", "text": {"content": task}}],
                "checked": False
            }
        })
    
    payload = {
        "parent": {"database_id": DEV_LOG_DB_ID},
        "properties": {
            "Name": {
                "title": [{"text": {"content": f"{today} — {title}"}}]
            },
            "Date": {
                "date": {"start": today}
            },
            "Duration": {
                "number": duration_minutes
            }
        },
        "children": children_blocks
    }
    
    response = requests.post(
        "https://api.notion.com/v1/pages",
        headers=HEADERS,
        json=payload,
        timeout=30
    )
    
    if response.status_code != 200:
        error_detail = response.json().get("message", response.text)
        raise RuntimeError(
            f"Notion API エラー (HTTP {response.status_code}): {error_detail}"
        )
    
    return response.json()
 
if __name__ == "__main__":
    # 使用例: 動作確認のためのサンプル実行
    try:
        result = create_dev_log(
            title="認証機能の実装",
            summary="JWT ベースの認証フローを実装。リフレッシュトークンの TTL 設定に時間がかかった。",
            commits=[
                "feat: JWT 認証ミドルウェアを追加",
                "test: 認証フローの統合テストを追加",
                "fix: リフレッシュトークンの有効期限バグを修正"
            ],
            next_tasks=[
                "OAuth2 対応を追加する",
                "認証エラーのエラーページを作成する"
            ],
            duration_minutes=180
        )
        page_url = result.get("url", "URL 不明")
        print(f"✅ Notion ページを作成しました: {page_url}")
    except RuntimeError as e:
        print(f"❌ エラー: {e}")
        exit(1)

# slack_notifier.py — 構造化メッセージを Slack に送信
import os
import requests
import json
from typing import Optional
 
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
DEFAULT_CHANNEL = os.environ.get("SLACK_DEFAULT_CHANNEL", "#dev-notifications")
 
def send_notification(
    message: str,
    channel: str = DEFAULT_CHANNEL,
    thread_ts: Optional[str] = None,
    blocks: Optional[list] = None
) -> dict:
    """
    Slack にメッセージを送信する。
    blocks を指定するとリッチフォーマットで表示できる。
    スレッド返信には thread_ts を渡す。
    
    Rate Limit: Slack は Tier 3（50回/分）のため、
    ループ内で呼ぶ場合は time.sleep(1.2) を挟むこと。
    """
    payload = {
        "channel": channel,
        "text": message  # blocks がない場合のフォールバックテキスト
    }
    
    if thread_ts:
        payload["thread_ts"] = thread_ts
    
    if blocks:
        payload["blocks"] = blocks
    
    response = requests.post(
        "https://slack.com/api/chat.postMessage",
        headers={
            "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=15
    )
    
    data = response.json()
    
    # Slack API は HTTP 200 でもエラーを返すことがある
    if not data.get("ok"):
        error_code = data.get("error", "unknown_error")
        
        if error_code == "rate_limited":
            retry_after = data.get("retry_after", 60)
            raise RuntimeError(
                f"レート制限に達しました。{retry_after}秒後に再試行してください。"
            )
        elif error_code == "channel_not_found":
            raise ValueError(
                f"チャンネル '{channel}' が見つかりません。チャンネル名を確認してください。"
            )
        elif error_code == "not_in_channel":
            raise PermissionError(
                f"Bot がチャンネル '{channel}' に参加していません。Bot をチャンネルに追加してください。"
            )
        else:
            raise RuntimeError(f"Slack API エラー: {error_code}")
    
    return data
 
def send_pr_review_notification(
    pr_title: str,
    pr_url: str,
    reviewer: str,
    summary: str,
    channel: str = DEFAULT_CHANNEL
) -> None:
    """
    PR レビュー完了の通知を送る。Block Kit でリッチ表示。
    """
    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": "✅ PR レビュー完了"
            }
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*PR:*\n<{pr_url}|{pr_title}>"},
                {"type": "mrkdwn", "text": f"*レビュアー:*\n{reviewer}"}
            ]
        },
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*レビューサマリー:*\n{summary}"
            }
        }
    ]
    
    send_notification(
        message=f"PR レビュー完了: {pr_title}",
        channel=channel,
        blocks=blocks
    )

<!-- CLAUDE.md の該当セクション -->
 
## 朝の作業開始ルーティン
 
「今日の作業を準備して」と言われたら以下を順番に実行すること:
 
1. GitHub: 自分のリポジトリのオープン PR と割り当て済み Issue を取得
2. Notion: 「タスク管理」データベースから今日期限のタスクを取得
3. Slack: #dev チャンネルの昨日からの未読メッセージを要約
4. 上記を統合して優先度付きのデイリーサマリーを日本語で出力
 
## PR レビュー完了フロー
 
「PR #[番号] のレビューをまとめて」と言われたら:
 
1. GitHub MCP で PR の diff とコメント履歴を取得
2. 関連する Notion のドキュメントページ（あれば）を参照
3. レビューサマリーを生成し GitHub コメントとして投稿
4. Slack の #dev-review チャンネルに完了通知を送信
 
## エラー時の動作
 
いずれかのサービスへのアクセスが失敗した場合:
- エラーをユーザーに報告し、残りのサービスで処理を継続する
- 全サービスが失敗した場合のみ処理を中断する

# healthcheck.sh — MCP サービスの疎通確認
#!/bin/bash
set -e
 
echo "=== MCP ヘルスチェック ==="
 
# GitHub: プロフィール取得で認証確認
GH_STATUS=$(gh auth status 2>&1)
if echo "$GH_STATUS" | grep -q "Logged in"; then
    echo "✅ GitHub: OK"
else
    echo "❌ GitHub: 認証エラー — gh auth login を実行してください"
    GH_FAILED=1
fi
 
# Notion: DB 一覧取得で接続確認
NOTION_STATUS=$(curl -s -o /dev/null -w "%{http_code}" \
    -H "Authorization: Bearer $NOTION_API_KEY" \
    -H "Notion-Version: 2022-06-28" \
    "https://api.notion.com/v1/users/me")
if [ "$NOTION_STATUS" = "200" ]; then
    echo "✅ Notion: OK"
else
    echo "❌ Notion: API エラー (HTTP $NOTION_STATUS) — トークンを確認してください"
    NOTION_FAILED=1
fi
 
# Slack: 認証確認
SLACK_STATUS=$(curl -s \
    -H "Authorization: Bearer $SLACK_BOT_TOKEN" \
    "https://slack.com/api/auth.test" | python3 -c "import sys,json; print(json.load(sys.stdin)['ok'])")
if [ "$SLACK_STATUS" = "True" ]; then
    echo "✅ Slack: OK"
else
    echo "❌ Slack: Bot トークンエラー — Bot の再インストールが必要かもしれません"
    SLACK_FAILED=1
fi
 
if [ -n "$GH_FAILED" ] || [ -n "$NOTION_FAILED" ] || [ -n "$SLACK_FAILED" ]; then
    echo ""
    echo "⚠️  一部のサービスに接続できません。上記のエラーを確認してください。"
    exit 1
fi
 
echo ""
echo "✅ 全サービスに接続できています"

# simple_cache.py — 軽量なファイルベースキャッシュ
import json
import time
from pathlib import Path
 
CACHE_DIR = Path.home() / ".claude_hub_cache"
CACHE_DIR.mkdir(exist_ok=True)
 
def get_cached(key: str, ttl_seconds: int = 300):
    """
    TTL 付きのシンプルなキャッシュ。
    デフォルト TTL は 5 分。
    """
    cache_file = CACHE_DIR / f"{key}.json"
    
    if not cache_file.exists():
        return None
    
    try:
        data = json.loads(cache_file.read_text())
        if time.time() - data["timestamp"] > ttl_seconds:
            cache_file.unlink()  # 期限切れのキャッシュを削除
            return None
        return data["value"]
    except (json.JSONDecodeError, KeyError):
        cache_file.unlink(missing_ok=True)
        return None
 
def set_cache(key: str, value) -> None:
    """キャッシュに値を保存する。"""
    cache_file = CACHE_DIR / f"{key}.json"
    cache_file.write_text(json.dumps({
        "timestamp": time.time(),
        "value": value
    }, ensure_ascii=False))