The Complete Guide to MCP Server Observability: From Prometheus Metrics and Distributed Trace to Anomaly Detection

MCP Client (에이전트)
    │ tools/call + _meta.traceparent (W3C Trace Context)
    ▼
MCP Server (Python/FastAPI)
    │ OTLP gRPC (트레이스·메트릭·로그)      │ 구조화 JSON 로그
    ▼                                         ▼
OTel Collector                           Datadog Cloud SIEM
    │                                     (이상 탐지 룰)
    ├──▶ Prometheus (메트릭)
    │         │
    │         └──▶ Grafana (대시보드·알람)
    │
    ├──▶ Grafana Tempo (분산 추적)
    │
    └──▶ Loki (로그)

{mcp.method.name} {target}
예: tools/call weather_tool

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": { "city": "Seoul" },
    "_meta": {
      "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
      "tracestate": "vendor=myapp"
    }
  },
  "id": 1
}

from opentelemetry.propagate import extract
from opentelemetry.trace import get_tracer
from typing import Any, TypedDict
 
tracer = get_tracer("mcp-server")
 
 
class McpParams(TypedDict, total=False):
    name: str
    arguments: dict[str, Any]
    _meta: dict[str, str]
 
 
class McpRequest(TypedDict):
    jsonrpc: str
    method: str
    params: McpParams
    id: int | str
 
 
def execute_tool(params: McpParams) -> Any:
    """실제 Tool 실행 로직으로 대체하세요."""
    raise NotImplementedError(f"Tool '{params.get('name')}' not implemented")
 
 
def handle_tool_call(request: McpRequest) -> Any:
    """_meta에서 W3C Trace Context를 추출해 OTel 스팬과 연결합니다."""
    meta = request.get("params", {}).get("_meta", {})
    carrier = {
        "traceparent": meta.get("traceparent", ""),
        "tracestate": meta.get("tracestate", ""),
    }
    ctx = extract(carrier)
    tool_name = request["params"].get("name", "unknown")
 
    with tracer.start_as_current_span(
        f"tools/call {tool_name}",
        context=ctx,
        attributes={
            "mcp.method.name": "tools/call",
            "mcp.tool.name": tool_name,
        },
    ):
        return execute_tool(request["params"])

from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
from prometheus_client import Histogram, Counter
from typing import Any, TypedDict
import time
 
app = FastAPI()
 
 
class ToolCallRequest(TypedDict):
    params: dict[str, Any]
 
 
# MCP 전용 커스텀 메트릭 정의
# 주의: 낮은 카디널리티 레이블(tool_name, status)만 사용합니다.
# user_id처럼 고유 값이 많은 필드는 Prometheus 레이블이 아닌
# 트레이스 속성이나 로그 필드로 기록하세요.
tool_duration = Histogram(
    "mcp_tool_invocation_duration_seconds",
    "MCP Tool Call 실행 시간",
    labelnames=["tool_name", "status"],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0],
)
 
tool_errors = Counter(
    "mcp_tool_errors_total",
    "MCP Tool Call 에러 총 횟수",
    labelnames=["tool_name", "error_type"],
)
 
tool_calls = Counter(
    "mcp_tool_calls_total",
    "MCP Tool Call 총 호출 수",
    labelnames=["tool_name"],
)
 
# FastAPI 기본 메트릭 자동 계측
Instrumentator().instrument(app).expose(app)
 
 
async def execute_tool(tool_name: str, arguments: dict[str, Any]) -> Any:
    """실제 Tool 실행 로직으로 대체하세요."""
    raise NotImplementedError(f"Tool '{tool_name}' not implemented")
 
 
@app.post("/mcp/tools/call")
async def call_tool(request: ToolCallRequest) -> Any:
    tool_name = request["params"].get("name", "unknown")
    tool_calls.labels(tool_name=tool_name).inc()
 
    start = time.time()
    try:
        result = await execute_tool(
            tool_name, request["params"].get("arguments", {})
        )
        duration = time.time() - start
        tool_duration.labels(tool_name=tool_name, status="success").observe(duration)
        return result
    except Exception as e:
        duration = time.time() - start
        tool_duration.labels(tool_name=tool_name, status="error").observe(duration)
        tool_errors.labels(tool_name=tool_name, error_type=type(e).__name__).inc()
        raise

# Tool별 p95 레이턴시
histogram_quantile(
  0.95,
  rate(mcp_tool_invocation_duration_seconds_bucket[5m])
) by (tool_name)
 
# 분당 에러율
rate(mcp_tool_errors_total[1m]) by (tool_name, error_type)
 
# Tool별 호출량 추이
rate(mcp_tool_calls_total[10m]) by (tool_name)

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  # MCP 서버 Prometheus /metrics 스크랩
  prometheus:
    config:
      scrape_configs:
        - job_name: "mcp-server"
          scrape_interval: 15s
          static_configs:
            - targets: ["mcp-server:8000"]
 
processors:
  # PII 마스킹 — Tool 인자에서 민감 정보 제거
  # replace_pattern()은 OTTL(OpenTelemetry Transformation Language) 문법입니다.
  # 아래 정규식의 YAML 이스케이프 처리는 실제 환경에서 반드시 검증하세요.
  transform/mask_pii:
    error_mode: ignore
    trace_statements:
      - context: span
        statements:
          - replace_pattern(attributes["mcp.tool.arguments"],
              "\"password\"\\s*:\\s*\"[^\"]+\"",
              "\"password\": \"[REDACTED]\"")
 
  # 배치 처리로 오버헤드 감소
  batch:
    send_batch_size: 1000
    timeout: 10s
 
  # 메모리 제한
  memory_limiter:
    check_interval: 1s
    limit_mib: 512
 
exporters:
  # Grafana Tempo로 트레이스 전송
  otlp/tempo:
    endpoint: tempo:4317
    tls:
      insecure: true
 
  # Prometheus Remote Write로 메트릭 전송
  prometheusremotewrite:
    endpoint: "http://prometheus:9090/api/v1/write"
 
  # Loki로 로그 전송
  loki:
    endpoint: http://loki:3100/loki/api/v1/push
 
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [transform/mask_pii, batch]
      exporters: [otlp/tempo]
    metrics:
      receivers: [otlp, prometheus]
      processors: [batch, memory_limiter]
      exporters: [prometheusremotewrite]
    logs:
      receivers: [otlp]
      processors: [transform/mask_pii, batch]
      exporters: [loki]

import { trace, context, SpanStatusCode } from "@opentelemetry/api";
import { W3CTraceContextPropagator } from "@opentelemetry/core";
 
const tracer = trace.getTracer("mcp-agent", "1.0.0");
const propagator = new W3CTraceContextPropagator();
 
interface ToolPlan {
  name: string;
  sessionId: string;
  args: Record<string, unknown>;
}
 
// Step 1에서 구성한 MCP 서버의 /mcp/tools/call 엔드포인트를 호출합니다.
async function callMcpServer(request: unknown): Promise<unknown> {
  // 예: fetch("http://mcp-server:8000/mcp/tools/call", { body: JSON.stringify(request) })
  throw new Error("callMcpServer: 실제 HTTP 요청 로직으로 대체하세요");
}
 
async function planToolCalls(userQuery: string): Promise<ToolPlan[]> {
  // 실제 LLM 추론 로직으로 대체하세요
  return [];
}
 
async function runAgentWithObservability(userQuery: string): Promise<void> {
  // 에이전트 추론 스팬 시작 (루트)
  return await tracer.startActiveSpan(
    "gen_ai.agent reasoning",
    {
      attributes: {
        "gen_ai.system": "anthropic",
        "gen_ai.request.model": "claude-sonnet-4-6",
        "user.query": userQuery,
      },
    },
    async (agentSpan) => {
      try {
        const toolsToCall = await planToolCalls(userQuery);
 
        for (const tool of toolsToCall) {
          // Tool Call 스팬을 에이전트 스팬의 자식으로 생성
          await tracer.startActiveSpan(
            `tools/call ${tool.name}`,
            {
              attributes: {
                "mcp.method.name": "tools/call",
                "mcp.tool.name": tool.name,
                "mcp.session.id": tool.sessionId,
              },
            },
            async (toolSpan) => {
              try {
                // _meta에 traceparent 주입 (W3C Trace Context 전파)
                const carrier: Record<string, string> = {};
                propagator.inject(context.active(), carrier);
 
                const mcpRequest = {
                  jsonrpc: "2.0",
                  method: "tools/call",
                  params: {
                    name: tool.name,
                    arguments: tool.args,
                    _meta: {
                      traceparent: carrier["traceparent"] ?? "",
                      tracestate: carrier["tracestate"] ?? "",
                    },
                  },
                  id: crypto.randomUUID(), // Node.js 15+ 필요
                };
 
                const result = await callMcpServer(mcpRequest);
                toolSpan.setStatus({ code: SpanStatusCode.OK });
                return result;
              } catch (err) {
                toolSpan.setStatus({
                  code: SpanStatusCode.ERROR,
                  message: String(err),
                });
                toolSpan.recordException(err as Error);
                throw err;
              } finally {
                toolSpan.end();
              }
            }
          );
        }
 
        agentSpan.setStatus({ code: SpanStatusCode.OK });
      } finally {
        agentSpan.end();
      }
    }
  );
}

gen_ai.agent reasoning (350ms)
├── tools/call get_weather (45ms)
├── tools/call query_database (180ms)
│   └── db.query SELECT ... (120ms)
└── tools/call send_report (80ms)

import logging
import json
from datetime import datetime, timezone
from typing import Optional
 
logger = logging.getLogger("mcp.security")
 
 
def log_tool_call(
    tool_name: str,
    user_id: str,
    session_id: str,
    status: str,
    duration_ms: float,
    error: Optional[str] = None,
) -> None:
    """Datadog SIEM이 파싱할 수 있는 구조화 로그를 출력합니다."""
    log_entry: dict = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event.category": "mcp.tool_call",
        "mcp.tool.name": tool_name,
        "usr.id": user_id,            # Datadog 표준 사용자 속성
        "mcp.session.id": session_id,
        "http.status": 200 if status == "success" else 500,
        "duration_ms": duration_ms,
        "status": status,
    }
    if error:
        log_entry["error.message"] = error
        log_entry["error.kind"] = "MCPToolError"
 
    logger.info(json.dumps(log_entry))

# 탐지 룰 1: 특정 Tool 호출 빈도 급증
# 사용자별 Tool Call이 기준 편차 3σ를 초과할 때 경보
@event.category:mcp.tool_call
| anomaly(count, direction:above, threshold:3)
  by usr.id, mcp.tool.name
  over 5m
 
# 탐지 룰 2: 연속 인증 실패 후 Tool 호출 시도
# 같은 세션에서 401 이후 5분 내 tools/call 탐지
@event.category:mcp.tool_call AND @http.status:401
| sequence by mcp.session.id
  [within 5m] @http.status:401
  [within 5m] @event.category:mcp.tool_call
 
# 탐지 룰 3: 비정상 시간대 고빈도 호출 (야간 자동화 탐지)
@event.category:mcp.tool_call
| @timestamp:[23:00 TO 06:00]
| count > 100 by usr.id over 10m