MCP协议实战：AI工程师的模型可控性架构指南

1. 这不是一本“手册”，而是一份AI工程师的MCP架构实战备忘录

“AI Engineer’s Handbook to MCP Architecture”——看到这个标题，我第一反应不是去翻 glossary 或查 RFC 文档，而是立刻打开终端，cd 到一个正在跑 LLM 微调 pipeline 的项目根目录，把requirements.txt里那行llama-cpp-python==0.2.87注释掉，换成了llama-cpp-python==0.3.1。为什么？因为就在上周，团队用旧版本在 A100 上跑 MCP-based agent loop 时，context window 切片逻辑在 token 边界处漏掉了 3 个 control token，导致整个 tool-calling chain 在第 7 轮就静默失败，日志里只留下一行{"status": "aborted", "reason": "invalid state transition"}。没人报错，没人抛异常，但业务流水线卡住了 47 分钟。这就是 MCP（Model Control Protocol）的真实切口：它不声不响地藏在你最信任的推理框架底层，一旦出问题，不是报错，而是“不工作”。

MCP 不是新模型，不是新训练范式，更不是又一个大模型评测榜单。它是 AI 工程师在真实生产环境中，为解决“如何让大模型稳定、可预测、可审计地调用外部能力”这一核心矛盾，自发演化出来的一套轻量级通信契约与状态管理协议。关键词是Control——不是“控制模型”，而是“让模型行为可控”。它直指当前 AI 应用落地中最痛的三根刺：tool calling 结果不可信、multi-step reasoning 链路不可追溯、agent 状态在异步服务间漂移丢失。你不需要从头造轮子，但必须理解 MCP 如何在你的 LangChain chain、LlamaIndex query engine、甚至自研的 streaming inference server 里悄悄起作用。这篇内容，就是写给那些已经能跑通 RAG、写过 function calling、部署过 vLLM 的工程师看的——它不教你怎么调参，而是告诉你，当你的 agent 在凌晨三点返回一个格式正确但语义荒谬的 JSON 时，该去哪一行代码里加断点。

它适合三类人：一是正在把 PoC 推向生产环境的算法工程师，你需要知道 MCP 如何影响 latency budget 和 error budget；二是负责搭建 AI 中台的后端架构师，你得判断是否要在 inference gateway 层显式支持 MCP handshake；三是独立开发者或小团队技术负责人，你没有资源堆 SRE，但必须让每个 agent call 都有 trace_id、state_hash 和 rollback_point。这不是理论推演，所有结论都来自我们过去 18 个月在金融风控、医疗问诊、工业设备诊断三个垂直场景中，累计 23 个 MCP-compliant agent 服务的实操沉淀。下面拆解的每一个参数、每一行配置、每一个 debug 技巧，都对应着至少一次线上事故的 root cause 分析报告。

2. MCP 架构的本质：一场关于“状态主权”的重新分配

2.1 为什么传统方案在复杂 Agent 场景下必然失效？

先说清楚 MCP 要解决什么。假设你正在构建一个“智能工单处理 agent”：用户输入“服务器 CPU 持续 95% 三天了，怎么查？”，agent 需要依次执行：① 调用 Prometheus API 获取指标；② 解析时间序列，识别异常模式；③ 查询 CMDB 获取该服务器所属业务线；④ 调用内部知识库检索同类故障案例；⑤ 综合生成处置建议。这看似标准的 chain-of-thought，但在生产环境会遭遇三重坍塌：

• 第一重坍塌：Tool Calling 的“黑盒延迟”
  传统 function calling（如 OpenAI 的tools参数）将工具描述、参数 schema、调用结果全部塞进 model context。当 Prometheus 查询返回 2000 行原始数据时，模型必须在有限 context 内完成解析、过滤、摘要——这直接导致 token 浪费、推理变慢，更致命的是，模型可能因上下文拥挤而忽略关键约束（例如“只返回最近 2 小时数据”）。MCP 的解法是：将工具执行完全剥离出模型推理循环。模型只输出结构化 action plan（JSON Schema 定义），由 MCP runtime 负责调用、超时控制、重试、结果清洗，再将精炼后的 payload（如{"anomaly_type": "spike", "duration_hours": 72}）注入下一轮 context。这相当于把“体力活”外包给专业工人，模型专注“脑力活”。

• 第二重坍塌：State Management 的“幽灵漂移”
  在长周期 multi-turn 对话中，agent 需维护对话状态（user_intent, current_step, pending_tool_results, retry_count）。传统做法是把 state 存在 session store（Redis）或拼进 system prompt。前者有网络延迟和一致性风险（并发请求可能读到 stale state），后者有 context 长度限制和 prompt injection 风险。MCP 引入State Token机制：每次模型输出 action 后，runtime 生成一个 cryptographically signed state hash（如sha256(user_id + step_id + tool_result_hash + timestamp)），并作为唯一 key 存入本地 state registry。下一轮请求必须携带此 token，runtime 校验签名有效性后才加载对应 state。这确保了 state 的“主权”始终在 runtime 手中，模型无法伪造或篡改。

• 第三重坍塌：Error Handling 的“静默雪崩”
  当 Prometheus API 返回 503（服务不可用）时，传统方案往往让模型自己决定“重试”还是“换工具”。但模型缺乏重试策略知识（指数退避？熔断阈值？），更无法感知下游服务健康度。MCP 将错误分类为Recoverable（网络超时、限流）和Terminal（schema mismatch、auth failed），并在 protocol level 定义 recovery policy：对 Recoverable 错误，runtime 自动按预设策略重试（最多 3 次，间隔 1s/2s/4s）；对 Terminal 错误，则立即终止 chain，返回 machine-readable error code（如MCP_ERR_TOOL_SCHEMA_MISMATCH）和 human-readable message。这避免了模型在错误状态下继续 hallucinate。

提示：MCP 不是替代 LLM 的协议，而是为 LLM 构建的“操作系统内核”。它不关心你用 Llama-3 还是 Qwen，只关心你能否按约定格式输出 action、校验 state token、处理 error code。就像 TCP 不关心上层是 HTTP 还是 FTP。

2.2 MCP 的核心组件与数据流：一张图看懂协议骨架

MCP 架构并非单体服务，而是由四个松耦合组件构成的协作网络。理解它们的职责边界，是避免设计陷阱的前提：

数据流遵循严格顺序：
User Request → Protocol Gateway → Model Adapter → LLM Inference → Model Adapter → Action Executor → State Registry → (Loop back to Model Adapter for next step)

注意两个关键设计：

1. 无状态 Gateway：Gateway 本身不存任何 session data，所有 state 操作通过X-MCP-State-Tokenheader 透传给下游组件。这保证了 Gateway 可无限水平扩展。
2. Adapter 的双重角色：它既是“翻译官”（text ↔ MCP action），也是“守门员”（校验 action schema 合法性，拦截非法 tool name）。我们在 adapter 层加了一道硬规则：任何 action 中tool_name字段必须存在于白名单ALLOWED_TOOLS = ["prometheus_query", "cmdb_search", "kb_retrieve"]，否则直接 400，绝不让非法请求进入 executor。

2.3 MCP vs. 其他协议：为什么不是简单套用 OpenAPI 或 gRPC？

常有人问：“既然 MCP 定义了 action schema，为什么不直接用 OpenAPI 描述工具，用 gRPC 调用？” 这是个好问题，答案藏在协议目标的差异里：

• OpenAPI 是面向人类的文档协议，它的schema用于生成 SDK 和文档，但 runtime 不强制校验。一个 OpenAPI 定义的queryendpoint 可能接受{"metric": "cpu_usage", "range": "24h"}，但实际返回{ "error": "invalid range format" }—— 这种弱契约在 agent 场景下是灾难性的。MCP 的 action schema 是runtime 强契约：{"tool_name": "prometheus_query", "parameters": {"metric": "cpu_usage", "range_seconds": 86400}}，其中range_seconds必须是 integer，且 executor 会在调用前做类型校验，非法输入直接拒绝，不发请求。

• gRPC 是面向服务的通信协议，它解决了网络传输效率，但没解决状态协同。gRPC stream 可以传多条消息，但无法保证“第 3 条消息一定在第 2 条消息的 state context 下执行”。MCP 的 state token 机制，本质是给每次 RPC 调用打上“事务快照 ID”，让 executor 知道“这次调用基于哪个世界状态”。这类似于数据库的 MVCC（多版本并发控制），但发生在 AI agent 的决策链路上。

• LangChain Tools 是面向开发者的抽象协议，它依赖 Python 对象和 runtime 环境。当你把 LangChain chain 部署为 serverless function 时，每次 cold start 都要 reload tools，state 无法跨 invocation 持久化。MCP 的 design goal 是language-agnostic和environment-agnostic：action 是 JSON，state token 是 base64 string，executor 可以是 Rust binary、Python script、甚至 Bash script（只要它能 parse JSON input 并 return JSON output）。我们在边缘设备上就用jq+curl实现了极简 MCP executor，证明了协议的普适性。

注意：MCP 不排斥现有生态。你可以用 LangChain 的BaseTool类封装 MCP executor，也可以用 FastAPI 的@app.post("/mcp/action")endpoint 暴露 executor。关键在于，所有交互必须通过 MCP 定义的字段（tool_name,parameters,state_token）进行，而非 LangChain 的run()方法或 OpenAPI 的 path parameter。

3. 核心细节解析：从协议规范到工程落地的 7 个生死关

3.1 Action Schema 的设计哲学：宁可笨重，不可模糊

MCP 的 action object 看似简单，但 schema 设计是成败关键。我们曾因一个字段命名失误，导致 3 个业务线同时出现工具调用错乱。以下是经过血泪验证的 schema 规则：

{ "action_id": "uuid4", "tool_name": "string (whitelist enforced)", "parameters": { "required": ["metric", "range_seconds"], "properties": { "metric": { "type": "string", "enum": ["cpu_usage", "memory_percent", "disk_io_wait"] }, "range_seconds": { "type": "integer", "minimum": 300, "maximum": 604800 } } }, "state_token": "base64url-encoded sha256 hash", "timeout_ms": 15000, "max_retries": 3 }

• action_id必须全局唯一且不可预测：不能用timestamp + counter（易被猜解），必须用 cryptographically secure random（如 Pythonsecrets.token_urlsafe(16)）。这是为了防止攻击者伪造 action 请求，消耗 executor 资源。我们曾被恶意扫描发现，某次测试环境未校验action_id，导致 Prometheus executor 被刷爆，QPS 从 50 涨到 2000。

• tool_name白名单必须硬编码在 executor：不能从数据库动态加载（有延迟和一致性风险），也不能允许通配符（如"tool_name": "prometheus_*"）。白名单在服务启动时加载到内存，变更需重启。理由很简单：tool_name是权限边界，动态加载等于开放任意代码执行入口。

• parameters的 enum 和 range 必须在 executor 层二次校验：即使 Model Adapter 已做过 schema check，executor 仍需用jsonschema.validate()再校验一次。因为 adapter 可能被绕过（如直接 curl executor endpoint）。我们在线上加了监控：当 executor 拒绝的请求中，adapter_validated=true占比 > 0.1%，就触发告警——这说明有人在 bypass adapter。

• timeout_ms和max_retries是保护性字段，非建议值：它们定义了 executor 的行为上限，而非模型的期望值。模型可以输出"timeout_ms": 5000，但 executor 可根据自身负载策略 override 为10000（需记录 override log）。这给了运维兜底能力。

实操心得：我们用 JSON Schema Draft-07 定义所有 tool schema，并用jsonschema库生成 Python dataclass（viadatamodel-code-generator）。这样，executor 的输入 validation 和 IDE 的 autocomplete 同时搞定，开发体验和运行安全兼得。

3.2 State Token 的生成与验证：安全与性能的钢丝绳

State token 是 MCP 的心脏，其设计直接决定系统安全性和吞吐量。我们放弃过两种方案：

• 方案A（纯 JWT）：用 HS256 签名，payload 包含user_id,session_id,step_num,exp。问题：JWT 默认不防重放，需要额外维护 jti blacklist，高并发下 Redis blacklist 成瓶颈。我们压测发现，当 QPS > 500 时，blacklist check 延迟飙升至 200ms+。

• 方案B（UUIDv4 + HMAC）：生成随机 UUIDv4 作为 token，同时用 secret key 计算HMAC-SHA256(uuid + user_id + exp_timestamp, secret_key)作为 signature。问题：token 本身无状态，每次验证都要查 DB 加载user_id和exp，同样有延迟。

最终采用方案C（Signed Timestamped Nonce）：

import time import hmac import hashlib import base64 def generate_state_token(user_id: str, session_id: str, nonce: str) -> str: # 1. 构建 payload: user_id + session_id + nonce + current_time + expiry_window now = int(time.time()) expiry = now + 300 # 5 minutes payload = f"{user_id}|{session_id}|{nonce}|{now}|{expiry}" # 2. 用 secret key 计算 HMAC signature = hmac.new( SECRET_KEY.encode(), payload.encode(), hashlib.sha256 ).digest() # 3. Base64URL encode signature + payload token_bytes = signature + b"." + payload.encode() return base64.urlsafe_b64encode(token_bytes).decode().rstrip("=") def verify_state_token(token: str) -> dict | None: try: # 1. Decode and split decoded = base64.urlsafe_b64decode(token + "=" * (4 - len(token) % 4)) sig, payload = decoded[:32], decoded[33:] # 32 bytes for SHA256 # 2. Recompute signature expected_sig = hmac.new( SECRET_KEY.encode(), payload, hashlib.sha256 ).digest() # 3. Constant-time compare if not hmac.compare_digest(sig, expected_sig): return None # 4. Parse payload and check expiry parts = payload.decode().split("|") if len(parts) != 5: return None user_id, session_id, nonce, issued_at, expiry = parts if int(time.time()) > int(expiry): return None return { "user_id": user_id, "session_id": session_id, "nonce": nonce, "issued_at": int(issued_at), "expiry": int(expiry) } except Exception: return None

为什么这个方案胜出？

• 无状态验证：signature 和 payload 都在 token 里，验证时无需查 DB，纯 CPU 计算，P99 < 0.5ms。
• 防重放：nonce是一次性随机字符串，由 Model Adapter 在每次生成 action 时生成（secrets.token_urlsafe(8)），并存入 State Registry 的nonce_used_set（Redis Set，TTL=5min）。验证时先查 nonce 是否已用，再校验 signature。
• 防篡改：HMAC 确保 payload 任何改动都会使 signature 失效。
• 可追溯：user_id和session_id明确归属，便于审计。

注意：SECRET_KEY必须轮换！我们每 30 天自动 rotate 一次，并保留上一版 key 用于验证旧 token（Grace Period）。Key 存储在 HashiCorp Vault，绝不硬编码。

3.3 Error Code 的分层设计：让故障可定位、可归因、可自动化

MCP 定义了 12 个标准 error code，分为三层，每层解决不同问题：

关键设计原则：

• Code 必须机器可 parse：所有 code 都是UPPER_SNAKE_CASE，不含空格，长度固定（MCP_ERR_+ 12 chars）。前端 JS 可用正则/^MCP_ERR_(\w+)_/提取层级和原因。
• Message 必须 human-readable：message字段永远用中文（或业务语言），包含具体失败原因和建议操作。例如MCP_ERR_EXECUTOR_TOOL_TIMEOUT的 message 是：“工具调用超时（15秒）。请检查 Prometheus 服务状态，或联系运维确认网络连通性。”
• Metadata 必须结构化：每个 error response 都带error_detailsobject，包含tool_name,action_id,timestamp,retryable: true/false。这让我们能用 ELK 做根因分析：SELECT error_code, COUNT(*) FROM mcp_errors WHERE error_details.tool_name = 'prometheus_query' GROUP BY error_code。

我们曾用这套 error code 发现一个隐藏 bug：MCP_ERR_EXECUTOR_TOOL_SCHEMA_MISMATCH在凌晨 2-4 点集中爆发。排查发现，CMDB 服务每天凌晨 3 点执行 schema migration，短暂返回新旧字段并存的 response，导致 executor 的 JSON schema validator 失败。修复方案不是改 executor，而是让 CMDB 加了 backward-compatible mode。没有分层 error code，这个跨服务的时序问题根本无法定位。

3.4 Streaming 的 MCP 适配：如何让“思考过程”真正可见

MCP 的核心是 action-driven，但用户需要看到模型“思考”的过程。我们实现了Streaming MCP，让前端实时渲染：Thinking... → Calling Prometheus → Analyzing data → Generating report。这要求打破传统 streaming 的 text-only 模式。

实现方案：

1. Model Adapter 输出 multipart SSE：

   event: mcp_action data: {"action_id":"abc123","tool_name":"prometheus_query","parameters":{"metric":"cpu_usage"}} event: mcp_thinking data: {"content":"正在查询 CPU 使用率指标..."} event: mcp_tool_result data: {"action_id":"abc123","result":{"anomaly_type":"spike","value":95.2}}

2. Protocol Gateway 负责 event routing：

   • mcp_action事件转发给 Action Executor
   • mcp_thinking事件直接透传给前端（用于 UI loading state）
   • mcp_tool_result事件触发下一轮 Model Adapter 调用

3. 前端用 EventSource 处理：

   const es = new EventSource("/mcp/stream"); es.addEventListener("mcp_thinking", e => { document.getElementById("status").textContent = JSON.parse(e.data).content; }); es.addEventListener("mcp_action", e => { // 显示工具调用动画 showToolIcon(JSON.parse(e.data).tool_name); });

关键挑战与解法：

• Challenge：Action 和 Thinking 的时序错乱
  模型可能先输出mcp_thinking，再输出mcp_action，但前端需保证“Thinking”文案在“Calling X”之前显示。
  Solution：Adapter 缓存所有mcp_thinking事件，直到收到mcp_action后，再按顺序 flush 出去。用setTimeout(..., 0)确保 microtask 顺序。

• Challenge：Tool Result 的 streaming 渲染
  Prometheus 查询结果可能很大，executor 不能等全部数据回来才发mcp_tool_result。
  Solution：executor 支持stream_result: true参数，对大 payload 分块发送mcp_tool_result_chunk事件，前端用ReadableStream拼接。

实操心得：我们禁用了所有模型的logprobs和top_logprobsstreaming，因为它们增加 30% 延迟且对 MCP 无用。把省下的 GPU memory 用来提升max_tokens，让模型有更多空间输出高质量 action。

3.5 Tool Executor 的幂等性实现：一次调用，永恒正确

幂等性是 MCP executor 的生命线。prometheus_query工具被调用两次，必须返回相同结果（或明确错误），否则 state 会混乱。我们为每个 tool 实现了三级幂等保障：

1. Request-Level Idempotency Key：
   executor 从 action 的parameters中提取业务唯一 key（如{"metric": "cpu_usage", "range_seconds": 3600}的 hash），作为 Redis key。首次调用时，用SET key result EX 300 NX（NX=only set if not exists）存结果。后续调用直接GET key。这保证了 5 分钟内相同请求必返回相同结果。

2. Response-Level Schema Normalization：
   Prometheus 原始 response 是时间序列数组，格式不稳定。executor 总是将其 normalize 为标准 schema：

   { "summary": { "min": 12.3, "max": 95.2, "avg": 48.7, "anomalies": [{"type": "spike", "start": "2024-05-20T02:15:00Z", "end": "2024-05-20T02:18:00Z"}] }, "raw_data_sample": [{"timestamp": "...", "value": 95.2}] }

   这样，即使 Prometheus API 版本升级，executor 的输出 schema 不变，Model Adapter 不用改。

3. Side-Effect-Free Execution：
   所有 tool 调用必须是 GET 请求或幂等 POST（如POST /api/querywith idempotency-key header）。我们禁用任何非幂等操作（如DELETE /api/alert/{id}）作为 MCP tool，除非包装成幂等 wrapper（如POST /mcp/wrapper/delete_alert?id=123&confirm=true）。

注意：幂等 key 的 TTL 必须大于业务最大容忍重复时间。我们设为 300 秒（5 分钟），因为最长的 multi-step agent flow 不超过 4 分钟。TTL 太短会导致频繁重复调用；太长会占用过多 Redis 内存。

3.6 MCP 的可观测性：没有 metrics 的协议就是黑盒

MCP 的复杂性要求比传统 API 更细粒度的监控。我们定义了 7 个核心 metrics，全部接入 Prometheus：

告警规则示例（Prometheus Alertmanager）：

- alert: MCP_StateTokenNonceReuse expr: rate(mcp_state_token_verification_total{result="nonce_reused"}[5m]) > 0 for: 1m labels: severity: critical annotations: summary: "MCP state token nonce reuse detected - possible replay attack" description: "Nonce reuse in {{ $labels.instance }} indicates potential security threat" - alert: MCP_ToolTimeoutHigh expr: rate(mcp_executor_retry_total{retry_count="3"}[1h]) / rate(mcp_request_total{status_code="200"}[1h]) > 0.1 for: 5m labels: severity: warning annotations: summary: "High tool timeout rate for {{ $labels.tool_name }}" description: "{{ $value | humanizePercentage }} of {{ $labels.tool_name }} calls timeout after 3 retries"

没有这些 metrics，MCP 就是黑盒。我们曾靠mcp_adapter_schema_validation_total{result="fail"}的 spike，提前 2 小时发现模型在新 prompt 下开始输出非法tool_name，避免了线上故障。

3.7 安全加固：MCP 不是银弹，而是需要层层防护的协议

MCP 协议本身不提供安全，它只是定义了交互契约。真正的安全来自协议之上的加固层：

• Network Layer：Protocol Gateway 必须启用 mTLS，所有 executor 与 gateway 之间用双向证书认证。我们用 cert-manager 自动签发证书，有效期 90 天。

• Auth Layer：Gateway 在接收请求时，必须校验Authorization: Bearer <jwt>，且 JWT 的scope必须包含mcp:execute:<tool_name>。例如，调用prometheus_query需要scope: "mcp:execute:prometheus_query"。这实现了 RBAC。

• Input Sanitization Layer：Model Adapter 在解析 action parameters 前，必须对所有 string 字段做 XSS 过滤（用bleach.clean()）和 SQL injection 检测（正则匹配' OR 1=1 --等 pattern）。我们曾发现一个 case：模型输出{"metric": "cpu_usage'; DROP TABLE alerts; --"}，幸亏有这层过滤。

• Output Escaping Layer：executor 返回的result字段，如果包含 HTML 或 JS，必须在 gateway 层 escape（html.escape()），防止前端 XSS。我们规定：所有result字段默认视为 untrusted content。

• Rate Limiting Layer：Gateway 对每个user_id+tool_name组合做滑动窗口限流（如100 req/minute）。用 Redis 的INCR+EXPIRE实现，精确到毫秒。

提示：安全不是 checklist，而是纵深防御。MCP 的state_token防篡改，gateway 的 mTLS 防窃听，adapter 的 input sanitization 防注入，executor 的 RBAC 防越权——五层防护缺一不可。

4. 实操过程：从零搭建一个 MCP-Compliant Prometheus Agent

4.1 环境准备与依赖安装：最小可行集

我们用 Python 3.11 + FastAPI 搭建 demo，所有依赖控制在 12 个以内，确保可复现：

# 创建虚拟环境 python -m venv mcp_env source mcp_env/bin/activate # Linux/Mac # mcp_env\Scripts\activate # Windows # 安装核心依赖（仅 7 个） pip install fastapi uvicorn pydantic jsonschema redis python-jose[cryptography] requests # 可选：用于本地测试的 mock 工具 pip install httpx pytest

为什么只选这 7 个？

• fastapi+uvicorn：高性能 async web framework，原生支持 streaming。
• pydantic：定义 MCP action schema 和 error response，自动 validation + serialization。
• jsonschema：executor 层二次校验，比 Pydantic 更灵活（支持$ref复用 schema）。
• redis：State Registry 的首选，INCR,SET NX,GET命令完美匹配 MCP 需求。
• python-jose：JWT-like state token 的 HMAC 实现，轻量无 OpenSSL 依赖。
• requests：executor 调用外部 API，比 httpx 更稳定（我们踩过 httpx 的 connection pool bug）。

注意：不安装langchain、llama-index等 heavy deps。MCP 是协议，不是框架。你可以用任何 LLM client，只要它能输出符合 schema 的 JSON。

4.2 定义 MCP Action Schema：用 Pydantic 写死契约

创建schemas.py：

from pydantic import BaseModel, Field, validator from typing import Dict, Any, Optional import re # 工具白名单（硬编码！） ALLOWED_TOOLS = ["prometheus_query", "cmdb_search"] class MCPAction(BaseModel): action_id: str = Field(..., min_length=16, max_length=32, regex=r'^[a-zA-Z0-9_-]+$') # URL-safe tool_name: str = Field(..., regex=r'^[a-zA-Z0-9_]+$') parameters: Dict[str, Any] = Field(...) state_token: str = Field(..., min_length=43, max_length=128) # base64url encoded timeout_ms: int = Field(15000, ge=1000, le=60000) max_retries: int = Field(3, ge=0, le=5) @validator('tool_name') def validate_tool_name(cls, v): if v not in ALLOWED_TOOLS: raise ValueError(f'tool_name must be one of {ALLOWED_TOOLS}') return v @validator('parameters') def validate_parameters(cls, v, values): tool = values.get('tool_name') if tool == "prometheus_query": if "metric" not in v or "range_seconds" not in v: raise ValueError('prometheus_query requires metric and range_seconds') if not isinstance(v["range_seconds"], int) or v["range_seconds"] < 300 or v["range_seconds"] > 604800: raise ValueError('range_seconds must be int between 300 and 604800') return v class MCPErrorResponse(BaseModel): error_code: str message: str error_details: Optional[Dict[str, Any]] = None