简介

将 MCP 连接到 AI Agent，意味着为 Agent 提供一种受控方式，使其能够发现并使用模型之外的能力。模型不会直接执行数据库查询、读取私有文件或调用业务 API。相反，Agent 内部的 MCP 客户端会连接到 MCP 服务器，获取服务器的工具定义，将这些定义作为可用能力发送给模型，并执行模型选定的工具调用。

这种分离形成了一个重要边界。模型负责理解用户请求，并提出包含结构化参数的工具名称。MCP 客户端负责维护协议会话、调用选定的工具、返回结果以及报告错误。因此，一个可靠的工作流包含三个明确阶段：发现、选择和调用。无论模型由商业 API、托管在本地的模型，还是内部 Agent framework 提供，这一模式都同样适用。

MCP 并不会消除应用层安全的必要性。Agent 只能连接到受信任的服务器，并应在需要时应用身份验证、验证用户权限，避免自动批准危险操作。工具描述有助于模型选择能力，但它不是安全策略。

Agent 工作流

启动时，MCP 客户端会打开到服务器的传输连接，并初始化 MCP 会话。随后，客户端调用协议提供的工具发现操作。响应包含工具名称、描述和输入 schema。这些 schema 描述工具接受的参数，使 Agent 能够向模型提供结构化信息，而不是一段非正式的说明。

模型会结合发现到的定义和用户请求进行处理。例如，如果服务器公开了 search_documents 工具和 create_ticket 工具，那么对于查找事件报告的请求，通常应生成对 search_documents 的调用，而不是对 create_ticket 的调用。模型的输出应被视为一个包含名称和参数的待定工具调用。主应用仍必须验证该名称确实已被发现，并确认参数符合本地策略，然后才能调用该工具。

验证通过后，MCP 客户端调用选定的工具，并接收结构化内容、文本或错误。主应用可以将该结果返回给模型，以生成最终的自然语言响应。当第一次工具调用的结果提供了第二个工具所需的信息时，多步骤 Agent 会重复这一循环。应用应限制步骤数量、执行时间和敏感操作，以防止出现失控循环。

一个最小化的 MCP 客户端

下面的 Python 程序演示了如何通过 Streamable HTTP 进行发现、选择和调用。它使用官方 MCP Python 客户端接口来创建客户端会话和传输连接。请在运行 agent 的环境中安装 MCP Python 包，并将 URL 替换为一个已经运行且配置为接受此传输方式的 MCP 服务器。

import argparse
import asyncio
import json

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client


async def run(url: str, tool_name: str, arguments: dict) -> None:
    async with streamable_http_client(url) as streams:
        read_stream, write_stream, _ = streams

        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            discovered = await session.list_tools()

            print("Discovered tools:")
            for tool in discovered.tools:
                description = tool.description or "No description"
                print(f"- {tool.name}: {description}")

            selected = next(
                (tool for tool in discovered.tools if tool.name == tool_name),
                None,
            )
            if selected is None:
                available = ", ".join(tool.name for tool in discovered.tools)
                raise ValueError(
                    f"Unknown tool '{tool_name}'. Available tools: {available}"
                )

            print(f"Invoking {selected.name} with {json.dumps(arguments)}")
            result = await session.call_tool(
                selected.name,
                arguments=arguments,
            )
            print("Tool result:")
            if hasattr(result, "model_dump_json"):
                print(result.model_dump_json(indent=2))
            else:
                print(result)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("url", help="MCP Streamable HTTP endpoint")
    parser.add_argument("tool", help="Discovered tool name to invoke")
    parser.add_argument(
        "--arguments",
        default="{}",
        help="JSON object containing tool arguments",
    )
    args = parser.parse_args()

    try:
        arguments = json.loads(args.arguments)
    except json.JSONDecodeError as error:
        raise SystemExit(f"Arguments must be valid JSON: {error}") from error

    if not isinstance(arguments, dict):
        raise SystemExit("Arguments must be a JSON object")

    try:
        asyncio.run(run(args.url, args.tool, arguments))
    except Exception as error:
        raise SystemExit(f"MCP request failed: {error}") from error


if __name__ == "__main__":
    main()

输入包括一个 MCP 端点、工具名称和一个 JSON 参数对象。程序首先打印通过发现流程返回的工具列表，然后检查请求的名称是否确实存在。这项检查非常重要，因为模型可能会虚构工具名称，或保留早期会话中的过时名称。输出包含服务器返回的结果，外围 agent 可以将其传回模型进行解读。

将发现流程连接到模型选择

在生产环境中的 agent 会在发现和调用之间加入模型适配器。`list_tools` 返回后，主机将每个工具的名称、描述和输入 schema 转换为模型提供商所使用的工具调用格式。具体的请求格式取决于模型 API，因此 MCP 客户端不应假设每个提供商都使用相同的消息结构。agent 接收模型响应，提取建议的名称和 JSON 参数，然后在调用 `session.call_tool` 之前执行本地验证。

当工具名称不在当前发现响应中、参数不是对象、缺少必填字段，或某个值违反应用策略时，主机应拒绝该工具调用。Schema 验证很有用，但授权检查必须单独执行。例如，格式有效的 `delete_record` 调用，对于只读用户仍可能被禁止。对于不可逆操作、金融操作、外部消息或生产数据变更，也应要求确认。

调用完成后，可以将结果作为工具结果消息追加到模型对话中。然后，模型可以总结结果，或决定是否需要另一个获准使用的工具。保留原始结果以便记录日志，但在将内容发送给第三方模型之前，应对凭据、个人数据和内部诊断信息进行脱敏。

常见故障与可靠实践

连接失败通常意味着端点错误、服务器不可用、传输方式不匹配或缺少身份验证。请确认服务器支持所选传输方式，并确认网络、代理和 TLS 设置允许建立连接。不要通过静默回退到未知协议来解决传输问题；应明确配置所使用的传输方式，并报告原始错误。

当 agent 调用了当前发现步骤未返回的名称时，就会出现未知工具错误。当会话或服务器能力可能发生变化时，应刷新发现结果，并始终将请求的名称与当前工具集进行比较。无效参数通常是由缺少必需属性、类型错误，或模型返回了编码为字符串而非对象的 JSON 导致的。在调用前验证并规范化参数，然后保留服务器提供的有用错误消息，以便调试。

另一个常见错误是向每个用户暴露所有已连接服务器的工具。工具名称和描述可能相互重叠，而恶意服务器或配置不当的服务器可能会暴露不安全的能力。请使用明确的服务器允许列表，隔离凭据，实施逐工具授权，设置超时，限制重复调用次数，并记录涉及的用户、模型决策、服务器和工具。这些控制措施可以让 agent 的行为可观察，也更便于审查。

总结

要正确理解支持 MCP 的 agent，最合适的方式是将其视为一个包含两个协同组件的宿主应用。模型负责推理用户意图并提出结构化调用，而 MCP 客户端负责发现能力并执行符合协议的操作。客户端绝不能仅因为模型提到了某个工具，就假定该工具存在；它应发现当前可用的工具，验证工具选择和参数，并在调用前执行本地策略检查。

实际工作流程是：初始化、列出工具、将工具 schema 提供给模型、验证模型提出的调用、调用选定的工具，并返回结果供模型进行最终推理。针对一个服务器完成这一循环后，无需改变核心边界，即可将其扩展到多个服务器和多步骤任务。最重要的实现决策仍包括明确的传输配置、严格的验证、安全的授权、受限的执行范围以及有用的错误报告。