【MCP集成实战】手把手打通LangChain + LangGraph + MCP生态！构建可交互AI Agent全流程
MCP (Model Context Protocol)定义：模型上下文协议，是一个开放标准，用于连接AI模型与外部数据源和工具。核心特性：标准化接口：统一的协议规范，支持多种数据源工具抽象：将外部功能抽象为工具，供AI模型调用资源管理：管理文件、数据库连接等资源流式支持：支持流式数据处理
🔍 核心概念解析
1. MCP (Model Context Protocol)
定义：模型上下文协议，是一个开放标准，用于连接AI模型与外部数据源和工具。
核心特性：
- 标准化接口：统一的协议规范，支持多种数据源
- 工具抽象：将外部功能抽象为工具，供AI模型调用
- 资源管理：管理文件、数据库连接等资源
- 流式支持：支持流式数据处理
协议组成：
{ "mcpServers": { "server-name": { "command": "启动命令", "env": { "TRANSPORT": "stdio", "DSN": "数据源连接串", "READONLY": "true" } } }}
2. LangChain
定义：一个用于构建LLM应用的框架，提供组件化、可组合的AI应用开发工具。
核心组件：
- Chains：链式调用多个组件
- Agents：智能体，能够使用工具
- Tools：工具，执行特定功能
- Memory：记忆组件，存储对话历史
- Callbacks：回调机制，监控执行过程
3. LangGraph
定义：基于LangChain的状态图框架，用于构建复杂的多智能体工作流。
核心特性：
- 状态管理：管理智能体间的状态传递
- 条件路由：根据条件决定下一步执行
- 并行处理：支持并行执行多个智能体
- 可视化：提供工作流图的可视化
4. MCP Server
定义：实现MCP协议的服务器，提供特定数据源或功能的访问接口。
常见MCP Server：
- @bytebase/dbhub：数据库连接器（MySQL、PostgreSQL等）
- mcp-server-filesystem：文件系统访问
- mcp-server-http：HTTP API访问
- mcp-server-weather：天气数据服务
5. MCP Client
定义：MCP协议的客户端实现，用于连接和调用MCP Server。
主要实现：
- mcp-python：Python官方客户端
- mcp-cli：命令行客户端
- langchain-mcp-adapters：LangChain集成客户端
6. langchain-mcp-adapters
定义：LangChain官方提供的MCP适配器，将MCP工具转换为LangChain工具。
核心功能：
- 工具转换：MCP工具 → LangChain工具
- 服务器管理：管理多个MCP服务器连接
- 资源处理：处理MCP资源（文件、数据等）
- 异步支持：支持异步操作
🏗️ 技术架构关系
整体架构图
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐│ LangGraph │ │ LangChain │ │ MCP Client ││ (工作流编排) │◄──►│ (智能体框架) │◄──►│ (协议客户端) │└─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ ▼ ▼ ┌─────────────────┐ ┌─────────────────┐ │ MCP Adapters │ │ MCP Protocol │ │ (工具转换层) │◄──►│ (协议层) │ └─────────────────┘ └─────────────────┘ │ ▼ ┌─────────────────┐ │ MCP Servers │ │ (数据源服务) │ └─────────────────┘
数据流向
- LangGraph 编排工作流，调用 LangChain 智能体
- LangChain 智能体通过 MCP Adapters 获取工具
- MCP Adapters 将MCP工具转换为LangChain工具
- MCP Client 通过协议与 MCP Server 通信
- MCP Server 访问实际的数据源（数据库、API、文件等）
📁 项目结构概览
核心目录结构
intelligent_modeling2/├── agents/ # 智能体定义│ ├── base_agent.py # 基础智能体类│ ├── requirement_analyst.py # 需求分析智能体│ ├── data_modeler.py # 数据建模智能体│ └── ...├── utils/ # 工具和配置│ ├── env_config.py # 环境变量配置管理│ ├── mcp_config_manager.py # MCP配置管理器│ ├── cache_manager.py # 缓存管理器│ ├── logger.py # 统一日志系统│ ├── graph.py # LangGraph工作流定义│ ├── langchain_mcp_integration.py # LangChain MCP集成│ ├── mcp_database_connector.py # MCP数据库连接器│ └── ...├── examples/ # 示例和演示│ ├── data_warehouse_modeling.py # 数仓建模主程序│ ├── simple_data_modeling.py # 简化建模流程│ ├── mcp_bytebase_integration.py # MCP集成示例│ ├── research_assistant.py # 研究助手│ └── ...├── tests/ # 测试文件│ ├── test_mcp_integration.py # MCP集成测试│ └── ...├── docs/ # 文档│ ├── MCP_Integration_Guide.md # MCP集成指南│ ├── Simple_Modeling_Guide.md # 简化建模指南│ └── ...├── cache/ # 缓存目录├── outputs/ # 输出目录├── mcp_config.json # MCP服务器配置├── env.example # 环境变量示例└── main.py # 主程序入口
配置文件说明
1. mcp_config.json - MCP服务器配置
{ "mcpServers": { "dbhub-mysql-demo": { "command": "npx @bytebase/dbhub", "env": { "TRANSPORT": "stdio", "DSN": "mysql://user:pass@host:port/db", "READONLY": "true" } } }}
2. env.example - 环境变量配置
# LLM模型配置LLM_TEMPERATURE=0.1LLM_MAX_TOKENS=4000LLM_TIMEOUT=300LLM_RETRY_COUNT=5LLM_RETRY_DELAY=5
# 工作流配置WORKFLOW_RECURSION_LIMIT=50WORKFLOW_MAX_LOOPS_PER_STAGE=3
# MCP配置MCP_INIT_TIMEOUT=5.0MCP_READ_TIMEOUT=10.0MCP_MYSQL_CONNECT_TIMEOUT=30MCP_MYSQL_READ_TIMEOUT=30MCP_MYSQL_WRITE_TIMEOUT=30
# API密钥OPENAI_API_KEY=your_openai_api_key_hereSILICONFLOW_API_KEY=your_siliconflow_api_key_hereSILICONFLOW_API_BASE=https://api.siliconflow.cn/v1
🔧 MCP集成方式
方式一：直接使用MCP Python客户端
优点
- 简单直接：无需额外框架
- 性能好：直接协议通信
- 灵活性高：完全控制通信过程
缺点
- 需要手动管理：工具注册、错误处理等
- 集成复杂：需要自己处理与智能体的集成
实现示例
from mcp import MCPClientimport asyncio
class DirectMCPClient: def __init__(self): self.clients = {}
async def connect_database(self, name: str, dsn: str): """连接数据库""" client = MCPClient( command="npx @bytebase/dbhub", env={ "TRANSPORT": "stdio", "DSN": dsn, "READONLY": "true" } ) self.clients[name] = client return client
async def get_schema(self, db_name: str, table_name: str = None): """获取schema信息""" client = self.clients.get(db_name) if not client: return None
if table_name: return await client.get_table_schema(table_name) else: return await client.list_tables()
async def execute_query(self, db_name: str, query: str): """执行查询""" client = self.clients.get(db_name) if not client: return None
return await client.execute_sql(query)
# 使用示例async def example_usage(): client = DirectMCPClient()
# 连接MySQL await client.connect_database( "mysql-demo", "mysql://user:pass@host:port/db" )
# 获取schema schema = await client.get_schema("mysql-demo") print(f"Tables: {schema}")
# 执行查询 result = await client.execute_query( "mysql-demo", "SELECT COUNT(*) FROM orders" ) print(f"Result: {result}")
方式二：使用langchain-mcp-adapters
优点
- 无缝集成：与LangChain生态系统完美集成
- 工具自动转换：MCP工具自动转换为LangChain工具
- 智能体友好：智能体可以直接使用工具
- 标准化：遵循LangChain工具标准
缺点
- 依赖较重：需要安装LangChain相关包
- 学习成本：需要了解LangChain概念
实现示例
from langchain_mcp import MCPToolkitfrom langchain_mcp.server import MCPServerfrom langchain_core.tools import BaseToolimport asyncio
class LangChainMCPManager: def __init__(self): self.servers = {} self.toolkits = {} self.tools = []
async def register_database(self, name: str, dsn: str): """注册数据库MCP服务器""" server = MCPServer( name=name, command="npx @bytebase/dbhub", env={ "TRANSPORT": "stdio", "DSN": dsn, "READONLY": "true" } )
self.servers[name] = server toolkit = MCPToolkit(server) self.toolkits[name] = toolkit
# 获取工具 tools = await toolkit.get_tools() self.tools.extend(tools)
return tools
def get_tools_for_agent(self): """获取供智能体使用的工具""" return self.tools
async def execute_tool(self, tool_name: str, **kwargs): """执行工具""" for tool in self.tools: if tool.name == tool_name: return await tool.ainvoke(kwargs) return None
# 在LangGraph中使用from langchain_core.tools import toolfrom langgraph import StateGraph, END
# 创建MCP管理器mcp_manager = LangChainMCPManager()
# 注册数据库await mcp_manager.register_database( "mysql-demo", "mysql://user:pass@host:port/db")
# 获取工具tools = mcp_manager.get_tools_for_agent()
# 在智能体中使用@toolasync def query_database(query: str): """查询数据库""" return await mcp_manager.execute_tool("query", sql=query)
# 创建工作流workflow = StateGraph(AgentState)workflow.add_node("query_agent", query_database)workflow.add_edge("query_agent", END)
🔗 相关资源
官方文档
- MCP协议规范
- LangChain文档
- LangGraph文档
- langchain-mcp-adapters
常用MCP Server
- @bytebase/dbhub - 数据库连接器
- mcp-server-filesystem - 文件系统
- mcp-server-http - HTTP API
示例项目
- MCP Python客户端示例
- LangChain MCP集成示例
📝 总结
MCP协议为AI应用提供了强大的数据源集成能力，通过三种不同的集成方式，可以满足不同场景的需求：
- 直接MCP客户端：适合简单、直接的场景
- langchain-mcp-adapters：适合LangChain生态系统
- 自定义适配器：适合需要深度定制的场景
选择合适的集成方式，结合最佳实践，可以构建出强大、可靠的AI应用系统。
最后
为什么要学AI大模型
当下，⼈⼯智能市场迎来了爆发期，并逐渐进⼊以⼈⼯通⽤智能（AGI）为主导的新时代。企业纷纷官宣“ AI+ ”战略，为新兴技术⼈才创造丰富的就业机会，⼈才缺⼝将达 400 万！
DeepSeek问世以来，生成式AI和大模型技术爆发式增长，让很多岗位重新成了炙手可热的新星，岗位薪资远超很多后端岗位，在程序员中稳居前列。
与此同时AI与各行各业深度融合，飞速发展，成为炙手可热的新风口，企业非常需要了解AI、懂AI、会用AI的员工，纷纷开出高薪招聘AI大模型相关岗位。
最近很多程序员朋友都已经学习或者准备学习 AI 大模型，后台也经常会有小伙伴咨询学习路线和学习资料，我特别拜托北京清华大学学士和美国加州理工学院博士学位的鲁为民老师给大家这里给大家准备了一份涵盖了AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频 全系列的学习资料，这些学习资料不仅深入浅出，而且非常实用，让大家系统而高效地掌握AI大模型的各个知识点。
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
AI大模型系统学习路线
在面对AI大模型开发领域的复杂与深入，精准学习显得尤为重要。一份系统的技术路线图，不仅能够帮助开发者清晰地了解从入门到精通所需掌握的知识点，还能提供一条高效、有序的学习路径。
但知道是一回事，做又是另一回事，初学者最常遇到的问题主要是理论知识缺乏、资源和工具的限制、模型理解和调试的复杂性，在这基础上，找到高质量的学习资源，不浪费时间、不走弯路，又是重中之重。
AI大模型入门到实战的视频教程+项目包
看视频学习是一种高效、直观、灵活且富有吸引力的学习方式，可以更直观地展示过程，能有效提升学习兴趣和理解力，是现在获取知识的重要途径
光学理论是没用的，要学会跟着一起敲，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战案例来学习。
海量AI大模型必读的经典书籍（PDF）
阅读AI大模型经典书籍可以帮助读者提高技术水平，开拓视野，掌握核心技术，提高解决问题的能力，同时也可以借鉴他人的经验。对于想要深入学习AI大模型开发的读者来说，阅读经典书籍是非常有必要的。
600+AI大模型报告（实时更新）
这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。
AI大模型面试真题+答案解析
我们学习AI大模型必然是想找到高薪的工作，下面这些面试题都是总结当前最新、最热、最高频的面试题，并且每道题都有详细的答案，面试前刷完这套面试题资料，小小offer，不在话下
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
14
0- 0
已为社区贡献121条内容
所有评论(0)