Build a Personal AI Automation Hub with Claude Code and MCP — Cross-Service Integration Guide

{
  "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 — collects full PR context for Claude Code
import subprocess
import json
import sys
 
def get_pr_context(repo: str, pr_number: int) -> dict:
    """
    Fetches a PR's full context via GitHub CLI.
    Returns diff, metadata, and comments in a single structure.
    
    Limits diff to 500 lines to keep context window usage reasonable
    for large PRs.
    """
    try:
        # Fetch PR metadata
        metadata_result = subprocess.run(
            ["gh", "pr", "view", str(pr_number), "--repo", repo, "--json",
             "title,body,files,additions,deletions,reviewRequests,labels,state"],
            capture_output=True, text=True, check=True
        )
        pr_data = json.loads(metadata_result.stdout)
 
        # Fetch diff, truncated if necessary
        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[... truncated for context ...]"
        else:
            diff_content = diff_result.stdout
 
        return {
            "pr": pr_data,
            "diff": diff_content,
            "error": None
        }
 
    except subprocess.CalledProcessError as e:
        # Surface the CLI error clearly rather than swallowing it
        return {
            "pr": None,
            "diff": None,
            "error": f"GitHub CLI error: {e.stderr.strip()}"
        }
    except json.JSONDecodeError as e:
        return {
            "pr": None,
            "diff": None,
            "error": f"JSON parse error: {str(e)}"
        }
 
if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: 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"Error: {context['error']}", file=sys.stderr)
        sys.exit(1)
 
    print(json.dumps(context, ensure_ascii=False, indent=2))

# issue_triage.py — fetches open issues with enriched context for triage
import subprocess
import json
from datetime import datetime, timezone
 
def fetch_open_issues(repo: str, limit: int = 50) -> list[dict]:
    """
    Fetches open issues sorted by recent activity.
    Enriches each issue with computed fields (age, label names)
    so Claude can make better prioritization decisions without
    re-parsing raw API data.
    """
    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)
 
        for issue in issues:
            # Compute age in days for easier prioritization
            created = datetime.fromisoformat(
                issue["createdAt"].replace("Z", "+00:00")
            )
            issue["age_days"] = (datetime.now(timezone.utc) - created).days
 
            # Flatten nested structures for readability
            issue["label_names"] = [l["name"] for l in issue.get("labels", [])]
            issue["assignee_logins"] = [a["login"] for a in issue.get("assignees", [])]
            issue["milestone_title"] = (
                issue["milestone"]["title"] if issue.get("milestone") else None
            )
 
        return issues
 
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Failed to fetch issues: {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 — creates structured development log pages in Notion
import os
import requests
from datetime import datetime
from typing import Optional
 
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,
    tags: Optional[list[str]] = None
) -> dict:
    """
    Creates a development session log page in Notion.
    
    Returns the created page object on success.
    Raises RuntimeError with Notion's error message on failure.
    """
    today = datetime.now().strftime("%Y-%m-%d")
 
    # Build block content
    children = [
        _heading("Session Summary"),
        _paragraph(summary),
        _heading("Commits"),
        *[_bullet(c) for c in commits],
        _heading("Next Tasks"),
        *[_todo(t) for t in next_tasks],
    ]
 
    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
    }
 
    # Add tags if the database has a multi-select Tags property
    if tags:
        payload["properties"]["Tags"] = {
            "multi_select": [{"name": tag} for tag in tags]
        }
 
    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 error (HTTP {response.status_code}): {error_detail}"
        )
 
    return response.json()
 
# Block builder helpers to keep the main function readable
def _heading(text: str) -> dict:
    return {
        "object": "block", "type": "heading_2",
        "heading_2": {"rich_text": [{"type": "text", "text": {"content": text}}]}
    }
 
def _paragraph(text: str) -> dict:
    return {
        "object": "block", "type": "paragraph",
        "paragraph": {"rich_text": [{"type": "text", "text": {"content": text}}]}
    }
 
def _bullet(text: str) -> dict:
    return {
        "object": "block", "type": "bulleted_list_item",
        "bulleted_list_item": {"rich_text": [{"type": "text", "text": {"content": text}}]}
    }
 
def _todo(text: str) -> dict:
    return {
        "object": "block", "type": "to_do",
        "to_do": {
            "rich_text": [{"type": "text", "text": {"content": text}}],
            "checked": False
        }
    }
 
if __name__ == "__main__":
    # Example run for testing the integration
    try:
        result = create_dev_log(
            title="Auth flow implementation",
            summary="Implemented JWT-based auth. Spent extra time on refresh token TTL configuration.",
            commits=[
                "feat: add JWT authentication middleware",
                "test: add integration tests for auth flow",
                "fix: correct refresh token expiry bug"
            ],
            next_tasks=[
                "Add OAuth2 provider support",
                "Create error page for auth failures"
            ],
            duration_minutes=180,
            tags=["backend", "auth"]
        )
        print(f"✅ Created Notion page: {result.get('url', 'URL unavailable')}")
    except RuntimeError as e:
        print(f"❌ Error: {e}")
        exit(1)

# slack_notifier.py — sends structured messages to Slack with proper error handling
import os
import requests
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:
    """
    Posts a message to Slack.
    
    Rate limit note: Slack's Tier 3 limit is 50 requests/minute.
    When calling this in a loop, add time.sleep(1.2) between calls.
    
    Returns the Slack API response dict on success.
    Raises RuntimeError, ValueError, or PermissionError on failure
    with actionable error messages.
    """
    payload = {
        "channel": channel,
        "text": message  # Required fallback for notifications
    }
    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 returns HTTP 200 even for errors — always check the "ok" field
    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"Rate limited. Retry after {retry_after} seconds."
            )
        elif error_code == "channel_not_found":
            raise ValueError(
                f"Channel '{channel}' not found. Check the channel name."
            )
        elif error_code == "not_in_channel":
            raise PermissionError(
                f"Bot is not in channel '{channel}'. "
                f"Invite the bot: /invite @your-bot-name"
            )
        else:
            raise RuntimeError(f"Slack API error: {error_code}")
 
    return data
 
def send_pr_review_complete(
    pr_title: str,
    pr_url: str,
    reviewer: str,
    summary: str,
    channel: str = DEFAULT_CHANNEL
) -> None:
    """Sends a rich Block Kit notification for a completed PR review."""
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": "✅ PR Review Complete"}
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*PR:*\n<{pr_url}|{pr_title}>"},
                {"type": "mrkdwn", "text": f"*Reviewer:*\n{reviewer}"}
            ]
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*Review Summary:*\n{summary}"}
        }
    ]
    send_notification(
        message=f"PR review complete: {pr_title}",
        channel=channel,
        blocks=blocks
    )

<!-- CLAUDE.md — workflow definitions -->
 
## Morning Standup Routine
 
When asked to "prepare for the day" or "morning standup", execute in order:
 
1. GitHub: fetch open PRs and issues assigned to me
2. Notion: retrieve tasks from the "Dev Tasks" database due today or overdue
3. Slack: summarize unread messages in #dev since yesterday 5pm
4. Output: a prioritized daily plan in English, grouped by urgency
 
## PR Review + Notify Workflow
 
When asked to "wrap up PR review for #[number]":
 
1. GitHub MCP: fetch the PR's diff, comments, and linked issues
2. Notion MCP: look up any linked spec pages (search by PR title keywords)
3. Generate a review summary with: overall assessment, key concerns, suggested changes
4. GitHub: post the summary as a PR review comment
5. Slack: send completion notification to #dev-review
 
## Error Handling Policy
 
If any single service is unavailable:
- Report the error for that service
- Continue processing with the remaining services
- Only abort if all three services are unreachable

#!/bin/bash
# healthcheck.sh — verify all three services before starting work
set -e
 
echo "=== Service Health Check ==="
 
# GitHub: verify auth is active
if gh auth status 2>&1 | grep -q "Logged in"; then
    echo "✅ GitHub: OK"
else
    echo "❌ GitHub: Run 'gh auth login' to reauthenticate"
    FAILED=1
fi
 
# Notion: test API endpoint
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")
[ "$NOTION_STATUS" = "200" ] \
    && echo "✅ Notion: OK" \
    || { echo "❌ Notion: HTTP $NOTION_STATUS — check your API key"; FAILED=1; }
 
# Slack: test bot token
SLACK_OK=$(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'])")
[ "$SLACK_OK" = "True" ] \
    && echo "✅ Slack: OK" \
    || { echo "❌ Slack: Bot token invalid — check SLACK_BOT_TOKEN"; FAILED=1; }
 
[ -n "$FAILED" ] && { echo "⚠️  Fix the above before starting"; exit 1; }
echo "All services reachable."

# simple_cache.py — file-based cache with TTL for stable API responses
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):
    """Returns cached value if it exists and hasn't expired. TTL defaults to 5 minutes."""
    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:
    """Stores a value in the cache with the current timestamp."""
    cache_file = CACHE_DIR / f"{key}.json"
    cache_file.write_text(json.dumps({
        "timestamp": time.time(),
        "value": value
    }, ensure_ascii=False))