使用说明

MCP LangChain 集成框架

基于 Streamable HTTP 的 MCP 工具调用框架，实现与 LangChain 的无缝集成。

架构设计

┌─────────────────────────────────────────────────────────────┐
│                    LangChain 应用层                          │
├─────────────────────────────────────────────────────────────┤
│                  LangChain Adaptor 层                        │
├─────────────────────────────────────────────────────────────┤
│                 MCP 工具封装层                               │
├─────────────────────────────────────────────────────────────┤
│              Streamable HTTP 客户端层                        │
├─────────────────────────────────────────────────────────────┤
│                   MCP 服务器层                               │
└─────────────────────────────────────────────────────────────┘

功能特性

✅ 分层架构：清晰的四层架构，便于维护和扩展
✅ 流式支持：完整的流式 HTTP 通信支持
✅ LangChain 集成：与 LangChain 生态无缝集成
✅ 错误处理：完善的错误处理和重试机制
✅ 配置管理：支持环境变量配置
✅ 工具发现：自动发现 MCP 服务器上的可用工具

快速开始

1. 环境准备

# 安装依赖
pip install -r requirements.txt

2. 配置环境变量

复制 .env.example 为 .env 并修改配置：

cp .env.example .env

编辑 .env 文件：

# MCP 服务器配置
MCP_SERVER_URL=http://localhost:8000

# OpenAI API 配置
OPENAI_API_KEY=sk-82e441a293c54c0d9a760466d8a2dcc4
OPENAI_BASE_URL=https://maas.hikvision.com.cn/v1

# MCP 认证头（如果需要）
MCP_HEADERS={"Authorization": "Bearer your-mcp-token"}

3. 运行应用

# 运行示例应用
python mcp_langchain_app.py

# 或运行交互式应用
python run_app.py

核心模块说明

Streamable HTTP 客户端层 (`streamable_http_client.py`)

StreamableHTTPClient: 支持 MCP 协议的流式 HTTP 客户端
支持 Server-Sent Events (SSE) 流式响应
提供工具调用和工具列表查询功能

MCP 工具封装层 (`mcp_tools.py`)

MCPStreamableTool: 基于 Streamable HTTP 的 MCP 工具封装
MCPToolManager: MCP 工具管理器，支持工具发现和缓存

LangChain Adaptor 层 (`langchain_adaptor.py`)

MCPLangChainAdaptor: MCP 工具与 LangChain 的适配器
支持流式和非流式查询处理
提供直接工具调用功能

LangChain 应用层 (`mcp_langchain_app.py`)

MCPLangChainApplication: 完整的应用类
支持对话历史记录和保存
提供交互式对话界面

高级功能

error_handling.py: 错误处理和重试机制
config.py: 配置管理类
run_app.py: 交互式应用运行入口

使用示例

基本使用

import asyncio
from mcp_langchain_app import MCPLangChainApplication

async def main():
    app = MCPLangChainApplication(
        mcp_server_url="http://localhost:8000",
        openai_api_key="your-api-key"
    )
    
    await app.initialize()
    response = await app.chat("今天北京的天气怎么样？")
    print(response)

asyncio.run(main())

流式对话

async def stream_chat():
    app = MCPLangChainApplication(
        mcp_server_url="http://localhost:8000",
        openai_api_key="your-api-key"
    )
    
    await app.initialize()
    await app.chat_stream("请介绍一下人工智能")

直接工具调用

async def direct_tool_call():
    app = MCPLangChainApplication(
        mcp_server_url="http://localhost:8000",
        openai_api_key="your-api-key"
    )
    
    await app.initialize()
    result = await app.call_tool_directly("weather", {"city": "上海"})
    print(result)

配置说明

环境变量

MCP_SERVER_URL: MCP 服务器地址（默认：http://localhost:8000）
OPENAI_API_KEY: OpenAI API 密钥（必需）
OPENAI_BASE_URL: OpenAI API 基础 URL（默认：https://maas.hikvision.com.cn/v1）
MCP_HEADERS: MCP 认证头（JSON 格式）

代码配置

也可以通过代码直接配置：

from config import MCPConfig

config = MCPConfig()
config.mcp_server_url = "http://your-mcp-server:8000"
config.openai_api_key = "your-api-key"

错误处理

框架内置了错误处理和重试机制：

from error_handling import RobustMCPAdaptor

robust_adaptor = RobustMCPAdaptor(...)
result = await robust_adaptor.robust_process_query("你的查询")

开发说明

添加新的 MCP 工具

框架会自动发现 MCP 服务器上的可用工具，无需手动添加。

自定义工具封装

如果需要自定义工具行为，可以继承 MCPStreamableTool 类：

class CustomMCPTool(MCPStreamableTool):
    async def _arun(self, **kwargs):
        # 自定义实现
        pass

故障排除

常见问题

连接失败: 检查 MCP 服务器是否运行，URL 是否正确
认证失败: 检查 MCP_HEADERS 配置是否正确
API 调用失败: 检查 OpenAI API 密钥和基础 URL

调试模式

启用详细日志输出：

import logging
logging.basicConfig(level=logging.DEBUG)

许可证

MIT License