安全性、可靠性与可观测性

imgimgimgimgimg
首先完成“将 MCP 连接到 AI Agent”即可解锁本课程。转到上一课

MCP 集成中的安全边界

MCP 集成会将由 AI 控制的决策流程连接到服务器提供的工具、资源和提示。这种连接构成一道安全边界，而不仅仅是 API 层面的便利。必须将模型生成的工具参数视为不可信输入，因为模型可能误解请求、遵循恶意内容，或选择会造成意外后果的工具。MCP 服务器返回的数据同样不可信：其中可能包含敏感信息、格式错误的结构，或旨在影响模型后续行为的指令。

实用的威胁模型应区分用户、宿主应用、模型、MCP 客户端、MCP 服务器和下游系统。身份验证用于确立身份，而授权用于决定该身份可以执行哪些操作。模型绝不应继承应用进程所拥有的全部权限。例如，客服助手可以读取客户记录，但不应获得不受限制的数据库访问权限，也不应在没有额外策略检查的情况下获得退款权限。

应针对单项操作分配信任，而不是信任整个服务器。服务器可能同时提供无害的商品搜索工具和具有破坏性的账户删除工具。只注册应用所需的功能，根据影响对其分类，并确保在当前工作流程确有需要之前，高影响工具不可用。当操作会影响资金、访问权限、法律状态、生产数据或对外通信时，适合要求人工审批。

信任边界示意图，展示用户、AI 宿主、MCP 客户端、MCP 服务器和下游服务，并标出工具执行前的验证与授权关卡，以及返回路径上的输出过滤

验证与最小权限

即使 MCP 工具定义已经描述了输入架构，也要在执行前立即验证工具参数。描述有助于模型构造请求，但不能强制执行应用策略。验证应拒绝缺失字段、意外字段、类型错误、长度超限、无效标识符以及超出允许范围的值。业务约束与结构约束同样重要：退款金额在语法上有效，也仍可能超过当前用户的审批限额。

下面这个可运行的 Python 示例为一个假设的 MCP 工具适配器实现了一个小型安全边界。它仅使用标准库。适配器函数代表真实 MCP 客户端通过其 SDK 或传输机制发起的调用。网关会验证参数、检查调用方权限，并要求对状态变更操作进行审批。

python
from dataclasses import dataclass
from typing import Any, Callable


@dataclass(frozen=True)
class Principal:
    user_id: str
    permissions: frozenset[str]


TOOL_POLICY = {
    "customer.read": {"permission": "customer:read", "mutating": False},
    "refund.create": {"permission": "refund:create", "mutating": True},
}


def validate_arguments(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    if tool == "customer.read":
        if set(arguments) != {"customer_id"}:
            raise ValueError("customer.read requires only customer_id")
        customer_id = arguments["customer_id"]
        if not isinstance(customer_id, str) or not customer_id.startswith("cus_"):
            raise ValueError("customer_id must start with cus_")
        return {"customer_id": customer_id}


    if tool == "refund.create":
        if set(arguments) != {"payment_id", "amount_cents"}:
            raise ValueError("refund.create requires payment_id and amount_cents")
        payment_id = arguments["payment_id"]
        amount = arguments["amount_cents"]
        if not isinstance(payment_id, str) or not payment_id.startswith("pay_"):
            raise ValueError("payment_id must start with pay_")
        if isinstance(amount, bool) or not isinstance(amount, int):
            raise ValueError("amount_cents must be an integer")
        if amount < 1 or amount > 10_000:
            raise ValueError("amount_cents is outside the allowed range")
        return {"payment_id": payment_id, "amount_cents": amount}


    raise ValueError("tool is not registered")


def secure_call(
    principal: Principal,
    tool: str,
    arguments: dict[str, Any],
    approved: bool,
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
) -> dict[str, Any]:
    policy = TOOL_POLICY.get(tool)
    if policy is None:
        raise PermissionError("tool is not allowed")
    if policy["permission"] not in principal.permissions:
        raise PermissionError("principal lacks the required permission")
    if policy["mutating"] and not approved:
        raise PermissionError("explicit approval is required")


    validated = validate_arguments(tool, arguments)
    result = adapter(tool, validated)
    if not isinstance(result, dict):
        raise ValueError("tool returned an invalid result")
    return result


def demo_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"tool": tool, "accepted": True, "arguments": arguments}


if __name__ == "__main__":
    principal = Principal(
        user_id="user_42",
        permissions=frozenset({"customer:read", "refund:create"}),
    )
    output = secure_call(
        principal,
        "refund.create",
        {"payment_id": "pay_123", "amount_cents": 2500},
        approved=True,
        adapter=demo_adapter,
    )
    print(output)

输入包含主体、工具名称、参数对象和审批决定。输出是适配器返回的经过验证的字典。将 approved 改为 false 会使退款请求在执行前失败。传入额外参数、格式错误的标识符或超过策略限额的金额，也会导致请求失败。在生产环境中，权限应来自可信的身份系统；审批则必须由服务器验证的凭证表示，而不能使用模型生成的布尔值。

可靠的工具执行

每次 MCP 操作都需要设置超时。没有超时机制时，不可用的服务器或下游依赖可能会无限期占用工作线程，从而阻塞代理工作流。当客户端技术栈支持时，应分别设置连接超时和操作超时，并为用户请求设置总体截止时间。取消操作应从应用程序传播到 MCP 客户端，以确保已放弃的任务不会继续消耗资源。

重试需要了解操作的副作用。读取资源或执行确定性搜索通常可以在发生瞬时传输故障后重试。创建付款、退款、工单或消息时，不能盲目重试，因为即使响应丢失，第一次尝试也可能已经成功。对于会改变状态的工具，如果下游系统支持，应使用稳定的幂等键。否则，应在再次尝试前核对操作状态。

应将故障分类为验证错误、授权错误、超时错误、传输错误、服务器错误、速率限制错误和应用程序错误。可以向用户返回诸如“无法完成该操作”之类的安全消息，而内部遥测则记录可采取行动的错误类别。不要向模型暴露堆栈跟踪、服务器地址、访问令牌或原始下游响应。断路器可以暂时停止对持续失败服务器的调用，但它应作为超时和监控的补充，而不能替代它们。

结构化日志记录与链路追踪

可观测性应回答以下问题：尝试执行了哪项操作、由谁发起、策略是否允许、耗时多久，以及失败原因是什么。应使用结构化事件，而不是自由格式的消息。实用字段包括时间戳、链路追踪标识符、请求标识符、主体标识符、服务器名称、工具名称、结果、错误类别、尝试次数和持续时间。当这些组件支持链路追踪时，应在应用程序、MCP 客户端、服务器和下游服务之间传播相同的链路上下文。

默认情况下，不要记录完整的工具参数和结果。其中可能包含客户记录、凭据、文档内容或模型 prompt。应优先使用安全元数据的允许列表，例如记录类型、结果数量或负载大小。在日志边界处对机密信息进行脱敏，并通过访问控制和保留期限保护日志。对标识符进行哈希处理有助于关联分析，但可预测的标识符仍可能被恢复；因此，在需要进行假名化时，使用带密钥的哈希更安全。

指标应包括请求量、成功率、被拒调用次数、验证失败次数、延迟百分位数、超时次数、重试次数和服务器可用性。告警应反映用户影响，而不是孤立的噪声。例如，当关键工具的失败率在一段有意义的时间窗口内持续升高时触发告警。敏感操作的审计记录应具有持久性，并应标识策略决策和审批依据，但不应存储模型的隐藏推理过程。

可靠且可观测的网关

下一个示例在之前的边界层基础上增加了超时、保守的重试、结构化日志记录和脱敏处理。它只会重试标记为非变更操作的工具。Python 线程无法强制停止底层任务，因此适配器本身也必须配置传输层截止时间。网关超时仍可防止调用方工作流无限期等待，并展示控制机制应放置的位置。

python
import concurrent.futures
import json
import time
import uuid
from typing import Any, Callable


SAFE_TO_RETRY = {"customer.read"}


def emit_event(event: dict[str, Any]) -> None:
    print(json.dumps(event, separators=(",", ":"), sort_keys=True))


def call_with_controls(
    tool: str,
    arguments: dict[str, Any],
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
    timeout_seconds: float = 1.0,
    max_attempts: int = 2,
) -> dict[str, Any]:
    request_id = str(uuid.uuid4())
    attempts = max_attempts if tool in SAFE_TO_RETRY else 1


    for attempt in range(1, attempts + 1):
        started = time.monotonic()
        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(adapter, tool, arguments)
                result = future.result(timeout=timeout_seconds)


            duration_ms = round((time.monotonic() - started) * 1000, 2)
            emit_event({
                "event": "mcp_tool_call",
                "request_id": request_id,
                "tool": tool,
                "attempt": attempt,
                "outcome": "success",
                "duration_ms": duration_ms,
                "argument_keys": sorted(arguments.keys()),
            })
            return result
        except concurrent.futures.TimeoutError:
            category = "timeout"
        except ConnectionError:
            category = "transport"
        except Exception:
            category = "internal"


        duration_ms = round((time.monotonic() - started) * 1000, 2)
        emit_event({
            "event": "mcp_tool_call",
            "request_id": request_id,
            "tool": tool,
            "attempt": attempt,
            "outcome": "failure",
            "error_category": category,
            "duration_ms": duration_ms,
            "argument_keys": sorted(arguments.keys()),
        })


        if attempt == attempts:
            raise RuntimeError(f"tool call failed: {category}")
        time.sleep(0.1 * attempt)


    raise RuntimeError("unreachable")


def stable_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"customer_id": arguments["customer_id"], "status": "active"}


if __name__ == "__main__":
    result = call_with_controls(
        "customer.read",
        {"customer_id": "cus_123"},
        stable_adapter,
    )
    print(json.dumps(result))

运行程序会先产生一条结构化日志事件，然后输出适配器结果。日志会记录参数名称但不记录参数值，从而减少意外暴露数据的风险。生产环境实现应将打印操作替换为组织的遥测导出器，附加现有的链路追踪标识符，并在内部保留原始的类型化异常。面向模型的层应接收一个稳定且经过清理的错误对象，帮助模型判断是向用户提问、选择其他工具，还是停止操作。

测试与运营准备

安全测试应尝试调用未知工具、测试缺少权限、格式错误的参数、超长字符串、意外字段、伪造的批准值，以及服务器返回的恶意内容。可靠性测试应模拟响应缓慢、连接中断、速率限制、部分失败，以及修改类调用结果不明确的情况。验证重试逻辑绝不会重复执行副作用，并确保超时处理不会将可能已经完成的事务错误报告为确定失败。

可观测性测试同样重要。确认每个请求都具有相关标识符，被拒绝的调用会生成审计事件，延迟测量使用单调时钟，并确保敏感值绝不会出现在日志中。主动制造可控故障，并验证仪表板和告警能够显示这些故障。未经测试的告警，只是对监控链路的一种假设。

因此，一个可用于生产环境的 MCP 集成会在模型意图与实际能力之间设置受控网关。该网关会验证输入和输出，应用最小权限策略，为具有重大影响的操作获取批准，设置截止时间，仅重试安全操作，清理并安全呈现失败信息，同时发出具有关联关系的遥测数据。这些控制措施无需依赖模型来执行安全规则，就能提高系统安全性；当真实应用的行为不同于开发演示时，它们还能够提供诊断故障所需的证据。

课后练习

课程检查点

巩固所学知识，获取作业反馈。

0/1
已回答 · 0/1