MCP to LangChain/LangGraph Adapter

MCP 到 LangChain/LangGraph 适配器

本项目提供了一个适配器，可让你在 LangChain 和 LangGraph 应用中使用 MCP（多模态对话流程）服务工具。借助此适配器，你可以将 MCP 工具无缝集成到 AI 应用管线中。

目录

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

简介

MCP 到 LangChain/LangGraph 适配器连接了 MCP 服务与 LangChain/LangGraph。MCP 服务通过标准化接口提供各种工具，而 LangChain/LangGraph 是用于构建大模型应用的热门框架。此适配器可让你：

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

安装

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

# If using pipenv (recommended)
pipenv install mcp langchain langchain-openai langgraph python-dotenv

# If using 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

# Load environment variables from .env file
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:
    """Add two numbers and return the result."""
    return a + b

@expose()
def get_weather(city: str) -> str:
    """
    Get the current weather for a city.

    Args:
        city: The name of the city to get weather for
    """
    # In a real application, you'd call a weather API here
    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

# Create an adapter instance, pointing to your MCP server script
adapter = MCPAdapter("simple_server.py")

# Initialize the connection and get the list of available tools
tools = adapter.get_tools()

# Print the available 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

# Initialize the language model
llm = ChatOpenAI(model="gpt-3.5-turbo")

# Create a prompt template for the agent
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)

# Create a LangChain agent with the MCP tools
agent = create_react_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

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

在 LangGraph 中使用 MCP 工具

LangGraph 提供了一种更现代、更灵活的 Agent 构建方式。下面介绍如何在 LangGraph 中使用我们的 MCP 工具：

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

# Initialize the language model
llm = ChatOpenAI(model="gpt-3.5-turbo")

# Create a memory saver for conversation history
memory = MemorySaver()

# Create a LangGraph react agent with the MCP tools
agent = create_react_agent(
    llm,
    tools,
    prompt="You are a helpful AI assistant that can use tools to solve problems.",
    checkpointer=memory
)

# Create the configuration with thread ID for memory
config = {"configurable": {"thread_id": "example-thread"}}

# Run the agent with a question
result = agent.invoke(
    {"messages": [HumanMessage(content="What is 5 + 7?")]},
    config
)

# Get the final answer
final_answer = result["messages"][-1].content
print(final_answer)

# Continue the conversation with a follow-up question
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

# Create an adapter instance
adapter = MCPAdapter("simple_server.py")

# Get all tools
tools = adapter.get_tools()

# Print information about the tools
print(f"Found {len(tools)} tools:")
for tool in tools:
    print(f"- {tool.name}: {tool.description}")

# Use a specific tool
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}")

# Use another tool
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 Agent 集成

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

主要功能：

• 连接到 MCP 服务
• 获取可用工具
• 使用这些工具创建 LangChain Agent
• 使用不同类型的查询执行 Agent

与 LangGraph Agent 集成

有关与 LangGraph Agent 集成的完整示例，请参阅 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()
    ]
)

# Rest of your MCP server code...

贡献

欢迎为改进此适配器做出贡献！以下是一些参与贡献的方式：

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

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

1. Fork 此仓库
2. 创建功能分支
3. 进行更改
4. 提交拉取请求