LangChain 接入 MCP 示例：构建更智能的上下文感知应用

pip install langchain langchain-mcp-adapters langchain-openai

import asyncio
from langchain_mcp_adapters.client import MCPClient
from langchain_mcp_adapters.server import MCPServer

# 假设我们有一个本地的 MCP 服务器（例如，一个文件系统工具服务器）
# 这里我们演示如何通过适配器客户端进行连接。
# 在实际场景中，你可能通过 stdio 或 HTTP 连接到服务器。

async def create_mcp_client():
    # 方法A：连接到通过 stdio 通信的本地服务器进程（常见模式）
    # 这通常需要配置服务器的启动命令
    # server = MCPServer(server_command=["python", "path/to/your/mcp_server.py"])
    # client = await server.connect()

    # 方法B：对于本教程，我们使用一个模拟的客户端来演示
    # 真实集成时，请替换为实际的连接逻辑
    from langchain_mcp_adapters.tools import load_tools_from_mcp_server
    # 假设我们有一个运行在 http://localhost:8000 的 MCP 服务器
    tools = await load_tools_from_mcp_server(
        server_type="http",
        url="http://localhost:8000/sse", # 或 /sse 端点用于 Server-Sent Events
        # 如果服务器需要认证，可以添加 headers
        # headers={"Authorization": "Bearer YOUR_TOKEN"}
    )
    return tools

# 注意：上面的 `load_tools_from_mcp_server` 是一个示意函数。
# 在 LangChain 的最新实践中，连接方式可能更简化。
# 一个更直接且常见的模式是通过 `MCPClient` 建立连接：

import subprocess
import time
from langchain_mcp_adapters.client import MCPClient

def start_and_connect_mcp_server():
    # 1. 启动 MCP 服务器进程（例如，一个提供文件读取和搜索的工具）
    # 这里 server_command 需要根据你实际使用的 MCP 服务器来调整
    server_process = subprocess.Popen(
        ["npx", "-y", "@modelcontextprotocol/server-filesystem", "/path/to/data/dir"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    # 给服务器一点启动时间
    time.sleep(2)

    # 2. 通过 stdio 连接到服务器
    # MCPClient 可以封装 stdio 通信
    client = MCPClient(server_process)
    return client, server_process

# 在实际异步环境中使用
async def main():
    client, process = start_and_connect_mcp_server()
    try:
        # 获取服务器提供的工具列表
        tools = await client.list_tools()
        print(f"可用的 MCP 工具: {[t.name for t in tools]}")
        # ... 后续使用这些工具
    finally:
        process.terminate()

from langchain.agents import Tool
from langchain_mcp_adapters.tools import convert_to_langchain_tool

async def setup_langchain_tools(mcp_client):
    # 从 MCP 客户端获取原始工具定义
    mcp_tools = await mcp_client.list_tools()
    
    langchain_tools = []
    for mcp_tool in mcp_tools:
        # 将每个 MCP 工具转换为 LangChain 工具
        lc_tool = convert_to_langchain_tool(
            mcp_tool,
            mcp_client # 传递客户端，用于实际执行工具调用
        )
        langchain_tools.append(lc_tool)
    
    return langchain_tools

# 假设我们有两个工具：`read_file` 和 `search_files`
# 转换后，它们在 LangChain 中就可以像普通工具一样被调用了。

from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType

async def create_mcp_enhanced_agent():
    # 1. 连接 MCP 服务器并获取工具
    mcp_client, process = start_and_connect_mcp_server()
    tools = await setup_langchain_tools(mcp_client)
    
    # 2. 初始化 LLM
    llm = ChatOpenAI(model="gpt-4o", temperature=0, openai_api_key="your-api-key")
    
    # 3. 创建代理
    agent = initialize_agent(
        tools,
        llm,
        agent=AgentType.OPENAI_FUNCTIONS, # 或 AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION
        verbose=True,
        handle_parsing_errors=True # 优雅地处理解析错误
    )
    
    return agent, process

# 使用代理
async def run_agent_example():
    agent, server_process = await create_mcp_enhanced_agent()
    try:
        response = await agent.ainvoke({
            "input": "请总结 `/path/to/data/dir` 目录下 `project_plan.md` 文件的主要内容，并告诉我其中提到的下一个截止日期是什么？"
        })
        print("Agent 响应:", response['output'])
    finally:
        server_process.terminate()

# 运行示例
import asyncio
asyncio.run(run_agent_example())