블로그에 AI 기능 붙이기 (2편) - MCP 서버로 Claude 교육시키기

stateDiagram-v2
    [*] --> Uninitialized: 서버 시작

    Uninitialized --> Cloning: initialize() 호출
    Cloning --> Ready: Clone 성공
    Cloning --> Failed: 인증 실패

    Failed --> Cloning: 재시도

    Ready --> Pulling: Webhook 수신
    Pulling --> Ready: Pull 성공
    Pulling --> Failed: Pull 실패

    Ready --> Serving: AI 요청
    Serving --> Ready: 응답 완료

# mcp-server/src/git_sync.py
import subprocess
from pathlib import Path
import os

class GitContentSync:
    def __init__(self, repo_url, content_dir="/tmp/blog-content"):
        # PAT로 인증 URL 생성 (https://token@github.com/... 형태)
        gh_token = os.getenv("GH_PAT", "")
        if gh_token:
            self.repo_url = repo_url.replace("https://", f"https://{gh_token}@")

        self.content_dir = Path(content_dir)
        self.is_initialized = False

    def initialize(self) -> bool:
        """서버 시작 시 Git Clone"""
        env = os.environ.copy()
        env['GIT_TERMINAL_PROMPT'] = '0'  # 컨테이너에서 프롬프트 비활성화

        result = subprocess.run([
            "git", "clone", "--depth", "1",  # 최근 커밋만 (shallow clone)
            self.repo_url, str(self.content_dir)
        ], env=env, timeout=60)

        return result.returncode == 0

    def pull(self) -> bool:
        """Webhook 수신 시 Git Pull"""
        result = subprocess.run([
            "git", "-C", str(self.content_dir),
            "pull", "origin", "master"
        ], timeout=30)

        return result.returncode == 0

# mcp-server/src/rest_api.py
from fastapi import FastAPI, Request, Header, HTTPException
import hmac
import hashlib

app = FastAPI(title="MCP REST API")

@app.on_event("startup")
async def startup_event():
    """서버 시작할 때 Git Clone"""
    success = initialize_git_sync()
    if success:
        logger.info("✅ Git sync 초기화 완료")
    else:
        logger.error("❌ Git sync 초기화 실패")

@app.post("/webhook/github")
async def github_webhook(
    request: Request,
    x_hub_signature_256: str = Header(None)
):
    """GitHub Webhook 수신 - Push 이벤트 처리"""
    payload_bytes = await request.body()
    payload = await request.json()

    # HMAC 서명 검증 (GitHub에서 온 요청인지 확인)
    expected_signature = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected_signature, x_hub_signature_256):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Push 이벤트만 처리
    event_type = request.headers.get("X-GitHub-Event", "")
    if event_type == "push":
        ref = payload.get("ref", "")

        # master/main 브랜치만 동기화
        if ref in ["refs/heads/master", "refs/heads/main"]:
            git_sync = get_git_sync()
            success = git_sync.pull()

            return {"success": success, "message": "Content synced"}

    return {"success": True, "message": "Ignored"}

# mcp-server/src/tools/search_posts.py
from typing import List, Dict
from ..blog_parser.markdown_parser import get_all_posts

def search_posts(keyword: str) -> List[Dict]:
    """키워드로 포스팅 검색"""
    posts = get_all_posts()
    results = []

    keyword_lower = keyword.lower()

    for post in posts:
        # 제목, 카테고리, 태그, 본문에서 검색
        if (keyword_lower in post.get('title', '').lower() or
            keyword_lower in post.get('category', '').lower() or
            any(keyword_lower in tag.lower() for tag in post.get('tags', [])) or
            keyword_lower in post.get('content', '').lower()):

            results.append({
                'title': post.get('title'),
                'date': post.get('date'),
                'category': post.get('category'),
                'tags': post.get('tags', []),
                'path': post.get('path'),
                'excerpt': post.get('content', '')[:200] + '...'
            })

    return results

# mcp-server/src/tools/get_stats.py
from typing import Dict
from ..blog_parser.markdown_parser import get_all_posts
from collections import defaultdict

def get_stats() -> Dict:
    """블로그 전체 통계"""
    posts = get_all_posts()

    categories = defaultdict(int)
    tags = defaultdict(int)
    years = defaultdict(int)

    for post in posts:
        # 카테고리별 집계
        category = post.get('category', 'Uncategorized')
        categories[category] += 1

        # 태그별 집계
        for tag in post.get('tags', []):
            tags[tag] += 1

        # 연도별 집계
        date = post.get('date', '')
        if date:
            year = date[:4]
            years[year] += 1

    return {
        "total_posts": len(posts),
        "categories": dict(categories),
        "top_tags": dict(sorted(tags.items(), key=lambda x: x[1], reverse=True)[:10]),
        "posts_by_year": dict(years)
    }

FROM python:3.11-slim

WORKDIR /app

# Git 설치 (필수! 이거 없으면 Clone이 안 됩니다)
RUN apt-get update && apt-get install -y \
    git \
    && rm -rf /var/lib/apt/lists/*

# Python 의존성
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 소스 코드
COPY src/ ./src/

# Git Clone할 임시 디렉토리
RUN mkdir -p /tmp/blog-content

EXPOSE 8001

CMD ["python", "-m", "uvicorn", "src.rest_api:app", "--host", "0.0.0.0", "--port", "8001"]

# .github/workflows/deploy-mcp-server.yml (핵심 부분)
gcloud run deploy mcp-api-server \
  --image $IMAGE_NAME \
  --memory 512Mi \
  --min-instances 0 \  # 사용 안 할 때 자동 종료 (비용 절감)
  --max-instances 1 \  # 개인용이라 1개면 충분
  --set-env-vars "WEBHOOK_SECRET=${{ secrets.WEBHOOK_SECRET }},GH_PAT=${{ secrets.GH_PAT }}"

sequenceDiagram
    participant Dev as 개발자
    participant Git as GitHub
    participant Hook as Webhook
    participant MCP as MCP Server
    participant Store as Local Storage

    Dev->>Git: git push origin master

    Note over Git: Push 이벤트 발생

    Git->>Hook: POST /webhook/github<br/>+ HMAC 서명

    Note over Hook: 서명 검증

    Hook->>MCP: 검증 성공

    MCP->>Store: git pull origin master

    Note over Store: 최신 컨텐츠로 업데이트

    Store-->>MCP: Pull 완료
    MCP-->>Hook: 200 OK
    Hook-->>Git: Webhook 처리 완료

    Note over MCP: 새로운 포스팅<br/>즉시 검색 가능!