MCP 到 LangChain/LangGraph 适配器

本项目提供了一个适配器，允许你在 LangChain 和 LangGraph 应用中使用 MCP (多模态对话过程) 服务器工具。通过此适配器，你可以无缝地将 MCP 的工具集成到你的 AI 应用流程中。

目录

• 简介
• 安装
• 快速开始
  • 设置 MCP 服务器
  • 连接到 MCP 服务器
  • 在 LangChain 中使用 MCP 工具
  • 在 LangGraph 中使用 MCP 工具
• API 参考
  • MCPAdapter
  • MCPToolWrapper
  • 实用函数
• 示例
  • 基本用法
  • 与 LangChain Agents 集成
  • 与 LangGraph Agents 集成
• 故障排除
• 贡献

简介

MCP 到 LangChain/LangGraph 适配器弥合了 MCP 服务器（通过标准化接口提供各种工具）与 LangChain/LangGraph（用于构建大型语言模型应用的流行框架）之间的差距。 此适配器使你能够：

• 连接到 MCP 服务器
• 发现可用工具
• 将 MCP 工具转换为 LangChain 兼容的工具
• 在 LangChain agents、chains 和 LangGraph agents 中使用这些工具

安装

要使用此适配器，你需要安装必要的软件包：

# 如果使用 pipenv (推荐)
pipenv install mcp langchain langchain-openai langgraph python-dotenv

# 如果使用 pip
pip install mcp langchain langchain-openai langgraph python-dotenv

设置 API 密钥

对于使用 OpenAI 模型的示例，你需要一个 OpenAI API 密钥。 推荐的设置方法是使用 .env 文件：

1. 在你的项目根目录中创建一个 .env 文件（基于 .env.example）：

OPENAI_API_KEY=your_actual_api_key_here

2. 在你的代码中加载环境变量：

from dotenv import load_dotenv

# 从 .env 文件加载环境变量
load_dotenv()

或者，你可以直接在你的环境或代码中设置 API 密钥：

import os
os.environ["OPENAI_API_KEY"] = "your_api_key_here"

快速开始

设置 MCP 服务器

在使用适配器之前，你需要运行一个 MCP 服务器。 该适配器设计为与你提供的 MCP 服务器脚本一起使用。

1. 创建一个基本的 MCP 服务器脚本（例如，simple_server.py）：

import mcp
from mcp.server import expose

@expose()
def add(a: int, b: int) -> int:
    """添加两个数字并返回结果。"""
    return a + b

@expose()
def get_weather(city: str) -> str:
    """
    获取一个城市的当前天气。

    Args:
        city: 要获取天气的城市名称
    """
    # 在实际应用中，你将在此处调用天气 API
    return f"Weather in {city}: Sunny +11°C"

if __name__ == "__main__":
    mcp.run(transport='stdio')

此示例服务器公开了两个工具：

• add: 接受两个整数并返回它们的和
• get_weather: 接受一个城市名称并返回一个模拟的天气报告

连接到 MCP 服务器

适配器将自动管理与 MCP 服务器的连接：

from mcp_langchain_adapter import MCPAdapter

# 创建一个适配器实例，指向你的 MCP 服务器脚本
adapter = MCPAdapter("simple_server.py")

# 初始化连接并获取可用工具的列表
tools = adapter.get_tools()

# 打印可用工具
print(f"Found {len(tools)} tools:")
for tool in tools:
    print(f"- {tool.name}: {tool.description}")

在 LangChain 中使用 MCP 工具

获得工具后，你可以在 LangChain 应用中使用它们：

from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 初始化语言模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 为 agent 创建一个 prompt 模板
template = """Answer the following questions as best you can using the provided tools.

Available tools:
{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: """

prompt_template = PromptTemplate.from_template(template)

# 使用 MCP 工具创建一个 LangChain agent
agent = create_react_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行 agent
result = agent_executor.invoke({"input": "What is 5 + 7?"})
print(result["output"])

在 LangGraph 中使用 MCP 工具

LangGraph 提供了一种更现代、更灵活的方法来构建 agents。 以下是如何将我们的 MCP 工具与 LangGraph 一起使用：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

# 初始化语言模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 为对话历史记录创建一个内存保存器
memory = MemorySaver()

# 使用 MCP 工具创建一个 LangGraph react agent
agent = create_react_agent(
    llm,
    tools,
    prompt="You are a helpful AI assistant that can use tools to solve problems.",
    checkpointer=memory
)

# 创建带有线程 ID 的配置以用于内存
config = {"configurable": {"thread_id": "example-thread"}}

# 使用问题运行 agent
result = agent.invoke(
    {"messages": [HumanMessage(content="What is 5 + 7?")]},
    config
)

# 获取最终答案
final_answer = result["messages"][-1].content
print(final_answer)

# 继续对话并提出后续问题
state = memory.get("example-thread")
messages = state["messages"] + [HumanMessage(content="What's the weather in London?")]
result = agent.invoke({"messages": messages}, config)
print(result["messages"][-1].content)

API 参考

MCPAdapter

MCPAdapter 类管理与 MCP 服务器的连接并将 MCP 工具转换为 LangChain 工具。

构造函数

MCPAdapter(server_script_path: str, env: Dict[str, str] = None)

• server_script_path: 要运行的 MCP 服务器脚本的路径
• env: 服务器进程的可选环境变量

方法

• initialize(): 同步初始化与 MCP 服务器的连接
• get_tools() -> List[BaseTool]: 获取所有可用工具作为 LangChain 工具
• get_tool_names() -> List[str]: 获取所有可用工具的名称
• get_tool_by_name(name: str) -> Optional[BaseTool]: 按名称获取特定工具
• close() -> None: 清理资源 (异步方法)

MCPToolWrapper

MCPToolWrapper 类扩展了 LangChain 的 BaseTool 以包装 MCP 工具：

MCPToolWrapper(
    name: str,
    description: str,
    server_script_path: str,
    env: Optional[Dict[str, str]] = None,
    args_schema: Optional[Type[BaseModel]] = None
)

• name: 工具的名称
• description: 工具的描述
• server_script_path: MCP 服务器脚本的路径
• env: 服务器进程的可选环境变量
• args_schema: 工具参数的可选 Pydantic 模型

实用函数

• get_langchain_tools(server_script_path: str, env: Dict[str, str] = None) -> List[BaseTool]: 从 MCP 服务器获取 LangChain 工具的便捷函数

示例

基本用法

这是一个如何使用适配器的完整示例：

from mcp_langchain_adapter import MCPAdapter

# 创建一个适配器实例
adapter = MCPAdapter("simple_server.py")

# 获取所有工具
tools = adapter.get_tools()

# 打印有关工具的信息
print(f"Found {len(tools)} tools:")
for tool in tools:
    print(f"- {tool.name}: {tool.description}")

# 使用特定工具
add_tool = adapter.get_tool_by_name("add")
if add_tool:
    result = add_tool.run({"a": 5, "b": 7})
    print(f"Result of add(5, 7): {result}")

# 使用另一个工具
weather_tool = adapter.get_tool_by_name("get_weather")
if weather_tool:
    result = weather_tool.run({"city": "London"})
    print(f"Result of get_weather('London'): {result}")

与 LangChain Agents 集成

有关与 LangChain agents 集成的完整示例，请参见 example_agent_integration.py 文件。

主要特点：

• 连接到 MCP 服务器
• 检索可用工具
• 使用这些工具创建一个 LangChain agent
• 使用不同类型的查询执行 agent

与 LangGraph Agents 集成

有关与 LangGraph agents 集成的完整示例，请参见 example_langgraph_integration.py 文件。

主要特点：

• 连接到 MCP 服务器
• 检索可用工具
• 使用这些工具创建一个 LangGraph react agent
• 使用检查点管理对话历史记录
• 使用不同类型的查询执行 agent
• 展示如何流式传输 agent 的思考过程

故障排除

常见问题

1. MCP 服务器连接问题

   • 确保你的 MCP 服务器脚本的路径正确
   • 检查服务器脚本是否具有正确的运行权限
   • 确保服务器脚本正确实现了 MCP 协议

2. 工具执行错误

   • 检查工具输入格式是否正确
   • 确保该工具已在 MCP 服务器中正确定义
   • 在工具响应中查找错误消息

3. LangChain/LangGraph 集成问题

   • 验证工具是否已正确转换为 LangChain 格式
   • 检查 agent 是否已正确配置
   • 确保你将正确格式的输入传递给 agent
   • 对于 LangGraph 问题，请检查线程 ID 和内存配置

调试

要调试连接问题，你可以将日志记录添加到你的 MCP 服务器脚本：

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("mcp_server.log"),
        logging.StreamHandler()
    ]
)

# 其余的 MCP 服务器代码...

贡献

欢迎贡献以改进适配器！ 你可以通过以下方式做出贡献：

• 报告错误和问题
• 添加新功能或改进现有功能
• 改进文档
• 编写测试
• 分享与不同 LangChain/LangGraph 组件集成的示例

请按照以下步骤进行贡献：

1. Fork 存储库
2. 创建一个功能分支
3. 进行更改
4. 提交 pull request