안녕하세요, 저는 HolySheep AI의 기술 엔지니어링 팀에서 일하고 있습니다. 이번 포스트에서는 MCP(Master Control Program) Server에서 Prometheus 메트릭을 노출하는 완전한解决方案를 다룹니다. HolySheep AI는 글로벌 AI API 게이트웨이として機能하며、단일 API 키로 여러 모델을 통합 관리할 수 있습니다.

MCP Server监控의 중요성

MCP Server를 운영할 때 중요한 것은 단순히 요청을 처리하는 것뿐 아니라, 시스템의 건강 상태를 실시간으로 모니터링하는 것입니다. Prometheus는 云原生监控领域的标准工具이며, 다음과 같은 메트릭을 수집할 수 있습니다:

Prometheus 메트릭 타입

Prometheus는 네 가지 주요 메트릭 타입을 지원합니다:

타입설명사용 예시
Counter단조 증가하는 값총 요청 수, 에러 누적
Gauge증가/감소 가능한 값현재 연결 수, 메모리 사용량
Histogram값의 분포를 기록응답 시간 분포
Summary百分位数 기반P99 지연 시간

실제 구현：MCP Server + Prometheus

1. 의존성 설치

# Python 프로젝트의 경우
pip install prometheus-client fastapi uvicorn


Node.js 프로젝트의 경우

npm install prom-client express

2. Python 기반 MCP Server 구현

#!/usr/bin/env python3
"""
MCP Server with Prometheus Metrics
HolySheep AI - Global AI API Gateway Integration
"""

from fastapi import FastAPI, HTTPException
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
import time
import asyncio
from contextlib import asynccontextmanager


Prometheus 메트릭 정의

REQUEST_COUNT = Counter(
    'mcp_server_requests_total',
    'Total number of requests',
    ['endpoint', 'method', 'status']
)

REQUEST_LATENCY = Histogram(
    'mcp_server_request_duration_seconds',
    'Request latency in seconds',
    ['endpoint'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

ACTIVE_CONNECTIONS = Gauge(
    'mcp_server_active_connections',
    'Number of active connections'
)

TOKEN_USAGE = Counter(
    'mcp_server_token_usage_total',
    'Total tokens used',
    ['model', 'type']  # type: prompt/completion
)

ERROR_COUNT = Counter(
    'mcp_server_errors_total',
    'Total number of errors',
    ['error_type', 'endpoint']
)


HolySheep AI API Configuration

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # HolySheep API Key로 교체

@asynccontextmanager
async def lifespan(app: FastAPI):
    """애플리케이션 라이프사이클 관리"""
    ACTIVE_CONNECTIONS.set(0)
    print(f"HolySheep AI Gateway: {HOLYSHEEP_BASE_URL}")
    yield
    print("MCP Server shutting down...")

app = FastAPI(title="MCP Server with Prometheus", lifespan=lifespan)

@app.get("/health")
async def health_check():
    """헬스 체크 엔드포인트"""
    return {"status": "healthy", "service": "mcp-server"}

@app.get("/metrics")
async def metrics():
    """Prometheus 메트릭 엔드포인트"""
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.post("/v1/completions")
async def create_completion(request: dict):
    """AI completion 생성 - HolySheep AI Gateway 사용"""
    ACTIVE_CONNECTIONS.inc()
    start_time = time.time()
    
    try:
        model = request.get("model", "gpt-4.1")
        prompt_tokens = len(str(request.get("prompt", "")))
        
        # HolySheep AI Gateway를 통한 API 호출
        import httpx
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{HOLYSHEEP_BASE_URL}/completions",
                headers={
                    "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model,
                    "prompt": request.get("prompt"),
                    "max_tokens": request.get("max_tokens", 1000)
                },
                timeout=30.0
            )
            
            if response.status_code != 200:
                raise HTTPException(status_code=response.status_code, detail=response.text)
            
            result = response.json()
            
            # 토큰 사용량 기록
            if "usage" in result:
                TOKEN_USAGE.labels(model=model, type="prompt").inc(result["usage"].get("prompt_tokens", 0))
                TOKEN_USAGE.labels(model=model, type="completion").inc(result["usage"].get("completion_tokens", 0))
            
            return result
            
    except Exception as e:
        ERROR_COUNT.labels(error_type=type(e).__name__, endpoint="/v1/completions").inc()
        raise HTTPException(status_code=500, detail=str(e))
    
    finally:
        duration = time.time() - start_time
        REQUEST_LATENCY.labels(endpoint="/v1/completions").observe(duration)
        REQUEST_COUNT.labels(endpoint="/v1/completions", method="POST", status="success").inc()
        ACTIVE_CONNECTIONS.dec()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. Prometheus 설정 파일

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

alerting:
  alertmanagers:
    - static_configs:
        - targets: []

rule_files:
  - "alert_rules.yml"

scrape_configs:
  - job_name: 'mcp-server'
    static_configs:
      - targets: ['localhost:8000']
    metrics

관련 리소스
관련 문서