feat: add module 10 code and related content
This commit is contained in:
@@ -0,0 +1,152 @@
|
||||
# 🧩 模块说明:MCP 基础篇 - 多模态协作协议客户端实现
|
||||
|
||||
> 📌 核心知识点:MCP协议原理|客户端封装|工具调用|LangChain集成|流式输出
|
||||
|
||||
---
|
||||
|
||||
### 1️⃣ `simple_client.py` (最小化MCP客户端MVP)
|
||||
|
||||
实现最基础的MCP单次调用客户端,提供最小可行性实现,是学习MCP的起点。
|
||||
|
||||
- ✅ 掌握点:
|
||||
- MCP协议的基本调用流程
|
||||
- 异步上下文管理器的应用
|
||||
- 单次工具调用的完整生命周期
|
||||
- 资源的自动创建与清理
|
||||
|
||||
- 特点:
|
||||
- 代码精简,易于理解
|
||||
- 封装程度低,更接近协议本质
|
||||
- 适合学习和理解MCP的基本概念
|
||||
- 单次调用模式,无需维护长连接
|
||||
|
||||
> 💡 这是理解MCP协议最简单的入口,通过`run_once`方法将启动进程、握手、调用、关闭等操作封装为一次性流程。
|
||||
|
||||
---
|
||||
|
||||
### 2️⃣ `simple_main.py` (基础使用示例)
|
||||
|
||||
展示如何使用`simple_client`进行单次工具调用,是MCP应用的最简示范。
|
||||
|
||||
- ✅ 掌握点:
|
||||
- SimpleClient的基本实例化方法
|
||||
- 环境变量配置与传递
|
||||
- 工具参数构造与调用
|
||||
- 异步代码的基本编写方式
|
||||
|
||||
- 功能演示:
|
||||
- 初始化MCP客户端
|
||||
- 调用高德地图搜索功能
|
||||
- 处理并显示结果
|
||||
- 完整的单次调用生命周期
|
||||
|
||||
> 💡 从这个简单示例开始,可以直观看到MCP工具的调用过程和结果处理方式,适合初学者上手。
|
||||
|
||||
---
|
||||
|
||||
### 3️⃣ `mcp_client.py` (生产级MCP客户端)
|
||||
|
||||
实现完整的MCP客户端功能,支持长连接和多次工具调用,是生产环境的标准实现。
|
||||
|
||||
- ✅ 掌握点:
|
||||
- MCP长连接的建立与维护
|
||||
- 工具列表的动态获取
|
||||
- 多次工具调用的会话管理
|
||||
- 错误处理与异常恢复机制
|
||||
- 资源生命周期的精确控制
|
||||
|
||||
- 核心功能:
|
||||
- `connect()`: 建立与MCP服务的连接
|
||||
- `list_tools()`: 获取可用工具列表
|
||||
- `call_tool()`: 调用指定工具
|
||||
- `cleanup()`: 清理资源
|
||||
- 异步上下文管理器支持
|
||||
|
||||
> 💡 此客户端相比simple版本,增加了长连接复用、错误处理、多次调用等生产级特性,适合构建稳定的应用。
|
||||
|
||||
---
|
||||
|
||||
### 4️⃣ `mcp_bridge.py` (LangChain桥接适配器)
|
||||
|
||||
实现MCP工具到LangChain工具的自动转换,是MCP与LangChain生态集成的关键桥梁。
|
||||
|
||||
- ✅ 掌握点:
|
||||
- MCP工具元数据到LangChain工具的转换
|
||||
- JSON Schema到Pydantic模型的动态映射
|
||||
- 异步工具与LangChain的集成
|
||||
- 工具参数的类型安全转换
|
||||
|
||||
- 技术要点:
|
||||
- 自动从MCP服务获取工具定义
|
||||
- 生成符合LangChain规范的工具描述
|
||||
- 处理参数验证和类型转换
|
||||
- 管理MCP客户端的生命周期
|
||||
|
||||
> 💡 此适配器使得任何MCP服务都能无缝集成到LangChain和LangGraph工作流中,大大扩展了AI应用的能力边界。
|
||||
|
||||
---
|
||||
|
||||
### 5️⃣ `agent_stream.py` (智能体流式输出处理)
|
||||
|
||||
提供智能体运行过程的流式可视化输出,增强用户交互体验,是构建用户友好应用的重要组件。
|
||||
|
||||
- ✅ 掌握点:
|
||||
- LangGraph v2事件流处理
|
||||
- 流式文本输出的实时渲染
|
||||
- 工具调用状态的可视化展示
|
||||
- 用户交互体验优化
|
||||
|
||||
- 实现特性:
|
||||
- 监听并处理LangGraph事件
|
||||
- 实时显示AI生成内容
|
||||
- 展示工具调用开始和结束状态
|
||||
- 优化控制台输出格式
|
||||
|
||||
> 💡 此组件将抽象的智能体决策过程转化为可感知的输出,让用户能够实时了解AI的思考和行动。
|
||||
|
||||
---
|
||||
|
||||
### 6️⃣ `mcp_main.py` (综合应用完成体)
|
||||
|
||||
融合`mcp_client`、`mcp_bridge`和`agent_stream`三大核心组件,实现完整的MCP工具调用智能体应用。
|
||||
|
||||
- ✅ 掌握点:
|
||||
- 多MCP服务的批量管理与初始化
|
||||
- LangGraph工作流的构建与优化
|
||||
- LLM与工具的智能绑定
|
||||
- 条件路由逻辑实现
|
||||
- 资源的统一管理(AsyncExitStack)
|
||||
|
||||
- 系统架构:
|
||||
1. **插件化注入层**:动态加载多个MCP服务
|
||||
2. **工具适配层**:自动将MCP工具转换为LangChain格式
|
||||
3. **工作流编排层**:构建基于LangGraph的智能体决策流
|
||||
4. **用户交互层**:提供流式输出和友好界面
|
||||
|
||||
- 运行流程:
|
||||
```
|
||||
启动应用 → 加载MCP服务 → 获取工具列表 → 构建LangGraph → 执行用户查询 → 流式展示结果
|
||||
```
|
||||
|
||||
> 💡 整个模块的集大成者,展示了如何将各个组件有机结合,构建一个功能完整、架构清晰的智能体应用。
|
||||
|
||||
---
|
||||
|
||||
### 🔔 全局注意事项
|
||||
|
||||
- **学习路径建议**:严格按照文档顺序学习
|
||||
`simple_client.py` → `simple_main.py` → `mcp_client.py` → `mcp_bridge.py` → `agent_stream.py` → `mcp_main.py`
|
||||
|
||||
- **环境准备**:
|
||||
- 所有示例依赖根目录 `.env` 中的 API 密钥配置
|
||||
- MCP服务需要Node.js环境,确保已安装并配置正确路径
|
||||
- 运行前请确保已安装必要依赖:`pip install -r requirements.txt`
|
||||
- 高德地图MCP服务需要 `AMAP_MAPS_API_KEY` 环境变量配置
|
||||
|
||||
---
|
||||
|
||||
### 💡 **扩展建议**
|
||||
- 尝试集成其他MCP服务,扩展智能体的能力范围
|
||||
- 实现自定义的MCP适配器,连接私有工具服务
|
||||
- 探索将MCP客户端与其他AI框架(如LangChain之外的框架)集成
|
||||
- 实现更复杂的工作流模式,如并行工具调用、超时控制等
|
||||
@@ -0,0 +1,40 @@
|
||||
from langchain_core.messages import HumanMessage
|
||||
|
||||
async def run_agent_with_streaming(app,query:str):
|
||||
"""
|
||||
通用流式运行器,负责将 LangGraph 的运行过程可视化输出到控制台
|
||||
|
||||
:param app: 编译好的 LangGraph 应用 (workflow.compile())
|
||||
:param query: 用户输入的问题
|
||||
"""
|
||||
print(f'\n用户:{query}\n')
|
||||
print("🤖 AI:",end="",flush=True)
|
||||
|
||||
# 构造输入消息
|
||||
inputs = {"messages":[HumanMessage(content=query)]}
|
||||
|
||||
# 核心:监听v2版本的事件流(相比v1更全面)
|
||||
async for event in app.astream_events(inputs,version="v2"):
|
||||
kind = event["event"]
|
||||
|
||||
# 1.监听LLM的流式吐字(嘴在动)
|
||||
if kind == "on_chat_model_stream":
|
||||
chunk = event["data"]["chunk"]
|
||||
# 过滤掉空的chunk(有时工具调用会产生空内容)
|
||||
if chunk.content:
|
||||
print(chunk.content,end="",flush=True)
|
||||
|
||||
# 2.监听工具开始调用(手在动)
|
||||
elif kind == "on_tool_start":
|
||||
tool_name = event["name"]
|
||||
# 不打印内部包装,只打印自定义的工具
|
||||
if not tool_name.startswith("_"):
|
||||
print(f"\n\n🔨 正在调用工具: {tool_name} ...")
|
||||
|
||||
# 3.监听工具调用结束(拿到结果)
|
||||
elif kind == "on_tool_end":
|
||||
tool_name = event["name"]
|
||||
if not tool_name.startswith("_"):
|
||||
print(f"✅ 调用完成,继续思考...\n")
|
||||
print("🤖 AI: ", end="", flush=True)
|
||||
print("\n\n😊 输出结束!")
|
||||
@@ -0,0 +1,98 @@
|
||||
from typing import Dict,Any,Type
|
||||
from langchain_core.tools import StructuredTool
|
||||
from m10_mcp_basics.mcp_client import MCPClient
|
||||
from pydantic import Field,create_model
|
||||
|
||||
class LangChainMCPAdapter:
|
||||
"""
|
||||
MCP适配器:将MCP客户端无缝转换为LangChain可用的工具集。
|
||||
实现了上下文管理器协议,
|
||||
"""
|
||||
def __init__(self,mcp_client:MCPClient):
|
||||
self.client = mcp_client
|
||||
|
||||
async def __aenter__(self):
|
||||
"""进入上下文,自动建立连接"""
|
||||
await self.client.connect()
|
||||
return self
|
||||
|
||||
async def __aexit__(self,exc_type,exc_value,exc_tb):
|
||||
"""退出上下文,自动清理资源"""
|
||||
await self.client.cleanup()
|
||||
|
||||
@staticmethod
|
||||
def _schema_to_pydantic(name:str,schema:Dict[str,Any]):
|
||||
"""
|
||||
将MCP的JSON Schema动态转换为Pydantic模型
|
||||
这是让LLM理解参数要求的关键
|
||||
"""
|
||||
# print(f"🔧 调试: 工具 '{name}' 的 inputSchema = {schema}") # 查看 MCP 返回的原始 inputSchema
|
||||
|
||||
# 所有参数定义
|
||||
properties = schema.get("properties",{}) # 允许为空
|
||||
# 必需字段
|
||||
required = schema.get("required",[]) # 允许为空
|
||||
|
||||
# 初始空字典
|
||||
fields = {}
|
||||
|
||||
# 类型映射表:将JSON类型映射为Python类型
|
||||
type_map = {
|
||||
"string":str,
|
||||
"integer":int,
|
||||
"number":float,
|
||||
"boolean":bool,
|
||||
"array":list,
|
||||
"object":dict
|
||||
}
|
||||
|
||||
for field_name,field_info in properties.items():
|
||||
# 1.获取字段类型
|
||||
json_type = field_info.get("type","string")
|
||||
python_type = type_map.get(json_type,Any)
|
||||
|
||||
# 2.获取描述
|
||||
description = field_info.get("description","")
|
||||
|
||||
# 3.是否为必需项
|
||||
# 如果是必填,默认值为 ... (Ellipsis): 否则为None
|
||||
if field_name in required:
|
||||
default_value = ...
|
||||
else:
|
||||
default_value = None
|
||||
|
||||
# 4.构建Pydantic字段定义
|
||||
fields[field_name] = (python_type,Field(default=default_value,description=description))
|
||||
|
||||
# 动态创建一个Pydantic模型类
|
||||
return create_model(f"{name}Schema",**fields)
|
||||
|
||||
async def get_tools(self):
|
||||
"""
|
||||
核心方法:获取并转换工具
|
||||
返回的是标准的LangChain Tool列表,可以直接喂给bind_tools
|
||||
"""
|
||||
# 从MCP Server 获取原始工具列表
|
||||
mcp_tools = await self.client.list_tools()
|
||||
langchain_tools = []
|
||||
|
||||
|
||||
for tool_info in mcp_tools:
|
||||
# 1.动态生成参数模型 -- 要处理schema为空的情况
|
||||
# inputSchema一般会放好MCP各种工具/参数的介绍
|
||||
raw_schema = tool_info.get("input_schema",{})
|
||||
args_model = self._schema_to_pydantic(tool_info["name"],raw_schema)
|
||||
# 2.定义执行函数
|
||||
async def _dynamic_tool_func(tool_name=tool_info["name"],**kwargs):
|
||||
# ⚠️:必须绑定 tool_name 默认参数,否则循环会覆盖 tool_name
|
||||
return await self.client.call_tool(tool_name,kwargs)
|
||||
|
||||
# 3.包装成llm可调用的工具(注入args_schema)
|
||||
tool = StructuredTool.from_function(
|
||||
coroutine=_dynamic_tool_func,
|
||||
name=tool_info["name"],
|
||||
description=tool_info["description"],
|
||||
args_schema=args_model # 把说明书传给 LangChain
|
||||
)
|
||||
langchain_tools.append(tool)
|
||||
return langchain_tools
|
||||
@@ -0,0 +1,73 @@
|
||||
from contextlib import AsyncExitStack
|
||||
from typing import Optional
|
||||
from mcp import ClientSession,StdioServerParameters
|
||||
from mcp.client.stdio import stdio_client
|
||||
|
||||
|
||||
class MCPClient:
|
||||
def __init__(self,command:str,args:list[str],env:dict=None):
|
||||
# MCP启动方式(npx/uvx/python -m xxx)
|
||||
self.params = StdioServerParameters(command=command,args=args,env=env)
|
||||
# 工程核心:资源栈
|
||||
self.exit_stack = AsyncExitStack()
|
||||
# 连接会话(长连接)
|
||||
self.session:Optional[ClientSession]=None
|
||||
|
||||
async def connect(self):
|
||||
"""建立MCP长连接(一次连接,多次调用)"""
|
||||
if self.session:
|
||||
return # 已连接无需重复
|
||||
|
||||
# 进入transport(读/写管道)
|
||||
transport = await self.exit_stack.enter_async_context(
|
||||
stdio_client(self.params)
|
||||
)
|
||||
# 创建JSON-RPC对话
|
||||
self.session = await self.exit_stack.enter_async_context(
|
||||
ClientSession(transport[0],transport[1])
|
||||
)
|
||||
# 等待MCP服务器返回工具清单
|
||||
await self.session.initialize()
|
||||
|
||||
async def list_tools(self):
|
||||
"""查询工具列表,为LLM建立上下文用"""
|
||||
if not self.session:
|
||||
raise RuntimeError("未连接,请先 connect()")
|
||||
|
||||
result = await self.session.list_tools()
|
||||
|
||||
# 🔍 调试:打印工具的完整信息,确认工具是否被正确封装
|
||||
# if result.tools:
|
||||
# import json
|
||||
# # 使用 model_dump() (Pydantic v2) 或 dict() (v1) 查看原始数据
|
||||
# first_tool = result.tools[0]
|
||||
# print(f"\n🔍 [DEBUG] 原始工具数据: {first_tool}\n")
|
||||
|
||||
# 转为纯字典,LLM能读
|
||||
return[
|
||||
{
|
||||
"name":tool.name,
|
||||
"description":tool.description,
|
||||
"input_schema":tool.inputSchema
|
||||
}
|
||||
for tool in result.tools
|
||||
]
|
||||
|
||||
async def call_tool(self,name:str,args:dict):
|
||||
"""调用工具(工程化:加上防御性处理)"""
|
||||
if not self.session:
|
||||
raise RuntimeError("未连接,请先connect()")
|
||||
|
||||
result = await self.session.call_tool(name,args)
|
||||
|
||||
# 有些工具可能执行成功但无文本返回
|
||||
if hasattr(result,"content") and result.content:
|
||||
return result.content[0].text
|
||||
|
||||
return "工具执行成功,但无文本返回"
|
||||
|
||||
async def cleanup(self):
|
||||
"""关闭MCP服务、会话和transport"""
|
||||
if self.session:
|
||||
await self.exit_stack.aclose()
|
||||
self.session = None
|
||||
@@ -0,0 +1,129 @@
|
||||
import os
|
||||
import sys
|
||||
from contextlib import AsyncExitStack
|
||||
import asyncio
|
||||
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langchain_core.messages import SystemMessage
|
||||
from langgraph.graph import StateGraph,MessagesState,START,END
|
||||
from langgraph.prebuilt import ToolNode
|
||||
|
||||
from config import OPENAI_API_KEY,AMAP_MAPS_API_KEY
|
||||
from m10_mcp_basics.agent_stream import run_agent_with_streaming
|
||||
from m10_mcp_basics.mcp_client import MCPClient
|
||||
from m10_mcp_basics.mcp_bridge import LangChainMCPAdapter
|
||||
|
||||
|
||||
|
||||
# ===环境配置===
|
||||
# 环境兼容
|
||||
COMMAND = "npx.cmd" if sys.platform == "win32" else "npx"
|
||||
# 复制当前py进程的环境变量,并在复制的环境变量里新增一条,确保安全可控
|
||||
env_vars = os.environ.copy()
|
||||
env_vars["AMAP_MAPS_API_KEY"] = AMAP_MAPS_API_KEY
|
||||
|
||||
MCP_SERVER_CONFIGS = [
|
||||
{
|
||||
"name":"高德地图", # 打印使用了什么MCP,可移除
|
||||
"command":COMMAND,
|
||||
"args":["-y", "@amap/amap-maps-mcp-server"],
|
||||
"env":env_vars
|
||||
}
|
||||
# {...} 之后MCP工具可随需求扩展增加
|
||||
]
|
||||
|
||||
# ===构建图逻辑===
|
||||
def build_graph(available_tools):
|
||||
"""
|
||||
这个函数只认tools列表,不关心tools的来源
|
||||
"""
|
||||
if not available_tools:
|
||||
print('⚠️ 当前没有注入任何工具,Agent将仅靠LLM回答。')
|
||||
llm = ChatOpenAI(
|
||||
model="deepseek-chat",
|
||||
api_key=OPENAI_API_KEY,
|
||||
base_url="https://api.deepseek.com",
|
||||
streaming=True
|
||||
)
|
||||
# 如果没工具,bind_tools 会被忽略或处理,LangGraph同样能正常跑纯对话
|
||||
llm_with_tools = llm.bind_tools(available_tools) if available_tools else llm
|
||||
|
||||
|
||||
sys_prompt = """
|
||||
你是一个专业的地理位置服务助手。
|
||||
1. 当用户查询模糊地点(如"西站")时,会优先使用相关工具获取具体经纬度或标准名称。
|
||||
2. 如果用户查询"附近"的店铺,请先确定中心点的坐标或具体位置,再进行搜索。
|
||||
3. 调用工具时,参数要尽可能精确。
|
||||
"""
|
||||
|
||||
async def agent_node(state:MessagesState):
|
||||
messages = [SystemMessage(content=sys_prompt)] + state["messages"]
|
||||
# ainvoke:异步调用版的invoke
|
||||
return {"messages":[await llm_with_tools.ainvoke(messages)]}
|
||||
|
||||
workflow = StateGraph(MessagesState)
|
||||
workflow.add_node("agent",agent_node)
|
||||
|
||||
# 动态逻辑:如果有工具才加工具节点,否则就是纯对话
|
||||
if available_tools:
|
||||
tool_node = ToolNode(available_tools)
|
||||
workflow.add_node("tools",tool_node)
|
||||
|
||||
def should_continue(state:MessagesState):
|
||||
last_msg = state["messages"][-1]
|
||||
if hasattr(last_msg,"tool_calls") and last_msg.tool_calls:
|
||||
return "tools"
|
||||
return END
|
||||
|
||||
workflow.add_edge(START,"agent")
|
||||
workflow.add_conditional_edges("agent",should_continue,{"tools":"tools",END:END})
|
||||
workflow.add_edge("tools","agent")
|
||||
else:
|
||||
workflow.add_edge(START,"agent")
|
||||
workflow.add_edge("agent",END)
|
||||
|
||||
return workflow.compile()
|
||||
|
||||
|
||||
# ===MCP工具批量初始化===
|
||||
async def load_mcp_tools(stack:AsyncExitStack,configs:list):
|
||||
"""
|
||||
负责遍历配置,批量建立连接,收集所有工具。
|
||||
使用stack将连接生命周期托管给上层
|
||||
"""
|
||||
all_tools = []
|
||||
for conf in configs:
|
||||
print(f'🔌 正在连接:{conf["name"]}...')
|
||||
# 初始化 Client
|
||||
client = MCPClient(
|
||||
command=conf["command"],
|
||||
args=conf["args"],
|
||||
env=conf.get("env") # 可选参数
|
||||
)
|
||||
# 🔥:enter_async_context 替代了async with 缩进
|
||||
# 这样无论有多少个MCP,代码层级都不会变深
|
||||
adapter = await stack.enter_async_context(LangChainMCPAdapter(client))
|
||||
# 批量获取一个MCP下的所有工具
|
||||
tools = await adapter.get_tools()
|
||||
print(f' ✅️ 获取工具{[t.name for t in tools]}')
|
||||
all_tools.extend(tools)
|
||||
|
||||
return all_tools
|
||||
|
||||
# ===主程序===
|
||||
async def main():
|
||||
# 使用ExitStack统一管理所有资源的关闭
|
||||
async with AsyncExitStack() as stack:
|
||||
# A.插件(MCP)注入阶段 -- 允许为空
|
||||
dynamic_tools = await load_mcp_tools(stack,MCP_SERVER_CONFIGS)
|
||||
|
||||
# B.图构建阶段
|
||||
app = build_graph(available_tools=dynamic_tools)
|
||||
|
||||
# C.运行阶段(流式)
|
||||
query = "帮我查一下杭州西湖附近的酒店"
|
||||
await run_agent_with_streaming(app,query)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,19 @@
|
||||
from mcp import ClientSession,StdioServerParameters
|
||||
from mcp.client.stdio import stdio_client
|
||||
|
||||
class SimpleClient:
|
||||
def __init__(self,command:str,args:list[str],env:dict=None):
|
||||
# 指定要启动的工具和参数
|
||||
self.params = StdioServerParameters(command=command,args=args,env=env)
|
||||
|
||||
async def run_once(self,tool_name:str,tool_args:dict):
|
||||
# 语法糖: async with 自动帮我们 打开连接 -> 运行 -> 关闭连接
|
||||
async with stdio_client(self.params) as (read,write):
|
||||
# 建立父子进程管道(stdin/stdout)
|
||||
async with ClientSession(read,write) as session:
|
||||
# 用JSON-RPC与工具对话
|
||||
await session.initialize()
|
||||
|
||||
# 直接调用工具
|
||||
result = await session.call_tool(tool_name,tool_args)
|
||||
return result.content[0].text
|
||||
@@ -0,0 +1,24 @@
|
||||
import asyncio
|
||||
import os
|
||||
from m10_mcp_basics.simple_client import SimpleClient
|
||||
from config import AMAP_MAPS_API_KEY
|
||||
|
||||
# 复制当前py进程的环境变量,并在复制的环境变量里新增一条,确保安全可控
|
||||
env_vars = os.environ.copy()
|
||||
env_vars["AMAP_MAPS_API_KEY"] = AMAP_MAPS_API_KEY
|
||||
|
||||
async def main():
|
||||
print('🔥 正在进行单次调用...')
|
||||
client = SimpleClient(
|
||||
command="npx",
|
||||
args=["-y","@amap/amap-maps-mcp-server",AMAP_MAPS_API_KEY],
|
||||
env=env_vars
|
||||
)
|
||||
|
||||
# 这一步会经历:启动进程 - 握手 - 调用 - 杀进程
|
||||
result = await client.run_once("maps_text_search", {"keywords": "北京大学"})
|
||||
print(f'✅️ 结果:{result[:300]}')
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user