챗봇: MCP 가드레일과 평가 핸드북 2026

"""
MCP Server 예시: 고객 주문 조회와 FAQ 검색 도구를 제공한다.
mcp 패키지(v1.2+)를 사용한다.
pip install mcp
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json
from typing import Any

# MCP Server 인스턴스 생성
app = Server("customer-support-tools")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """MCP Client가 사용 가능한 도구 목록을 요청할 때 호출된다."""
    return [
        Tool(
            name="order_lookup",
            description="주문 번호로 주문 상태를 조회합니다. 주문 번호는 ORD-로 시작하는 형식입니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "주문 번호 (예: ORD-20260304-001)",
                        "pattern": r"^ORD-\d{8}-\d{3,6}$",
                    },
                },
                "required": ["order_id"],
                "additionalProperties": False,
            },
        ),
        Tool(
            name="faq_search",
            description="고객 지원 FAQ에서 관련 답변을 검색합니다.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색할 질문",
                        "minLength": 2,
                        "maxLength": 200,
                    },
                    "category": {
                        "type": "string",
                        "enum": ["payment", "shipping", "return", "account"],
                        "description": "FAQ 카테고리",
                    },
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """MCP Client가 도구 실행을 요청할 때 호출된다."""
    if name == "order_lookup":
        # 실제 구현에서는 DB 조회
        order_id = arguments["order_id"]
        result = await lookup_order_from_db(order_id)
        return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]

    elif name == "faq_search":
        query = arguments["query"]
        category = arguments.get("category")
        results = await search_faq_index(query, category)
        return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False))]

    else:
        raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

"""
MCP Gateway: MCP Client -> Gateway -> MCP Server 사이에서
모든 tool call을 가로채어 가드레일 검증을 수행한다.
"""
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
from enum import Enum
import re
import json
import logging

logger = logging.getLogger("mcp_gateway")

class GatewayDecision(Enum):
    ALLOW = "allow"
    DENY = "deny"
    REQUIRE_APPROVAL = "require_approval"
    RATE_LIMITED = "rate_limited"

@dataclass
class GatewayPolicy:
    """MCP Gateway 정책 정의"""
    # 도구별 접근 제어
    tool_permissions: Dict[str, List[str]] = field(default_factory=dict)
    # 예: {"order_lookup": ["customer", "agent", "admin"],
    #      "order_cancel": ["agent", "admin"]}

    # 도구별 호출 제한
    rate_limits: Dict[str, int] = field(default_factory=dict)
    # 예: {"order_lookup": 30, "faq_search": 60}  # per minute

    # 인자 정제 규칙
    argument_sanitizers: Dict[str, dict] = field(default_factory=dict)

    # 위험 도구 목록 (사용자 확인 필요)
    high_risk_tools: List[str] = field(default_factory=list)

    # 차단 패턴
    blocked_argument_patterns: List[str] = field(default_factory=list)

# 기본 정책
DEFAULT_POLICY = GatewayPolicy(
    tool_permissions={
        "order_lookup": ["customer", "agent", "admin"],
        "faq_search": ["customer", "agent", "admin"],
        "order_cancel": ["agent", "admin"],
        "refund_process": ["admin"],
        "user_data_export": ["admin"],
    },
    rate_limits={
        "order_lookup": 30,
        "faq_search": 60,
        "order_cancel": 5,
        "refund_process": 3,
        "user_data_export": 1,
    },
    high_risk_tools=["order_cancel", "refund_process", "user_data_export"],
    blocked_argument_patterns=[
        r"(?i)(drop|delete|truncate)\s+(table|database)",
        r"(?i)(union\s+select|or\s+1\s*=\s*1)",
        r"\.\./",  # path traversal
        r"<script",  # XSS
    ],
)

class MCPGateway:
    def __init__(self, policy: GatewayPolicy = DEFAULT_POLICY):
        self.policy = policy
        self._call_history: Dict[str, List[datetime]] = {}

    def evaluate(
        self,
        tool_name: str,
        arguments: Dict[str, Any],
        user_role: str,
        session_id: str,
    ) -> dict:
        """
        Tool call에 대한 가드레일 검증을 수행한다.
        Returns: {"decision": GatewayDecision, "reason": str, "sanitized_args": dict}
        """
        # 1. Permission 검증
        allowed_roles = self.policy.tool_permissions.get(tool_name, [])
        if not allowed_roles:
            return self._deny(f"Tool '{tool_name}' is not registered in gateway policy")
        if user_role not in allowed_roles:
            return self._deny(
                f"Role '{user_role}' is not authorized for tool '{tool_name}'. "
                f"Required: {allowed_roles}"
            )

        # 2. Rate limit 검증
        rate_limit = self.policy.rate_limits.get(tool_name, 60)
        if self._is_rate_limited(session_id, tool_name, rate_limit):
            return {
                "decision": GatewayDecision.RATE_LIMITED,
                "reason": f"Rate limit exceeded: {rate_limit}/min for '{tool_name}'",
                "sanitized_args": arguments,
            }

        # 3. Argument sanitization
        sanitized_args, blocked_patterns = self._sanitize_arguments(arguments)
        if blocked_patterns:
            logger.warning(
                f"Blocked patterns detected in tool '{tool_name}': {blocked_patterns}"
            )
            return self._deny(
                f"Potentially malicious argument patterns detected: {blocked_patterns}"
            )

        # 4. High-risk tool 확인
        if tool_name in self.policy.high_risk_tools:
            return {
                "decision": GatewayDecision.REQUIRE_APPROVAL,
                "reason": f"Tool '{tool_name}' requires user confirmation before execution",
                "sanitized_args": sanitized_args,
            }

        return {
            "decision": GatewayDecision.ALLOW,
            "reason": "All checks passed",
            "sanitized_args": sanitized_args,
        }

    def _is_rate_limited(self, session_id: str, tool_name: str, limit: int) -> bool:
        key = f"{session_id}:{tool_name}"
        now = datetime.utcnow()
        cutoff = now - timedelta(minutes=1)

        if key not in self._call_history:
            self._call_history[key] = []

        self._call_history[key] = [
            t for t in self._call_history[key] if t > cutoff
        ]

        if len(self._call_history[key]) >= limit:
            return True

        self._call_history[key].append(now)
        return False

    def _sanitize_arguments(self, arguments: Dict[str, Any]) -> tuple:
        blocked = []
        args_str = json.dumps(arguments)

        for pattern in self.policy.blocked_argument_patterns:
            if re.search(pattern, args_str):
                blocked.append(pattern)

        return arguments, blocked

    def _deny(self, reason: str) -> dict:
        return {
            "decision": GatewayDecision.DENY,
            "reason": reason,
            "sanitized_args": {},
        }

"""
MCP 도구의 기능적 정확성을 평가하는 테스트 스위트.
각 도구에 대해 정상 케이스, 엣지 케이스, 에러 케이스를 검증한다.
"""
import pytest
from dataclasses import dataclass
from typing import Any, Optional, List

@dataclass
class ToolTestCase:
    """단일 도구 테스트 케이스"""
    test_id: str
    tool_name: str
    arguments: dict
    expected_status: str  # success, error, validation_error
    expected_output_contains: Optional[str] = None
    expected_error_type: Optional[str] = None
    description: str = ""

TOOL_TEST_CASES = [
    # 정상 케이스
    ToolTestCase(
        test_id="order_lookup_valid",
        tool_name="order_lookup",
        arguments={"order_id": "ORD-20260304-001"},
        expected_status="success",
        expected_output_contains="order_status",
        description="유효한 주문 번호로 조회",
    ),
    # 엣지 케이스: 존재하지 않는 주문
    ToolTestCase(
        test_id="order_lookup_not_found",
        tool_name="order_lookup",
        arguments={"order_id": "ORD-99991231-999"},
        expected_status="success",
        expected_output_contains="not_found",
        description="존재하지 않는 주문 번호 조회 시 정상 응답",
    ),
    # 에러 케이스: 잘못된 형식
    ToolTestCase(
        test_id="order_lookup_invalid_format",
        tool_name="order_lookup",
        arguments={"order_id": "INVALID-FORMAT"},
        expected_status="validation_error",
        expected_error_type="pattern_mismatch",
        description="주문 번호 형식이 잘못된 경우",
    ),
    # 보안 케이스: SQL injection 시도
    ToolTestCase(
        test_id="order_lookup_sql_injection",
        tool_name="order_lookup",
        arguments={"order_id": "ORD-20260304-001' OR '1'='1"},
        expected_status="validation_error",
        expected_error_type="pattern_mismatch",
        description="SQL injection 시도 차단",
    ),
    # FAQ 정상 검색
    ToolTestCase(
        test_id="faq_search_valid",
        tool_name="faq_search",
        arguments={"query": "환불 절차", "category": "return"},
        expected_status="success",
        expected_output_contains="환불",
        description="유효한 FAQ 검색",
    ),
    # FAQ 빈 결과
    ToolTestCase(
        test_id="faq_search_no_results",
        tool_name="faq_search",
        arguments={"query": "xyzzyspoon"},
        expected_status="success",
        expected_output_contains="no_results",
        description="검색 결과 없음 시 정상 빈 응답",
    ),
]

@pytest.mark.parametrize("case", TOOL_TEST_CASES, ids=lambda c: c.test_id)
async def test_mcp_tool(mcp_client, case):
    """MCP 도구의 기능적 정확성을 검증한다."""
    try:
        result = await mcp_client.call_tool(case.tool_name, case.arguments)

        if case.expected_status == "validation_error":
            pytest.fail(
                f"Expected validation error but tool executed successfully: {result}"
            )

        if case.expected_output_contains:
            result_text = str(result)
            assert case.expected_output_contains in result_text, (
                f"Expected output to contain '{case.expected_output_contains}', "
                f"got: {result_text[:200]}"
            )

    except Exception as e:
        if case.expected_status == "validation_error":
            if case.expected_error_type:
                assert case.expected_error_type in str(type(e).__name__).lower() or \
                       case.expected_error_type in str(e).lower(), (
                    f"Expected error type '{case.expected_error_type}', got: {type(e).__name__}: {e}"
                )
        else:
            raise

"""
MCP Gateway 가드레일의 효과를 정량적으로 측정한다.
공격 시나리오 데이터셋에 대한 차단율, 오탐율을 계산한다.
"""
from typing import List, Dict
from dataclasses import dataclass

@dataclass
class GuardrailEvalResult:
    total_scenarios: int
    true_positives: int     # 공격을 정상 차단
    false_positives: int    # 정상을 오차단
    true_negatives: int     # 정상을 정상 통과
    false_negatives: int    # 공격을 미차단
    precision: float
    recall: float
    f1_score: float

def evaluate_guardrails(
    gateway,
    attack_scenarios: List[dict],
    benign_scenarios: List[dict],
) -> GuardrailEvalResult:
    """
    공격 시나리오와 정상 시나리오에 대해 가드레일의 정밀도/재현율을 측정한다.
    """
    tp = fp = tn = fn = 0

    # 공격 시나리오: 차단되어야 정상
    for scenario in attack_scenarios:
        result = gateway.evaluate(
            tool_name=scenario["tool_name"],
            arguments=scenario["arguments"],
            user_role=scenario.get("user_role", "customer"),
            session_id=scenario.get("session_id", "eval"),
        )
        if result["decision"] in (GatewayDecision.DENY, GatewayDecision.RATE_LIMITED):
            tp += 1
        else:
            fn += 1
            print(f"MISS: Attack scenario not blocked: {scenario.get('name', 'unknown')}")

    # 정상 시나리오: 통과되어야 정상
    for scenario in benign_scenarios:
        result = gateway.evaluate(
            tool_name=scenario["tool_name"],
            arguments=scenario["arguments"],
            user_role=scenario.get("user_role", "customer"),
            session_id=scenario.get("session_id", "eval"),
        )
        if result["decision"] == GatewayDecision.ALLOW:
            tn += 1
        elif result["decision"] == GatewayDecision.REQUIRE_APPROVAL:
            tn += 1  # high-risk 도구의 확인 요청은 정상 동작
        else:
            fp += 1
            print(f"FALSE POSITIVE: Benign scenario blocked: {scenario.get('name', 'unknown')}")

    total = tp + fp + tn + fn
    precision = tp / max(tp + fp, 1)
    recall = tp / max(tp + fn, 1)
    f1 = 2 * precision * recall / max(precision + recall, 1e-9)

    return GuardrailEvalResult(
        total_scenarios=total,
        true_positives=tp,
        false_positives=fp,
        true_negatives=tn,
        false_negatives=fn,
        precision=round(precision, 4),
        recall=round(recall, 4),
        f1_score=round(f1, 4),
    )

# prometheus-rules.yaml
groups:
  - name: mcp_server_monitoring
    interval: 15s
    rules:
      - alert: MCPToolCallErrorRateHigh
        expr: |
          sum(rate(mcp_tool_call_total{status="error"}[5m])) by (tool_name) /
          sum(rate(mcp_tool_call_total[5m])) by (tool_name) > 0.05
        for: 3m
        labels:
          severity: warning
          team: chatbot
        annotations:
          summary: 'MCP 도구 {{ $labels.tool_name }}의 에러율이 5%를 초과합니다'
          description: |
            현재 에러율: {{ $value | humanizePercentage }}
            도구 구현 또는 하위 서비스 상태를 확인하세요.

      - alert: MCPGatewayBlockRateSpike
        expr: |
          sum(rate(mcp_gateway_decision_total{decision="deny"}[15m])) /
          sum(rate(mcp_gateway_decision_total[15m])) > 0.20
        for: 5m
        labels:
          severity: warning
          team: security
        annotations:
          summary: 'MCP Gateway 차단율이 20%를 초과합니다'
          description: '공격 시도 증가 또는 정책 오설정 가능성. 감사 로그를 확인하세요.'

      - alert: MCPToolLatencyHigh
        expr: |
          histogram_quantile(0.95,
            rate(mcp_tool_call_duration_seconds_bucket[5m])
          ) > 3.0
        for: 3m
        labels:
          severity: warning
          team: chatbot
        annotations:
          summary: 'MCP 도구 호출 p95 지연이 3초를 초과합니다'

증상: 챗봇이 모든 도구 호출에 실패. 사용자에게 "기능을 사용할 수 없습니다" 반복 표시.
에러 로그:
  MCPConnectionError: Failed to connect to MCP server at unix:///tmp/mcp-customer-support.sock
  Timeout after 5000ms

원인: MCP Server 프로세스가 OOM으로 종료됨

해결:
  1. MCP Server 프로세스 재시작 (systemd 또는 supervisor)
  2. 메모리 제한 설정 및 OOM killer 우선순위 조정
  3. Graceful degradation: MCP Server 장애 시 도구 없이 일반 대화 모드로 전환
  4. Health check 엔드포인트 추가 및 자동 재시작 설정

증상: "주문 취소해 주세요"라는 정상 요청이 Gateway에서 차단됨
에러 로그:
  Gateway DENY: Blocked pattern detected - "취소" matched "cancel" block rule

원인: blocked_argument_patterns에 과도하게 넓은 패턴이 포함됨
     ("cancel"이라는 단어가 인자뿐 아니라 도구 설명에서도 매칭)

해결:
  1. 패턴 매칭 범위를 인자 값으로 한정 (도구 이름/설명 제외)
  2. 차단 패턴 추가 시 정상 시나리오 회귀 테스트 필수화
  3. 오탐 발생 시 즉시 패턴 비활성화할 수 있는 핫픽스 절차 마련

증상: CS 상담사가 대량 주문 조회 시 rate limit에 걸려 업무 지연
에러 로그:
  Gateway RATE_LIMITED: Rate limit exceeded: 30/min for 'order_lookup'

해결:
  1. 역할별 rate limit 차등 적용 (customer: 30/min, agent: 200/min, admin: 무제한)
  2. 배치 조회 도구 (order_batch_lookup) 별도 제공
  3. Rate limit 히스토리를 대시보드에 표시하여 임계치 튜닝 근거 확보