사내 MCP 가드레일 레이어 구축 - 모든 에이전트가 공유하는 중앙 방어 허브 설계

들어가며

지난 세 편에서 1인 팀도 적용 가능한 AI 에이전트 방어 스택을 완성했습니다.

• #127 방어선 구축 - 비용·폭주·드리프트
• #128 LLM-as-Judge - 품질 자동 채점
• #129 보안 가드레일 - PII·인젝션·탈옥 방어

스택이 커지면서 현실의 병목이 드러납니다. "에이전트가 20개인데 가드레일 코드를 20곳에 복붙하고 있다"는 게 전형적입니다. 한 곳 고치면 19곳이 뒤처지고, 버전이 뒤섞이면 보안 회귀(regression)가 매 주 생깁니다.

해결책은 가드레일을 중앙 MCP 서버로 밀어내는 것입니다. 에이전트는 이 서버에 tool 호출 한 번으로 "이 프롬프트 검증해 줘"만 요청하고, 반응하는 로직은 전부 중앙에서 관리합니다. 오늘 글은 이 허브를 코드 레벨로 구축합니다.

• 1부 - 왜 MCP 허브인가, 직렬 vs. 사이드카 선택
• 2부 - FastAPI 기반 MCP 가드레일 서버 구현
• 3부 - Spring Boot 백엔드에서 소비하는 클라이언트 예시
• 4부 - 정책 버저닝·롤백·카나리 배포
• 5부 - 관측·감사 로그·조직 단위 분리

1. 왜 중앙 허브인가

가드레일 중복의 비용은 조용히 누적됩니다.

분산형(지금까지)	중앙형(목표)
에이전트마다 PII 코드 복붙	MCP 서버 1곳에서 관리
정책 변경 시 N회 배포	1회 배포
위협 인텔 업데이트 지연	즉시 전체 반영
에이전트별 우회 가능	경로 강제 차단
감사 로그 파편화	통합 감사

특히 위협 인텔 반영 속도가 결정적입니다. 새 탈옥 패턴이 유출되면 10분 안에 전 조직 에이전트에 반영되어야 합니다. 분산형으로는 절대 불가능합니다.

MCP가 답인 이유

HTTP API로도 할 수는 있지만 MCP(Model Context Protocol)를 쓰는 실질적 이유는 세 가지입니다.

• 에이전트 런타임이 이미 MCP 클라이언트. Claude Code·Cursor·각종 프레임워크가 표준 지원
• 도구 메타데이터 표준화. 스키마·권한·레이트 리밋이 프로토콜 레벨
• 스트리밍·세션 상태 지원. 장시간 검증이나 배치 채점에 유리

MCP 완벽 가이드와 MCP 서버 직접 만들기를 전제로 오늘 글은 "그 위에 가드레일 전용 허브를 올리는" 구조를 다룹니다.

2. 아키텍처 선택 - 직렬 vs. 사이드카

중앙화에는 두 가지 패턴이 있습니다.

패턴 A - 직렬(Inline) 프록시

[에이전트] → [가드레일 MCP] → [Claude/GPT API]

모든 LLM 호출이 허브를 경유. 강제력은 최강이지만 허브 장애가 전체 정지로 이어집니다.

패턴 B - 사이드카(Advisory) 호출

[에이전트] →(tool: check)→ [가드레일 MCP]
         ↓
        [Claude/GPT API]

에이전트가 자발적으로 check 도구를 먼저 부른 뒤 API로 진입. 장애 내성은 좋지만 강제력은 약합니다.

현실적 선택: 하이브리드

• 외부 서비스 대면 에이전트 - 직렬 프록시 강제
• 내부 개발자용 에이전트 - 사이드카 권고
• 샌드박스/테스트 - 사이드카, 로그만

오늘은 양쪽 모두 다루지만 구현 디테일은 사이드카를 중심으로 갑니다. 직렬 프록시 흐름은 지난 #127의 프록시 예시를 확장하면 됩니다.

3. MCP 가드레일 서버 스펙

서버가 노출할 도구 4개와 리소스 2개를 먼저 정의합니다.

3-1. Tool 목록

도구	입력	출력
guardrail.check_input	text, agent_id	allowed, masked_text, findings
guardrail.check_output	text, agent_id, trace_id	sanitized_text, violations
guardrail.check_tool_call	tool, args, agent_id	allowed, reason
guardrail.judge	task, response, rubric_id	overall, scores, rationale

3-2. Resource 목록

• guardrail://policies/{agent_id} - 현재 적용 중인 정책 버전
• guardrail://rubrics/{rubric_id} - 판사 루브릭 원문

4. FastAPI + MCP SDK 구현

Python mcp SDK로 서버를 구현합니다. 각 도구는 지난 편 코드를 호출하는 얇은 래퍼입니다.

4-1. 프로젝트 구조

guardrail-hub/
├── server.py          # MCP 엔트리포인트
├── policies.py        # 에이전트별 정책 로딩
├── core/
│   ├── pii.py         # #129 pii_scanner 재사용
│   ├── injection.py   # #129 인젝션 탐지
│   ├── tool_guard.py  # 도구 allowlist
│   └── judge.py       # #128 판사
├── audit.py           # 감사 로그
├── metrics.py         # Prometheus
├── policies/
│   ├── default.yaml
│   ├── customer-facing.yaml
│   └── internal-dev.yaml
└── rubrics/
    ├── backend.yaml
    └── support.yaml

4-2. MCP 서버 엔트리포인트

# server.py
from mcp.server.fastmcp import FastMCP
from core.pii import scan_and_mask
from core.injection import detect_injection
from core.tool_guard import check_tool
from core.judge import judge_with_rubric
from policies import load_policy
from audit import record
from metrics import (
    pii_hits, injection_blocks, tool_blocks
)

mcp = FastMCP("guardrail-hub")

@mcp.tool()
async def check_input(text: str, agent_id: str,
                     session_id: str = "") -> dict:
    policy = load_policy(agent_id)
    masked, findings = scan_and_mask(text, policy)
    for f in findings:
        pii_hits.labels(f["kind"], "input").inc()

    inj = detect_injection(masked, policy)
    if inj.get("blocked"):
        injection_blocks.labels(inj["detector"]).inc()
        record(agent_id, "check_input", {
            "blocked": True, "detector": inj["detector"],
            "session_id": session_id
        })
        return {
            "allowed": False,
            "reason": inj.get("reason"),
            "detector": inj["detector"],
        }

    record(agent_id, "check_input", {
        "blocked": False, "pii": len(findings),
        "session_id": session_id
    })
    return {
        "allowed": True,
        "masked_text": masked,
        "pii_kinds": [f["kind"] for f in findings],
    }

@mcp.tool()
async def check_output(text: str, agent_id: str,
                     trace_id: str = "") -> dict:
    policy = load_policy(agent_id)
    masked, findings = scan_and_mask(text, policy)
    for f in findings:
        pii_hits.labels(f["kind"], "output").inc()
    record(agent_id, "check_output", {
        "pii": len(findings), "trace_id": trace_id
    })
    return {
        "sanitized_text": masked,
        "pii_kinds": [f["kind"] for f in findings],
        "had_pii": bool(findings),
    }

@mcp.tool()
async def check_tool_call(tool: str, args: dict,
                        agent_id: str) -> dict:
    policy = load_policy(agent_id)
    ok, reason = check_tool(tool, args, policy)
    if not ok:
        tool_blocks.labels(tool, reason).inc()
    record(agent_id, "check_tool_call", {
        "tool": tool, "allowed": ok, "reason": reason
    })
    return {"allowed": ok, "reason": reason}

@mcp.tool()
async def judge(task: str, response: str,
              rubric_id: str = "backend") -> dict:
    return judge_with_rubric(task, response, rubric_id)

@mcp.resource("guardrail://policies/{agent_id}")
async def get_policy(agent_id: str) -> str:
    return load_policy(agent_id).dump_yaml()

@mcp.resource("guardrail://rubrics/{rubric_id}")
async def get_rubric(rubric_id: str) -> str:
    with open(f"rubrics/{rubric_id}.yaml") as f:
        return f.read()

if __name__ == "__main__":
    mcp.run(transport="sse", port=8765)

4-3. 정책 파일 예시

# policies/customer-facing.yaml
agent_id_prefix: "customer-"
pii:
  mask_kinds: [resident_id, phone_kr, email, card, business_id]
  block_on_output_leak: true
injection:
  signature_set: strict
  llm_detector: claude-haiku-4-5
  block_threshold: 0.6
tools:
  allowlist:
    - name: search_order
      arg_pattern: "^order_id=\\d{10}$"
    - name: send_email
      arg_pattern: ".*@(mycompany|partner)\\.com$"
  denylist_always:
    - shell
    - filesystem.write

# policies/internal-dev.yaml
agent_id_prefix: "dev-"
pii:
  mask_kinds: [resident_id, card, aws_key]
  block_on_output_leak: false
injection:
  signature_set: medium
  llm_detector: null
  block_threshold: 0.8
tools:
  allowlist:
    - name: shell
      arg_pattern: "^(ls|cat|git|grep)"
    - name: filesystem.read
      arg_pattern: "^/app/"

같은 가드레일 서버가 정책만 다르게 고객 대면 에이전트와 개발자용 에이전트를 커버합니다.

4-4. 정책 핫리로드

# policies.py
import yaml, os, time
from threading import RLock

_POLICY_DIR = "policies"
_cache = {}
_mtime = {}
_lock = RLock()

def load_policy(agent_id: str):
    with _lock:
        # agent_id → 파일 매핑
        fname = _resolve_file(agent_id)
        path = os.path.join(_POLICY_DIR, fname)
        m = os.path.getmtime(path)
        if _mtime.get(fname) != m:
            with open(path) as f:
                _cache[fname] = yaml.safe_load(f)
            _mtime[fname] = m
        return Policy(_cache[fname])

def _resolve_file(agent_id: str) -> str:
    if agent_id.startswith("customer-"):
        return "customer-facing.yaml"
    if agent_id.startswith("dev-"):
        return "internal-dev.yaml"
    return "default.yaml"

파일 mtime 기반 재로드라 재시작 없이 정책이 반영됩니다. 10초면 모든 에이전트가 새 정책을 따릅니다.

5. Spring Boot 에이전트 측 클라이언트

이제 실제 서비스(Spring Boot)가 어떻게 이 허브를 쓰는지 봅시다.

5-1. MCP 클라이언트 설정

// build.gradle.kts
dependencies {
    implementation("io.modelcontextprotocol:mcp-client:0.5.0")
    implementation("com.anthropic:anthropic-java:0.7.0")
}

// McpConfig.java
@Configuration
public class McpConfig {
    @Bean
    public McpClient guardrailClient() {
        return McpClient.builder()
            .transport(SseTransport.of("http://guardrail.internal:8765/sse"))
            .connect()
            .orElseThrow();
    }
}

5-2. 호출 래퍼

// SafeAgentService.java
@Service
@RequiredArgsConstructor
public class SafeAgentService {

    private final McpClient guardrail;
    private final AnthropicClient anthropic;

    public AgentResult handle(String agentId,
                               String userInput,
                               String sessionId) {
        // 1. 입구 가드레일
        var in = guardrail.callTool("check_input", Map.of(
            "text", userInput,
            "agent_id", agentId,
            "session_id", sessionId
        ));
        if (!in.bool("allowed")) {
            return AgentResult.blocked(in.str("reason"));
        }
        String masked = in.str("masked_text");

        // 2. LLM 호출
        var resp = anthropic.messages().create(req -> req
            .model("claude-opus-4-7")
            .maxTokens(2000)
            .addUserMessage(masked));
        String draft = resp.content().get(0).text();

        // 3. 출구 가드레일
        var out = guardrail.callTool("check_output", Map.of(
            "text", draft,
            "agent_id", agentId,
            "trace_id", resp.id()
        ));

        return AgentResult.ok(
            out.str("sanitized_text"),
            out.bool("had_pii")
        );
    }
}

5-3. 도구 호출 게이트

// ToolGate.java
@Component
@RequiredArgsConstructor
public class ToolGate {
    private final McpClient guardrail;

    public <T> T execute(String agentId, String toolName,
                         Map<String, Object> args,
                         Function<Map<String, Object>, T> actual) {
        var r = guardrail.callTool("check_tool_call", Map.of(
            "tool", toolName,
            "args", args,
            "agent_id", agentId
        ));
        if (!r.bool("allowed")) {
            throw new PolicyException(
                "Tool blocked: " + r.str("reason"));
        }
        return actual.apply(args);
    }
}

서비스 코드에서는 ToolGate로 항상 감싸기가 규약이 됩니다. 규약 위반은 ArchUnit 같은 아키텍처 테스트로 CI에서 잡습니다.

6. 정책 버저닝과 롤백

가드레일 정책도 코드입니다. Git으로 관리하고 릴리즈 파이프라인을 둡니다.

6-1. 버전 스키마

# policies/customer-facing.yaml
version: "2026.04.22-1"
previous_version: "2026.04.20-3"
owner: security-team
changes:
  - "add: 한국어 탈옥 패턴 'DAN 한글 버전'"
  - "tighten: 카드번호 Luhn 필수"

6-2. 카나리 배포

새 정책은 전체 에이전트의 5%에만 먼저 적용합니다.

# policies.py (추가)
import hashlib

CANARY_PCT = 5  # %
CANARY_FILE = "canary.yaml"

def _resolve_file(agent_id: str) -> str:
    base = _base_file(agent_id)
    # 세션 단위로 분배 (에이전트 ID 해시 기준)
    bucket = int(hashlib.md5(agent_id.encode())
                 .hexdigest()[:4], 16) % 100
    if bucket < CANARY_PCT and os.path.exists(
        os.path.join(_POLICY_DIR, CANARY_FILE)
    ):
        return CANARY_FILE
    return base

6-3. 자동 롤백 규칙

카나리 구간 알람 임계값을 넘으면 이전 버전으로 자동 복귀합니다.

# canary_watch.py
import sqlite3, os, subprocess, requests

ROLLBACK_IF = {
    "injection_blocks_per_min": 30,  # 오탐 의심
    "tool_blocks_per_min": 20,
    "error_rate_pct": 3.0,
}

def check_and_rollback():
    # Prometheus에서 최근 5분 지표 수집
    resp = requests.get(
        "http://prometheus:9090/api/v1/query",
        params={"query": "rate(ai_injection_blocks_total[5m])*60"}
    ).json()
    rate = float(resp["data"]["result"][0]["value"][1])
    if rate > ROLLBACK_IF["injection_blocks_per_min"]:
        subprocess.run(["cp",
            "policies/archive/last_known_good.yaml",
            "policies/canary.yaml"])
        alert("🛑 canary 자동 롤백")

def alert(msg):
    requests.post(os.environ["SLACK_WEBHOOK"],
                  json={"text": msg}, timeout=5)

7. 감사 로그와 멀티테넌시

여러 팀/서비스가 한 허브를 쓰면 테넌트 분리가 필수입니다.

7-1. 감사 스키마

-- audit.sql
CREATE TABLE audit_log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    tenant TEXT NOT NULL,       -- 팀/서비스
    agent_id TEXT NOT NULL,
    operation TEXT NOT NULL,    -- check_input | check_output | ...
    result_json TEXT NOT NULL,
    session_id TEXT,
    trace_id TEXT,
    policy_version TEXT
);
CREATE INDEX idx_audit_tenant_ts ON audit_log(tenant, ts);
CREATE INDEX idx_audit_trace ON audit_log(trace_id);

7-2. 테넌트 식별

# server.py (middleware)
from fastapi import Header, HTTPException

async def resolve_tenant(
    x_tenant: str = Header(...),
    x_api_key: str = Header(...),
) -> str:
    tenant = TENANTS.get(x_api_key)
    if not tenant or tenant != x_tenant:
        raise HTTPException(403, "invalid tenant")
    return tenant

7-3. 테넌트별 레이트 리밋

# rate_limit.py
from collections import defaultdict
from time import time

_buckets = defaultdict(list)
LIMIT = {"check_input": 200, "judge": 20}  # req/min

def allow(tenant: str, op: str) -> bool:
    now = time()
    bucket = _buckets[(tenant, op)]
    # 1분 이내만 유지
    while bucket and bucket[0] < now - 60:
        bucket.pop(0)
    if len(bucket) >= LIMIT.get(op, 500):
        return False
    bucket.append(now)
    return True

허브 자체가 과부하 먹고 쓰러지면 조직 전체 AI 기능이 셧다운되므로 레이트 리밋은 선택이 아니라 필수입니다.

8. 관측과 대시보드

Prometheus+Grafana 기반 대시보드에 아래 패널을 추가합니다.

패널	쿼리
테넌트별 호출량	sum(rate(guardrail_calls_total[5m])) by (tenant, op)
차단율(tenant별)	sum(rate(guardrail_blocks_total[1h])) by (tenant)
정책 버전 분포	count by (policy_version) (guardrail_policy_active)
허브 p95 지연	histogram_quantile(0.95, rate(guardrail_latency_bucket[5m]))
판사 평균 점수(rubric별)	avg(guardrail_judge_overall) by (rubric_id)

SLA 권장값

• 허브 p95 지연: 50ms 이하 (판사 호출 제외)
• 가용성: 99.9%
• 정책 반영 시간: 5분 이내
• 카나리 롤백 시간: 2분 이내

9. 고가용 배포

허브가 죽으면 모든 에이전트가 멈춥니다. HA가 아니면 의미 없습니다.

9-1. Kubernetes 권장 구성

# k8s/guardrail-hub.yaml (일부)
apiVersion: apps/v1
kind: Deployment
spec:
  replicas: 3
  strategy:
    rollingUpdate: {maxSurge: 1, maxUnavailable: 0}
  template:
    spec:
      containers:
      - name: hub
        image: registry.internal/guardrail-hub:2026.04.22-1
        readinessProbe:
          httpGet: {path: /health, port: 8765}
          periodSeconds: 5
        livenessProbe:
          httpGet: {path: /health, port: 8765}
          periodSeconds: 10
        resources:
          requests: {cpu: "500m", memory: "512Mi"}
          limits:   {cpu: "2",    memory: "2Gi"}
---
apiVersion: v1
kind: Service
spec:
  type: ClusterIP
  selector: {app: guardrail-hub}

Kubernetes 입문에서 다룬 readiness/liveness probe 설정이 그대로 쓰입니다.

9-2. 클라이언트측 페일세이프

허브가 일시 불가능할 때 안전한 기본 동작을 정해둬야 합니다.

// 클라이언트 기본값 예시
public enum FailureMode { FAIL_CLOSED, FAIL_OPEN }

public AgentResult handle(...) {
    try {
        var in = guardrail.callTool("check_input", ...);
    } catch (McpUnavailable e) {
        if (mode == FAIL_CLOSED) {
            return AgentResult.blocked("guardrail offline");
        } else {
            log.warn("guardrail offline, falling back");
            // 최소한의 로컬 PII 마스킹만 수행
            return localFallback(userInput);
        }
    }
}

• 고객 대면 = FAIL_CLOSED: 안전 우선, 장애 시 서비스 차단
• 내부 개발자 = FAIL_OPEN: 생산성 우선, 로컬 최소 방어로 계속

10. 정책 작성 표준

정책 YAML이 지저분해지면 허브 가치가 반감됩니다. 팀 공통 규약을 정합니다.

• 모든 정책은 version, previous_version, owner, changes 필수
• PR 리뷰에 최소 1명의 보안팀 승인
• 변경 전후 레드팀 세트 통과율 비교 리포트 자동 코멘트
• 정책 파일은 policies/ 디렉토리에만, 코드에 상수화 금지
• 긴급 추가는 emergency/로 분리, 72시간 내 정규 리뷰

11. 허브를 안 써야 하는 경우

솔직하게, 중앙 허브가 항상 정답은 아닙니다.

• 에이전트가 2~3개: 분산형이 단순하고 충분. 허브는 오버엔지니어링.
• 초저지연 요구(p50 10ms 미만): 네트워크 홉 추가가 부담. 로컬 SDK로 배포하고 정책만 중앙 업데이트.
• 완전 오프라인 에지: 허브 접근 자체가 불가. SDK + 주기적 정책 다운로드 방식이 현실.
• 단일 팀 소유: 공유 이익이 없음. Git 서브모듈로 정책만 공유하면 충분.

12. 마이그레이션 전략

분산형에서 허브로 한 번에 갈 수 없습니다. 단계적으로.

1. Phase 0 - 허브를 섀도우 모드로 배포. 클라이언트가 병렬 호출해 결과만 비교·기록.
2. Phase 1 - 비중요 에이전트 1~2개부터 advisory로 전환. 알림만 받고 차단은 로컬.
3. Phase 2 - 고객 대면 에이전트를 inline 차단으로. 페일 세이프는 FAIL_CLOSED.
4. Phase 3 - 로컬 가드레일 코드 제거. 허브가 단일 소스.

각 단계별 롤백 스위치(환경변수 GUARDRAIL_MODE=local|advisory|inline)를 두면 문제 발생 시 즉시 되돌릴 수 있습니다.

13. 알려진 함정

• 허브가 곧 SPOF: HA 구성 없이 운영하면 반드시 당합니다. 최소 replica=3 권장.
• 정책 복잡성 폭발: 테넌트가 많아지면 정책 YAML이 수백 줄로 불어납니다. 공통부 상속 구조 설계 필수.
• 판사 도구의 재귀 호출: 가드레일 자체도 LLM을 부르는데, 허브 안에서 다시 자기 허브를 부르지 않도록 internal_bypass 플래그 필요.
• 감사 로그 보존: PII 자체가 감사 로그에 쌓이지 않도록 저장 시점에도 마스킹 유지.
• 네트워크 비용: 한 요청에 check_input + check_output + check_tool_call 3회면 엣지 비용이 올라갑니다. 가능하면 배치 엔드포인트 제공.

14. 1년 로드맵

분기	목표
Q1	섀도우 모드 배포, 3개 에이전트 advisory 전환
Q2	고객 대면 inline, 판사 도구 런칭, SLA 확립
Q3	정책 Canary + 자동 롤백, 감사 로그 파이프라인
Q4	멀티리전 HA, 외부 파트너 공유, 정책 마켓플레이스

마치며

사내 MCP 가드레일 허브 구축의 핵심 포인트를 정리합니다.

• 가드레일 중복은 조직이 커지면 반드시 터집니다. 에이전트 20개에 PII 코드 복붙하는 순간부터 보안 회귀가 매주 발생합니다. 중앙 허브 한 군데에서 정책을 관리해야 위협 인텔이 10분 안에 전체로 반영됩니다.
• MCP가 HTTP 대비 실질적인 이점을 줍니다. 에이전트 런타임이 이미 클라이언트, 도구 메타데이터가 표준, 스트리밍 지원 - 새로 프로토콜 설계하지 않아도 됩니다. 특히 Claude Code·Cursor 같은 IDE 에이전트까지 한 번에 붙습니다.
• 직렬 프록시 vs. 사이드카는 하이브리드가 답. 고객 대면은 inline 강제, 내부 개발자는 advisory 권고. 생산성과 안전의 균형은 팀 특성별 정책으로 맞춥니다.
• 정책도 코드이므로 Git·버저닝·카나리·롤백이 필수. 새 정책은 5% 트래픽에 먼저 적용하고, 오탐 급증 시 자동 롤백. 정책 배포가 애플리케이션 배포만큼 진지하게 다뤄져야 합니다.
• 허브는 곧 SPOF이므로 HA·페일세이프가 선택이 아닙니다. replica 최소 3, 클라이언트 측 FAIL_CLOSED/FAIL_OPEN 모드 명시, p95 50ms SLA. 허브가 죽으면 조직 전체 AI가 멈춘다는 걸 구조적으로 막아야 합니다.
• 분산→허브 이행은 단계적으로. 섀도우 → advisory → inline 3단계 마이그레이션이 안전합니다. 각 단계마다 롤백 스위치를 두고, Phase 0에서 결과 비교로 신뢰를 쌓은 뒤에 Phase 1로 넘어가세요.

127·128·129에서 쌓은 비용·품질·안전 3축을 130에서 조직 단위로 확장한 구조입니다. 이제 한 팀이 아니라 여러 팀·여러 서비스가 하나의 방어선 아래에서 AI를 운영할 수 있습니다. 다음 포스트에서는 이 허브를 공급업체 간 모델 라우팅(Claude·GPT·Gemini 동적 선택)에 어떻게 연동하는지를 다뤄볼 예정입니다. 가드레일 허브가 "라우터" 역할까지 겸하면 조직의 AI 운영 비용이 다시 한 단계 크게 떨어집니다.