用于AI代理的MCP：在LangGraph中实现工具集成的标准化

简介与回顾

到第二部分结束时，该系统已达到真正的协调状态：有一组专门的智能体，每个负责处理特定的任务片段，在协调器的管理下基于共享状态进行操作，而协调器则决定下一步该执行什么任务。不过在所有示例中都假设这些智能体已经能够访问共享状态中的所需数据，随时可以读取使用。

在实际应用中，这种假设很少成立。智能体通常需要来自图结构之外的信息：数据库中的某一行数据、外部 API 的响应、从知识库中获取的段落，或是系统边界之外的其他资源。每当智能体需要访问这些外部资源时，它都需要相应的路径来实现，而如果每条路径都需手动构建，那么对于每个智能体和每种外部资源，就不得不重复进行类似但略有不同的集成工作。

这种重复性正是本文要解决的问题，这也解释了为何 MCP 在过去一年里成为频繁被讨论的话题。在决定是否采用它之前，先明确它究竟解决了什么问题，以及在没有 MCP 之前的时代将工具与智能体连接起来需要做什么，是很有帮助的。

MCP声称要解决的问题

在MCP出现之前，要让智能体访问外部资源，就需要为该特定资源手动构建定制化的集成方案，其设计方式完全取决于当时的实际情况。有的资源可通过REST API访问，有的需通过数据库客户端，还有些则依赖带有自身认证和错误处理规则的SDK。所有这些差异都必须直接融入智能体自身的代码中。

当只有一个代理与一个工具交互时，这确实是个不错的折中方案。但一旦系统的任一维度开始扩展，这种情况就不再理想了。如果再增加一个需要相同资源的代理，要么复制现有的集成方式，要么最终有人会将其提取为独立模块，而往往是在重复代码已经出现之后才会这么做。如果改为增加第二个工具，那么该代理的代码就必须同时处理两种完全不同的集成模式。

这些集成方案很少彼此相似，因为根本没有任何规定要求如此。有的封装工具会自动重试失败的调用，而有的则根本不会重试。有的会将失败以异常的形式呈现出来，有的则会将错误隐藏在状态字段中，需要调用方主动去查看。对于“将代理与工具连接起来”这一行为究竟意味着什么，并没有统一的标准——每个集成方案最终都是以自己的方式来回答这个问题的。

这正是MCP旨在填补的空白。它并非引入智能体此前所缺乏的新功能，而是对获取已有功能的机制进行标准化，这样一来，将新智能体接入现有工具，或将新工具接入现有智能体，就不再需要从头开始编写专门的集成代码。一旦看到临时解决方案在代码中的实际表现，就能更轻松地判断它是否真的能实现这一承诺——接下来的讨论也将围绕这一点展开。

MCP出现之前：临时拼凑的工具集成方式

以一种相当常见的集成场景为例：某个需要查询外部系统的智能体，依靠各种临时拼凑的代码来维持连接功能。

import requests
class LookupToolClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key
    def lookup(self, query: str) -> dict:
        response = requests.get(
            f"{self.base_url}/search",
            params={"q": query},
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        if response.status_code != 200:
            return {"error": f"lookup failed: {response.status_code}"}
        return response.json()
def agent_node(state: GraphState) -> dict:
    client = LookupToolClient(base_url="https://internal-tool.example.com", api_key="...")
    result = client.lookup(state["extracted_fields"]["query"])
    return {"tool_result": result}

就其本身而言并没有什么问题。它是一个简洁的HTTP客户端，包含一些错误处理机制，还有一个在Node内部调用它的函数。问题出现在第二个工具加入之后——这次不是另一个REST API，而是一个结构完全不同的数据库客户端：

import psycopg2
class RecordsClient:
    def __init__(self, connection_string: str):
        self.conn = psycopg2.connect(connection_string)
    def fetch_record(self, record_id: str) -> dict:
        with self.conn.cursor() as cur:
            cur.execute("SELECT * FROM records WHERE id = %s", (record_id,))
            row = cur.fetchone()
            if row is None:
                raise ValueError(f"no record found for {record_id}")
            return dict(zip([desc[0] for desc in cur.description], row))

这两个客户端没有共同的接口，也没有统一的命名规范，甚至连处理故障的方式都不一致：一个返回错误字典，另一个则直接抛出异常。任何需要同时使用这两个客户端的程序都必须单独了解它们的特殊之处，并按各自的方式处理问题。如果再考虑到系统日后可能需要的其他工具——各自的客户端、认证机制以及故障处理方式——原本只是几处简单的集成工作，就会变成一项沉重的维护负担，因为这些组件之间没有任何统一的架构将它们联系在一起。

这里需要明确的一个基本点是：这并非编写糟糕的集成，而只是典型的集成方式，也就是在没有任何标准约束的情况下，大多数工具集成最终呈现出的样子。

MCP实际上标准化了什么

考虑到那个具体的例子，我们就能更轻松地准确描述MCP的功能，而无需依赖那些通常对它所做的更为宽泛且模糊的表述。

从根本上说，MCP定义了一种通用协议，用于向智能体暴露各种工具，无论这些工具内部如何运作，或是使用何种语言或框架构建。无需每个工具都自带符合自身规范的专用客户端，只需通过MCP服务器来暴露这些工具，该服务器会以一种可预测的标准格式展示其功能：名称、描述、输入结构以及输出结构。任何理解该协议的智能体都能发现这些工具，并以与调用其他工具相同的方式使用它们，无论其底层实现是REST接口、数据库还是其他类型的东西。

该标准化规范仅涵盖三个领域，明确指出是哪三个领域非常重要，因为人们很容易误以为MCP的适用范围比实际更广。

• 发现：智能体可以查询MCP服务器以获取其提供的工具列表，并获得结构化的响应，而无需依赖硬编码在某处的信息或与实际实现分开记录的文档。
• 调用：无论使用何种工具，每次工具调用都遵循相同的模式——具有固定结构的请求和具有固定结构的响应——而非由每个客户端自行定义方法签名和返回类型。

错误处理：故障会以统一的格式报告，因此智能体无需追踪特定工具是抛出异常、返回错误字段，还是以其他方式失败。其格式每次都保持一致。

MCP并不能免除编写工具底层逻辑的工作，也无法仅凭协议封装就保证工具的正确运行。它规范的是智能体与工具之间的接口契约，而非该契约背后实现的质量或可靠性——文章在后续通过对比进一步明确这一点后，又再次强调了这一点。

MCP服务器的架构

鉴于上述标准化措施，我们有必要了解究竟是哪些技术实现了它。从结构上看，MCP服务器不过是一组被明确定义的工具，每个工作工具都有各自的架构规范，这些工具都被封装在协议层中，从而使智能体能够以统一的方式发现并调用它们。

在MCP服务器上定义一个工具至少需要声明三个要素：智能体用来引用该工具的名称、描述预期输入格式的架构规范，以及当实际调用该工具时所执行的函数。

from mcp.server import Server
from mcp.types import Tool
server = Server("lookup-tools")
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="lookup",
            description="Search for a record by query string",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"],
            },
        )
    ]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> dict:
    if name == "lookup":
        return perform_lookup(arguments["query"])
    raise ValueError(f"unknown tool: {name}")

与之前讨论的临时客户端相比，这里有两个显著特点。首先，输入架构会明确地在最开始就声明出来，而不是从函数所接受的参数中推断——这样一来，无论是智能体还是人工审核员都能直接了解工具的需求，而无需深入研究其实现细节。其次，无论服务器内部包含多少工具，或者这些工具在底层有着多大的差异，它都只需暴露两个入口点：list_tools和call_tool即可。至于lookup工具是调用REST API、查询数据库还是执行其他操作，都完全隐藏在这两个功能接口之后。

在智能体端，无论服务器暴露了哪些工具，连接到该服务器的方式都完全相同：

from mcp.client import ClientSession
async def call_lookup_tool(query: str) -> dict:
    async with ClientSession(server_params) as session:
        result = await session.call_tool("lookup", {"query": query})
        return result

将此与之前两个临时客户端作比较：一个基于requests，另一个基于psycopg2，两者各有不同的结构与规范。在这里，无论工具内部如何实现，代理的代码都保持不变——它只需传入名称和参数集调用session.call_tool，每次都会以相同的结构收到返回结果。这种一致性正是服务器架构的真正优势，而非工具逻辑本身，因为无论如何都还是需要有人来编写工具逻辑。

通过MCP重新构建相同的集成

要直观地看到实际差异，最好的方法就是使用之前构建的完全相同的集成方案——查询工具和记录客户端——然后改用MCP来重新实现它。功能保持不变，底层系统也完全相同，只有接口发生了变化。

查询工具最初是一个基于requests的独立客户端，现在则变成了在MCP服务器上注册的工具声明：

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="lookup",
            description="Search for a record by query string",
            inputSchema={
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        ),
        Tool(
            name="fetch_record",
            description="Fetch a record by ID",
            inputSchema={
                "type": "object",
                "properties": {"record_id": {"type": "string"}},
                "required": ["record_id"],
            },
        ),
    ]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> dict:
    if name == "lookup":
        return perform_lookup(arguments["query"])
    elif name == "fetch_record":
        return fetch_record_from_db(arguments["record_id"])
    raise ValueError(f"unknown tool: {name}")

其内部逻辑丝毫未动，perform_lookup仍然调用相同的外部API，fetch_record_from_db也依然执行相同的数据库查询。不同之处在于，尽管这两个工具基于完全不同的系统，但现在它们被并列声明，拥有相同的架构结构，并通过相同的两个入口点暴露出来。

最明显的变化出现在负责调用这些函数的节点上，因为它不再需要了解 requests 或 psycopg2 的相关功能：

async def agent_node(state: GraphState) -> dict:
    async with ClientSession(server_params) as session:
        result = await session.call_tool(
            "lookup", {"query": state["extracted_fields"]["query"]}
        )
    return {"tool_result": result}

与早期版本相比，早期版本需要导入特定的 HTTP 库、解析特定的错误格式，而且仅为了用 fetch_record 替代 lookup 就需要完全独立的代码路径。而在当前版本中，调用另一个工具只需更换字符串和参数字典即可，无需再编写具有自身特殊规则的第二个客户端。

这些工具的实际功能并未发生任何变化。改变的是，调用它们的代理不再需要了解其实现细节，只需知道它们的名称以及声明的架构即可。这就是本系列之前讨论过的标准化概念，现在你可以在实际代码中看到它的体现，而无需仅凭信念行事。

对比分析：即席方式与MCP

当这两种方案都完全构建完成后，你就无需再思考理论上的优势，可以直接查看即席版本与MCP版本之间的实际差异。

• 每个新工具所需的工作量：在临时方案中，每增加一个工具就需要处理该工具专属的客户端、认证逻辑、错误格式以及一系列使用前必须掌握的规则。而在MCP架构下，添加工具只需在list_tools中新增一条记录，并在call_tool函数内增加一个分支，这些内容都会遵循之前已有工具的相同模式。
• 调用代码需要理解的内容：临时方案中的代理节点必须导入特定的客户端库并处理该库特有的错误格式。而MCP架构下的代理节点则无需导入任何与工具相关的特定内容，只需传入工具名称及参数字典调用session.call_tool即可，无论底层工具是REST接口、数据库还是其他类型，该调用方式都保持一致。

该架构在多个智能体之间的复用性如何：在临时设计中，如果第二个智能体也需要相同的查询功能，它要么直接导入同一个客户端，从而受限于该特定实现方式，要么就需要有人开发出共享的封装层来避免重复工作。而使用MCP服务器时，第二个智能体只需连接到该服务器，即可继承相同的工具集，这些工具能够以相同的方式被发现和调用，且无需额外掌握超出第一个智能体已具备的知识。

需要维护的内容：要调整查询工具的行为、添加重试策略或修改超时时间，都需要直接编辑LookupToolClient，这样依赖它的所有代理都会自动应用这些更改。MCP并未消除这种维护负担，但它将其整合了起来：现在每个工具的行为都通过相同的两个函数list_tools和call_tool来控制，而不再像以前那样分散在各个不同的客户端实现中。

这一切并没有简化工具的实际逻辑。perform_lookup和fetch_record_from_db依然需要被编写出来，并且无论如何都必须保持正常运行。发生变化的是围绕该逻辑的所有方面：它是如何被发现的、如何被调用的、错误是如何呈现的，以及这些工作负担中有多少需要由代理自身的代码承担，又有多少可以通过遵循统一的协议来自动处理。

将MCP工具集成到LangGraph系统中（续第二部分）

到目前为止，所有的示例都集中在单个代理独立调用单个工具的情况。通过将基于MCP的工具重新引入第二部分构建的多代理图中，可以填补这一空白，因为本系列三篇文章的内容实际上都在这里交汇。

回想一下第二部分中的流程图：数据接入步骤、大语言模型解析、第一部分介绍的规则引擎，以及将流程发送到通知节点或人工审核环节的条件路由机制。假设现在规则引擎在做出决策之前需要先查询某个外部系统，这种依赖关系原本意味着要像本文早前提到的那样，构建一个专用的临时客户端并直接与该节点相连。

当MCP服务器将同样的查询功能作为工具提供时，该节点的职责几乎不会发生变化。它依然会从图状态中读取数据，并将决策结果写回状态中，只不过它是通过调用session.call_tool来访问外部系统，而非使用专用客户端。

async def rules_engine_node(state: GraphState) -> dict:
    async with ClientSession(server_params) as session:
        lookup_result = await session.call_tool(
            "lookup", {"query": state["extracted_fields"]["category"]}
        )
    decision = evaluate_rules(state["extracted_fields"], lookup_result)
    return {"decision": decision["decision"], "decision_reason": decision["reason"]}

该节点在整体流程中的位置并未因此发生任何变化。它依然位于与llm_interpretation相同的下游位置，以及第二部分中已存在的条件分支的上游位置；其运行仍受该部分中规定的权限范围限制，数据记录也依然使用其中可观测性章节所描述的相同追踪ID。唯一真正的变化在于该节点与外部世界的交互方式：它不再依赖专为该节点构建的客户端，而是通过某个工具发起调用，而图中的其他节点或未来的节点都可以通过完全相同的路径来调用该工具。

这正是该系列中三大要素交汇的地方：一个仅能根据明确授权做出决策的智能体，在一个与其他智能体协同工作的图结构中运行，通过所有智能体共享的标准接口与外部系统相连。规则引擎、图结构以及MCP这三者无需详细了解其他两个组件的运作方式，只需遵循该系列一贯强调的准则：明确的职责范围、清晰的契约约定，以及杜绝任何臆测。

成本与延迟：编排机制的实际代价

在本系列中每添加一层都会带来一定的结构支撑，而这些结构都不是免费获得的。与其将成本隐藏起来，不如追踪当请求经过迄今为止所描述的所有环节后，响应时间究竟发生了怎样的变化。

考虑一个在第2部分介绍的图结构中传递的请求，现在该结构已加入上文所述的基于MCP的查询功能。其处理路径大致如下：接收层仅进行简单的格式化操作且不发起任何外部调用，因此耗时最多仅为几毫秒。大语言模型解析阶段会调用模型从原始请求中提取结构化字段，这通常是整个处理流程中最耗时的环节，根据所选模型及提示词长度的不同，往往需要额外几百毫秒的时间。规则引擎通过MCP向查询工具发起请求时，除了要等待工具本身的响应时间外，还需额外消耗一次网络往返的时间，虽然成本不容忽视，但通常仍小于大语言模型解析阶段的耗时。至于规则本身的评估，由于只是基于确定性逻辑，几乎可以视为无需成本。路由处理及最终通知环节则只会再增加少量耗时。

将这些数值加总起来就能一目了然：仅需要一次模型调用和一次工具调用的请求，其总耗时几乎完全由这两次操作决定，图表本身的协调过程几乎不会产生影响。第二部分中介绍的节点、边以及共享状态虽然提供了结构，但并不会带来额外的延迟——读取和写入共享状态对象的操作成本很低。真正耗费时间的是向模型或外部系统发起请求。

这重新定义了我们应该如何看待性能调优。在图中增加更多节点、路由分支或防护检查几乎不会影响速度，因为这些只不过是函数调用以及对字典的查询而已。真正会拖慢系统速度的，是位于请求关键路径上的每一次大语言模型调用以及每一次外部工具调用。无论周围的编排机制多么完善，由三个代理依次调用模型的系统，其运行速度都会明显慢于仅调用模型一次的系统。

实际可行的建议很简单：当工作流的延迟成为关键因素时，不要首先去研究流程图的架构。应先统计一个普通请求在完成之前需要经过的模型调用次数以及外部工具调用次数，然后思考这些操作是否有可以并行执行而非依次进行的，或者对于不需要它们的请求是否可以完全省略。

MCP 无法解决的难题

与之前讨论的 MCP 优势一样，我们也有必要坦诚地看待它的局限性，因为目前关于这个主题的大多数论述都侧重于优势，很少提及其缺陷。之前提到的三个方面——发现、调用和错误处理——同样需要从相反的角度再仔细审视一番。

• 标准化发现与调用机制并不能解决工具本身设计缺陷的问题：MCP仅规范了如何定位和调用某个工具，而非其内部运作方式。那些运行速度慢、稳定性差或结构混乱的工具，在接入MCP服务器后依然会保持这些缺陷。该协议虽将问题从调用代码转移到了工具自身的实现中，但并未让这些问题消失。

统一的错误形态并不等同于完善的错误处理：故障仍然会发生，工具仍会超时，外部系统也依然可能瘫痪。MCP能让这些故障在出现时呈现出可预测、一致的形式，但调用代码仍需负责决定后续该怎么做——是重试、回退还是将错误向上层传递，这与之前并无不同。一致的错误格式并不意味着故障真的得到了处理。

运行该协议会带来额外的运营成本：MCP服务器是需要额外运行、部署并维护的进程。对于仅调用单个简单工具的单个代理而言，这确实属于额外的开销，而本系列之前介绍的临时客户端方案在设置上更快、也更易于理解。由于该生态系统尚处于发展初期，工具支持、调试功能以及成熟的规范都还落后于像普通REST API这样的成熟技术，进一步加剧了这种开销。实际上，这意味着需要花费更多时间直接阅读源代码和规范，同时可依赖的成熟方案也相对较少。

以上内容并非反对采用MCP。这只是提醒我们，选择某种协议并不能替代其背后仍需完成的工程工作。正如前文所述，MCP确实带来了变革，但其范围比“智能体与工具的连接”这一概念整体要窄。在决定是否采用它之前，明确这一界限所在至关重要，下一节将探讨这一问题。

何时值得采用MCP

鉴于上述各种权衡，这里并没有一概而论的答案——既没有“始终应当采用”的说法，也没有“根本不必考虑”的结论。关键在于在将MCP应用于任何系统之前，先检查几项具体的条件。

• 多个代理需要使用相同的工具。前面提到的几乎所有好处都源于工具的重复利用：第二个代理可以直接接入已搭建好的服务器，而无需复制客户端或事后再创建客户端。当只有一个代理和一种工具时，这类好处还不存在——因为没有东西可以共享——此时前面提到的临时解决方案仍是更简单的选择。
• 工具的数量预计会不断增加。随着更多工具依托于标准化接口，该接口的价值也会提升。当只有两三种工具时，运行专用服务器可能还不值得承担相应开销。但一旦需要处理十种或二十种工具——每种工具否则都需要独立的定制客户端——临时解决方案的维护负担就会变成真正的负担。

系统需要在其首次发布后依然能够正常运行。MCP的优势之一在于，日后添加新的智能体或工具时无需重新设计已有的所有架构。这一优势会在系统的整个生命周期中持续发挥作用，如果只考虑初始部署的成本，就很容易忽视它。

另一方面，对于仅有一个智能体与一个稳定且不会变化的工具进行交互的场景来说，MCP并不适用。在这种情况下，直接构建专用客户端更为快捷，需要维护的组件也更少，而且MCP所提供的协调功能也没有其他使用者可以实际利用。

此处要应用的实际测试与之前用于规则引擎本身的测试类似：增加这种复杂性能否解决该系统在当前阶段确实面临的问题，还是仅仅因为这是针对系统尚未出现的问题的一种时髦解决方案而被采用。

结论：系列总结

通过三篇文章的阐述，一个系统逐步形成，每一层都在前一层的基础上构建。第一篇介绍了如何通过严格区分解释与决策行为，打造出一个足够可信、能够承担实际决策任务的单一智能体。第二篇则为该智能体增加了协作能力，将多个专业智能体整合到一个图结构中，这些智能体之间可以共享状态，并通过明确的权限控制及端到端的追踪机制，确保从请求进入系统到离开的整个过程都可被监控。本文则填补了最后的空白——即这些智能体如何突破图结构限制——通过在确实有必要时通过MCP标准化访问方式，同时明确指出在哪些情况下无需如此操作。

None of these three pieces is especially complex on its own. What actually makes the resulting system dependable is one habit, repeated at every layer without exception: responsibilities stay narrow, contracts stay explicit, and nothing is left to guesswork or allowed to happen quietly in the background. That habit is the real subject of this series, more than any particular tool—LangGraph and MCP just happened to be the frameworks used to put it into practice, but the underlying principles hold regardless of which tools you reach for.