MCP 多工具协作链路设计：打造真正的智能工作流

[TOC]

🚀 MCP 多工具协作链路设计：打造真正的智能工作流

🌟 多工具协作链核心思想

🛠️ 设计示例：智能文档分析系统

📑 1. MCP Server 定义多工具

list_txt_files.py

read_file_content.py

summarize_text.py

📈 2. 设计协作链逻辑（LangChain示例）

🧪 3. 运行与观察

🧠 技术总结

🎯 预告：下一步探索

[TOC]

🚀 MCP 多工具协作链路设计：打造真正的智能工作流

在前面的教程中，我们已经完成了：

MCP Server 与 MCP Client 搭建
将 MCP 工具接入 LangChain 和 Semantic Kernel
单工具调用测试

但在真实世界中，用户需求往往不是调用一个工具就能搞定的。
比如一个复杂任务可能需要：

文件检索 → 文件分析 → 结果总结 → 报告生成

因此，本篇我们将设计 MCP 多工具协作链路，让 AI 具备自动串联多工具完成任务的能力，真正实现智能工作流！

🌟 多工具协作链核心思想

MCP 的每个 Tool 相当于一个“技能”或“动作”，
而多工具协作，就是让 AI：

智能判断什么时候用哪个工具
有序串联多个工具执行任务
组合输出最终结果

用一句话总结：

工具即函数，协作即流程，智能体即调度器。

🛠️ 设计示例：智能文档分析系统

假设我们希望 AI 完成如下复杂任务：

列出当前目录下所有 .txt 文件
读取每个 .txt 文件的内容
总结每个文件的核心信息
生成一份整体摘要报告

那么，我们需要 MCP Server 提供至少 3 个工具：

工具名	功能说明
`list_txt_files`	列出所有 .txt 文件
`read_file_content`	读取单个文件内容
`summarize_text`	总结一段文字的要点

📑 1. MCP Server 定义多工具

在 tools/ 目录中定义三个工具：

list_txt_files.py

import os
from mcp.server.fastmcp import tool@tool()
def list_txt_files(directory: str = ".") -> list[str]:"""列出指定目录下所有 .txt 文件"""return [f for f in os.listdir(directory) if f.endswith(".txt")]

read_file_content.py

from mcp.server.fastmcp import tool@tool()
def read_file_content(file_path: str) -> str:"""读取指定文件的内容"""with open(file_path, 'r', encoding='utf-8') as f:return f.read()

summarize_text.py

from mcp.server.fastmcp import tool@tool()
def summarize_text(text: str) -> str:"""总结文本要点（示例版，可接LLM微调版更智能）"""# 这里只是简单截断前200字作为示例return text[:200] + "..."

然后在 server.py 中导入注册所有工具模块：

import tools.list_txt_files
import tools.read_file_content
import tools.summarize_text

📈 2. 设计协作链逻辑（LangChain示例）

在 LangChain 中，可以用SequentialChain来串联多个步骤。

示例代码：

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from mcp.client import MCPClient# MCP工具封装
class MCPTool(Tool):def __init__(self, tool_name, server_url="http://localhost:8000"):super().__init__(name=tool_name, description=f"MCP Tool {tool_name}")self.tool_name = tool_nameself.client = MCPClient(server_address=server_url)self.client.connect()def _run(self, input_text: str) -> str:params = {"directory": input_text} if self.tool_name == "list_txt_files" else {"file_path": input_text, "text": input_text}result = self.client.call_tool(self.tool_name, params)return str(result)async def _arun(self, input_text: str) -> str:raise NotImplementedError()# 初始化 MCP 工具
list_files_tool = MCPTool("list_txt_files")
read_file_tool = MCPTool("read_file_content")
summarize_tool = MCPTool("summarize_text")# 组装工具链
agent = initialize_agent(tools=[list_files_tool, read_file_tool, summarize_tool],llm=OpenAI(temperature=0),agent="zero-shot-react-description",verbose=True,
)# 测试复杂指令
response = agent.run("请帮我列出当前目录下所有txt文件，读取每个文件内容，总结要点，最后形成一份整体报告。")
print(response)