MCP 到 LangChain 和 LangGraph 的适配器文档

安装说明

安装依赖

pip install mcp-langchain-adapter

快速上手

初始化适配器

from mcp_langchain_adapter import MCPAdapter

# 创建一个适配器实例，指定MCP服务器脚本路径
adapter = MCPAdapter(server_script_path="path/to/your/mcp_server.py")

# 同步初始化连接到MCP服务器
adapter.initialize()

获取工具列表

tools = adapter.get_tools()

# 打印可用工具的信息
print(f"发现 {len(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}")

weather_tool = adapter.get_tool_by_name("get_weather")
if weather_tool:
    result = weather_tool.run({"city": "London"})
    print(f"天气查询结果：{result}")

功能特性

工具适配

• 将MCP协议工具转换为LangChain和LangGraph兼容的BaseTool类。
• 支持自定义参数传递，包括环境变量。

服务器启动与管理

• 提供简便方法启动MCP服务器进程。
• 支持异步初始化和清理资源。

高级功能

工具包装器

from mcp_langchain_adapter import MCPToolWrapper

# 创建工具包装实例
wrapper = MCPToolWrapper(
    name="计算器",
    description="执行基本算术运算的工具",
    server_script_path="path/to/calculator_server.py",
    env={"ENV_VAR": "value"},
    args_schema={  # 可选参数定义
        "a": {"type": "int", "required": True},
        "b": {"type": "int", "required": True}
    }
)

自动检测与发现

• 通过扫描指定目录自动发现MCP工具。
• 支持动态加载新添加的工具。

示例代码

基本用法示例

from mcp_langchain_adapter import MCPAdapter

adapter = MCPAdapter("path/to/mcp_server.py")
adapter.initialize()

# 获取所有可用工具
tools = adapter.get_tools()
print(f"检测到 {len(tools)} 个工具：")

for tool in tools:
    print(f"- 名称: {tool.name}")
    print(f"  描述: {tool.description}")
    print(f"  参数: {tool.args_schema}\n")

故障排除

常见问题

1. 无法连接到MCP服务器？

   • 检查服务器脚本路径是否正确。
   • 确保服务器正在运行且监听正确端口。

2. 工具未被发现？

   • 确认工具符合适配器的规范。
   • 查看日志输出以获取更多信息。

3. 环境变量配置错误？

   • 检查env参数是否正确传递。
   • 确保服务器端能够读取这些环境变量。

日志记录

# 配置日志记录
import logging
logging.basicConfig(level=logging.DEBUG)

adapter = MCPAdapter("path/to/mcp_server.py", log_level=logging.DEBUG)

定制扩展

自定义工具加载

def custom_tool_loader(directory):
    # 手动扫描指定目录并返回MCP工具列表
    tools = []
    for script in os.listdir(directory):
        if script.endswith(".py"):
            try:
                tool = MCPToolWrapper(..., server_script_path=os.path.join(directory, script))
                tools.append(tool)
            except Exception as e:
                print(f"加载脚本 {script} 时出错：{e}")
    return tools

adapter = MCPAdapter(custom_tool_loader=custom_tool_loader)

异步支持

async def async_usage():
    adapter = MCPAdapter("path/to/mcp_server.py")
    await adapter.initialize()  # 异步初始化连接

    # 异步工具调用示例
    result = await add_tool.async_run({"a": 5, "b": 7})
    print(f"异步加法结果：{result}")

文档扩展

更多资源

• MCP协议官方文档
• LangChain社区支持

贡献指南

欢迎贡献修复和功能增强！ 请 Fork 本仓库并在提交前阅读Contributing.md。

许可证

MIT License

联系方式

• 邮件：support@mcp-langchain-adapter.com
• GitHub：https://github.com/mcp/langchain-adapter