LangChain集成MCP协议实现多服务协同工具链开发

1. 价值定位：连接MCP生态与LangChain的桥梁

实现跨平台工具调用标准化

通过模型上下文协议（MCP）实现不同服务间的通信标准化，解决工具调用兼容性问题。LangChain MCP Adapters作为中间层，将MCP工具转换为LangChain可识别的格式，实现跨平台工具链的无缝集成。

构建多服务协同工作流

提供多服务器管理能力，允许同时连接多个MCP服务并统一调度工具资源，打破单一服务功能边界，构建复杂业务场景的自动化工作流。

2. 场景解析：MCP适配器的典型应用场景

文档处理自动化场景

通过文件转换服务与内容提取服务的协同，实现文档从格式转换到信息提取的全流程自动化。适用于报告生成、数据录入等需要处理多种文件格式的业务场景。

数据处理流水线场景

整合数据清洗服务、格式转换服务和分析服务，构建端到端的数据处理流水线。支持从原始数据到可视化报告的全流程自动化处理。

3. 实践指南：从零构建多服务协同工具链

准备→配置→验证：搭建文件处理服务

准备工作

确保已安装必要依赖：

pip install langchain-mcp-adapters langgraph langchain-openai

配置文件处理服务器

# file_processor_server.py
from mcp.server.fastmcp import FastMCP

class FileProcessorServer:
    def __init__(self, name="FileProcessor"):
        self.mcp = FastMCP(name)
        self._register_tools()
        
    def _register_tools(self):
        @self.mcp.tool()
        def convert_to_pdf(input_path: str, output_path: str) -> bool:
            """将文件转换为PDF格式"""
            # 实际实现略，返回转换结果
            return True
            
        @self.mcp.tool()
        def extract_text(file_path: str) -> str:
            """从文件中提取文本内容"""
            # 实际实现略，返回提取的文本
            return "示例文本内容"
    
    def run(self, transport="stdio"):
        self.mcp.run(transport=transport)

if __name__ == "__main__":
    server = FileProcessorServer()
    server.run()

验证服务运行状态

启动服务器并检查运行状态：

python file_processor_server.py

✅ 验证要点：服务启动后显示"Listening for connections"

准备→配置→验证：实现数据转换服务

准备工作

创建数据转换服务所需的工具函数库。

配置数据转换服务器

# data_transform_server.py
from mcp.server.fastmcp import FastMCP

class DataTransformServer:
    def __init__(self, name="DataTransform"):
        self.mcp = FastMCP(name)
        self._register_tools()
        
    def _register_tools(self):
        @self.mcp.tool()
        def csv_to_json(input_path: str, output_path: str) -> bool:
            """将CSV文件转换为JSON格式"""
            # 实际实现略，返回转换结果
            return True
            
        @self.mcp.tool()
        def aggregate_data(data_paths: list[str], output_path: str) -> bool:
            """聚合多个数据文件"""
            # 实际实现略，返回聚合结果
            return True
    
    def run(self, transport="stdio"):
        self.mcp.run(transport=transport)

if __name__ == "__main__":
    server = DataTransformServer()
    server.run()

验证服务功能

使用测试客户端连接服务并调用工具： ✅ 验证要点：工具调用返回True且目标文件生成

准备→配置→验证：构建多服务客户端

准备工作

确保两个服务都已在后台运行。

配置多服务客户端

# multi_server_client.py
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio

class MCPToolchain:
    def __init__(self, model_name="gpt-4o"):
        self.model = ChatOpenAI(model=model_name)
        self.client = None
        self.agent = None
        
    async def initialize(self, server_config):
        """初始化多服务器客户端和代理"""
        self.client = MultiServerMCPClient(server_config)
        async with self.client:
            tools = self.client.get_tools()
            self.agent = create_react_agent(self.model, tools)
            yield self.agent
    
    async def process_document_workflow(self, document_path):
        """执行文档处理工作流"""
        if not self.agent:
            raise ValueError("Agent not initialized")
            
        messages = [
            f"请处理以下文档: {document_path}",
            "步骤1: 将文档转换为PDF",
            "步骤2: 提取文档文本",
            "步骤3: 将提取的文本转换为JSON格式"
        ]
        
        return await self.agent.ainvoke({"messages": "\n".join(messages)})

async def main():
    server_config = {
        "file_processor": {
            "command": "python",
            "args": ["file_processor_server.py"],
            "transport": "stdio"
        },
        "data_transform": {
            "command": "python",
            "args": ["data_transform_server.py"],
            "transport": "stdio"
        }
    }
    
    toolchain = MCPToolchain()
    async for agent in toolchain.initialize(server_config):
        result = await toolchain.process_document_workflow("report.docx")
        print(result)

if __name__ == "__main__":
    asyncio.run(main())

验证多服务协同

运行客户端程序并检查工作流执行结果： ✅ 验证要点：控制台输出工作流执行结果且生成最终JSON文件

4. 扩展应用：优化与定制化开发

实现异步调用优化

通过批量请求处理和连接池管理，提升多服务并发调用性能。使用异步任务队列管理工具调用顺序，避免资源竞争和超时问题。

# 异步调用优化示例
async def batch_process_documents(self, document_paths):
    """批量处理文档"""
    tasks = [self.process_document_workflow(path) for path in document_paths]
    return await asyncio.gather(*tasks)

构建跨平台适配工具链

通过配置不同传输协议（stdio、sse等），实现不同环境下的服务通信。添加服务健康检查和自动重连机制，提高工具链的稳定性。

# 跨平台传输配置示例
server_config = {
    "local_processor": {
        "command": "python",
        "args": ["file_processor_server.py"],
        "transport": "stdio"
    },
    "remote_transform": {
        "url": "http://remote-service:8000/sse",
        "transport": "sse"
    }
}

5. 问题排查指南

服务连接超时问题

错误表现：客户端无法连接到MCP服务器，提示超时
解决方案：检查服务是否已启动，验证端口占用情况，确保防火墙设置允许通信。使用netstat命令检查服务监听状态：

netstat -tuln | grep <port>

工具调用参数不匹配

错误表现：工具调用返回参数错误或类型不匹配
解决方案：使用类型检查工具验证参数类型，确保客户端与服务器端的工具定义一致。建议使用pydantic定义工具参数模型：

from pydantic import BaseModel

class ConvertParams(BaseModel):
    input_path: str
    output_path: str

@mcp.tool()
def convert_to_pdf(params: ConvertParams) -> bool:
    """类型安全的工具定义"""
    return True

多服务资源竞争

错误表现：同时调用多个服务时出现资源冲突或死锁
解决方案：实现服务调用的互斥锁机制，或使用任务队列管理工具调用顺序。设置合理的超时时间避免长时间阻塞：

# 添加超时控制的工具调用
try:
    result = await asyncio.wait_for(tool_call, timeout=30)
except asyncio.TimeoutError:
    # 处理超时情况
    pass

知识链接

• 协议转换实现：langchain_mcp_adapters/interceptors.py
• 服务发现机制：langchain_mcp_adapters/client.py