9.7 KiB
🧩 模块说明：MCP 高级篇 - 多模态协作协议客户端实现
📌 核心知识点：MCP协议高级应用｜传输层封装｜LangChain集成｜流式输出｜多服务管理
1. agent_stream.py （通用流式输出组件）
实现通用的LangGraph事件流监听和可视化输出功能，提供友好的用户交互体验。
-
✅ 掌握点：
- LangGraph v2事件流的监听与处理
- LLM流式吐字的实时渲染
- 工具调用过程的可视化展示
- 异步事件处理的最佳实践
-
功能：
- 监听LLM的流式输出并实时打印
- 显示工具调用的开始和结束状态
- 过滤内部包装工具，只显示自定义工具
- 优化控制台输出格式，提升用户体验
💡 这是一个独立的工具组件，可以与任何LangGraph应用集成，用于增强用户交互体验。
2. final_mcp_main.py （官方库实现示例）
使用官方langchain_mcp_adapters库实现的完整MCP应用示例，展示了如何快速集成MCP服务。
-
✅ 掌握点：
- 官方MultiServerMCPClient的使用方法
- MCP服务的配置与初始化
- LangGraph工作流的构建
- 官方库与自定义组件的结合使用
-
功能演示：
- 初始化多服务器MCP客户端
- 加载高德地图MCP服务
- 构建基于LangGraph的地理位置助手
- 使用自定义流式输出组件展示结果
💡 这是一个独立的示例应用，展示了如何使用官方库快速实现MCP功能，适合作为实际项目的参考。
组件系统：自定义MCP客户端实现
以下文件共同构成一个完整的自定义MCP客户端组件系统，实现了从传输层到应用层的完整封装。
3. transports/base.py （传输层协议接口）
定义MCP传输层的抽象协议接口，为所有传输实现提供统一的规范。
-
✅ 掌握点：
- Python Protocol的使用方法
- 抽象接口的设计原则
- MCP协议的核心方法定义
-
功能：
- 定义MCP传输层必须实现的四个核心方法：connect、list_tools、call_tool、cleanup
- 提供类型注解，确保接口一致性
- 为不同传输实现提供统一的调用方式
💡 这是整个组件系统的基础，定义了传输层的契约，使得上层代码可以与具体传输实现解耦。
4. transports/http.py （HTTP传输实现）
实现基于HTTP协议的MCP传输层，支持与远程MCP服务器通信。
-
✅ 掌握点：
- HTTP JSON-RPC请求的实现
- 异步HTTP客户端的使用
- 会话管理与超时处理
- 流式响应的处理
-
功能：
- 建立与远程MCP服务器的HTTP连接
- 发送initialize请求并管理会话
- 查询工具列表和调用工具
- 处理普通JSON响应和SSE流式响应
💡 此实现支持远程MCP服务调用，适合构建分布式系统中的MCP客户端。
5. transports/stdio.py （标准输入输出传输实现）
实现基于标准输入输出的MCP传输层，支持与本地MCP服务通信。
-
✅ 掌握点：
- AsyncExitStack资源管理
- 子进程通信的实现
- MCP协议的低级实现
- 异步上下文管理器的应用
-
功能：
- 启动本地MCP服务进程
- 建立标准输入输出管道通信
- 管理MCP会话生命周期
- 自动清理资源
💡 此实现支持本地MCP服务调用，适合开发和调试阶段使用。
6. mcp_client.py （客户端主类）
实现MCP客户端的主类，封装传输层实现，提供统一的客户端接口。
-
✅ 掌握点：
- 工厂模式的应用
- 依赖注入的实现
- 客户端接口的设计
- 错误处理的最佳实践
-
功能：
- 支持stdio和http两种传输方式
- 封装连接、工具列表查询、工具调用和资源清理
- 提供统一的客户端接口，隐藏传输层细节
- 实现防御性编程，增强代码健壮性
💡 这是客户端组件的核心，为上层应用提供简洁易用的接口，同时屏蔽了底层传输的复杂性。
7. mcp_bridge.py （LangChain桥接器）
实现MCP工具到LangChain工具的自动转换，使MCP服务能够无缝集成到LangChain生态中。
-
✅ 掌握点：
- JSON Schema到Pydantic模型的动态转换
- LangChain工具的创建与配置
- 批量工具加载的实现
- 异步上下文管理器的应用
-
功能：
- 将MCP工具转换为LangChain可用的工具
- 动态生成Pydantic参数模型
- 支持批量加载多个MCP服务的工具
- 管理MCP客户端的生命周期
💡 这是MCP与LangChain集成的关键组件，实现了两种生态系统之间的无缝对接。
8. mcp_main.py （完整应用示例）
使用自定义MCP客户端组件实现的完整应用示例，展示了整个组件系统的协作使用。
-
✅ 掌握点：
- 组件系统的整体架构
- 多MCP服务的配置与管理
- LangGraph工作流的构建
- 资源的统一管理
-
功能演示：
- 配置多个MCP服务（云端和本地）
- 批量加载MCP工具
- 构建基于LangGraph的智能体
- 使用流式输出展示结果
💡 这是整个组件系统的完整演示，展示了如何使用自定义实现构建功能完整的MCP应用。
组件系统架构图
┌─────────────────────────────────────────────────────────┐
│ 应用层 │
│ ┌───────────────┐ ┌────────────────────────────────┐ │
│ │ mcp_main.py │ │ final_mcp_main.py (官方库) │ │
│ └───────────────┘ └────────────────────────────────┘ │
│ │ │ │
└──────────────┼─────────────────────┼────────────────────┘
│ │
┌──────────────┼─────────────────────┼────────────────────┐
│ 集成层 │
│ ┌───────────────┐ ┌─────────────────┐ │
│ │ mcp_bridge.py│ │ agent_stream.py │ │
│ └───────────────┘ └─────────────────┘ │
│ │ │
└──────────────┼──────────────────────────────────────────┘
│
┌──────────────┼──────────────────────────────────────────┐
│ 客户端层 │
│ ┌───────────────┐ │
│ │ mcp_client.py│ │
│ └───────────────┘ │
│ │ │
└──────────────┼──────────────────────────────────────────┘
│
┌──────────────┼──────────────────────────────────────────┐
│ 传输层 │
│ ┌───────────────┐ ┌───────────────┐ ┌─────────────┐ │
│ │ transports/ │ │ transports/ │ │ transports/ │ │
│ │ base.py │ │ http.py │ │ stdio.py │ │
│ └───────────────┘ └───────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────┘
🔔 全局注意事项
-
学习路径建议：
- 先学习独立组件：
agent_stream.py→final_mcp_main.py - 再学习组件系统：
transports/base.py→transports/http.py→transports/stdio.py→mcp_client.py→mcp_bridge.py→mcp_main.py
- 先学习独立组件：
-
环境准备：
- 所有示例依赖根目录
.env中的 API 密钥配置 - MCP服务需要Node.js环境，确保已安装并配置正确路径
- 运行前请确保已安装必要依赖：
pip install -r requirements.txt - 高德地图MCP服务需要
AMAP_MAPS_API_KEY环境变量配置
- 所有示例依赖根目录
-
运行说明：
- 独立组件可以直接运行：
python final_mcp_main.py - 组件系统示例：
python mcp_main.py - 本地MCP服务需要先启动：
python -m m10_mcp_basics.streamable_http_server
- 独立组件可以直接运行：
💡 扩展建议
- 扩展MCP客户端，支持更多高级特性（如超时控制、重试机制等）
- 实现自定义的MCP服务，与客户端组件配合使用
- 探索将MCP客户端与其他AI框架集成
- 优化流式输出组件，支持更多展示效果