opc_agent/agent-craft
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(module11): restructure MCP client implementation and migrate to chapter 11

Browse Source
This commit is contained in:
Annyfee
2025-12-10 15:38:56 +08:00
parent fd3774d849
commit bca3f7486f
13 changed files with 671 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
m11_mcp_advanced/README.md
+232
View File
@@ -0,0 +1,232 @@
# 🧩 模块说明：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 │ │
│ └───────────────┘ └───────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────┘
```
---
### 🔔 全局注意事项
- **学习路径建议**：
1. 先学习独立组件：`agent_stream.py` → `final_mcp_main.py`
2. 再学习组件系统：`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框架集成
- 优化流式输出组件，支持更多展示效果
m11_mcp_advanced/__init__.py
View File
m11_mcp_advanced/agent_stream.py
+40
View File
@@ -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😊 输出结束!")
m11_mcp_advanced/final_mcp_main.py
+107
View File
@@ -0,0 +1,107 @@
import os
import asyncio
# --- 核心：导入官方库 ---
from langchain_mcp_adapters.client import MultiServerMCPClient
# LangChain/LangGraph 组件
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 m11_mcp_advanced.agent_stream import run_agent_with_streaming
from config import OPENAI_API_KEY, AMAP_MAPS_API_KEY
# === 配置 MCP 服务器 ===
MCP_SERVERS = {
# 方式1.1: 云端代理 —— stdio模式
"高德地图": {
"transport": "stdio",
"command": "npx",
"args": ["-y", "@amap/amap-maps-mcp-server"],
"env": {**os.environ, "AMAP_MAPS_API_KEY": AMAP_MAPS_API_KEY}
},
# 方式1.2: 云端MCP服务 —— Streamable HTTP模式
# "高德地图" :{
# "transport":"streamable_http",
# "url": f"https://mcp.amap.com/mcp?key={AMAP_MAPS_API_KEY}"
# },
# 方式2.1: 本地工具 —— stdio模式
# "本地天气":{
# "transport": "stdio",
# "command": "python",
# "args": ["-m", "m10_mcp_basics.stdio_server"],
# "env": None
# },
# 方式2.2:本地MCP服务 —— Streamable HTTP 模式
# 注:此方法需要提前运行m10的 streamable_http_server.py
# "本地天气":{
# "transport":"streamable_http",
# "url": "http://127.0.0.1:8001/mcp"
# }
}
def build_graph(available_tools):
"""构建图逻辑 (保持不变)"""
if not available_tools:
print("⚠️ 未加载任何工具")
llm = ChatOpenAI(
model="deepseek-chat",
api_key=OPENAI_API_KEY,
base_url="https://api.deepseek.com",
streaming=True
)
llm_with_tools = llm.bind_tools(available_tools) if available_tools else llm
sys_prompt = "你是一个地理位置助手，请根据用户需求调用工具查询信息。"
async def agent_node(state: MessagesState):
messages = [SystemMessage(content=sys_prompt)] + state["messages"]
return {"messages": [await llm_with_tools.ainvoke(messages)]}
workflow = StateGraph(MessagesState)
workflow.add_node("agent", agent_node)
if available_tools:
workflow.add_node("tools", ToolNode(available_tools))
def should_continue(state):
last_msg = state["messages"][-1]
return "tools" if last_msg.tool_calls else END
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
else:
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)
return workflow.compile()
async def main():
print("🔌 正在初始化 MCP 客户端...")
client = MultiServerMCPClient(MCP_SERVERS)
# 显式建立连接并获取工具
# 注意：这个 client 对象会保持连接，直到脚本结束
tools = await client.get_tools()
print(f"✅ 成功加载工具: {[t.name for t in tools]}")
# 构建并运行
app = build_graph(tools)
query = "帮我查一下杭州西湖附近的酒店"
await run_agent_with_streaming(app, query)
if __name__ == "__main__":
asyncio.run(main())
m11_mcp_advanced/mcp_bridge.py
+136
View File
@@ -0,0 +1,136 @@
from typing import Dict,Any,Type
from langchain_core.tools import StructuredTool
from m11_mcp_advanced.mcp_client import MCPClient
from pydantic import Field,create_model
from contextlib import AsyncExitStack
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字段定义 —— create_model 要求的特定格式
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
@classmethod
async def load_mcp_tools(cls,stack: AsyncExitStack, configs: list):
"""
负责遍历配置，批量建立连接，收集所有工具。
使用stack将连接生命周期托管给上层
"""
all_tools = []
for conf in configs:
print(f'🔌 正在连接:{conf["name"]} == ({conf.get("transport","stdio")})...')
# 根据 transport 类型创建不同的客户端
transport = conf.get("transport","stdio")
if transport == "stdio":
# 初始化 Client
client = MCPClient(
transport="stdio",
command=conf["command"],
args=conf["args"],
env=conf.get("env") # 可选参数
)
else: # http
client = MCPClient(
transport="http",
url=conf["url"]
)
# 🔥:enter_async_context 替代了async with 缩进
# 这样无论有多少个MCP，代码层级都不会变深
adapter = await stack.enter_async_context(cls(client))
# 批量获取一个MCP下的所有工具
tools = await adapter.get_tools()
print(f' ✅️ 获取工具{[t.name for t in tools]}')
all_tools.extend(tools)
return all_tools
m11_mcp_advanced/mcp_client.py
+92
View File
@@ -0,0 +1,92 @@
import uuid
from contextlib import AsyncExitStack
from typing import Optional,Literal
from .transports.base import MCPTransport
from .transports.http import HttpMCPTransport
from .transports.stdio import StdioMCPTransport
class MCPClient:
# 编辑器 _impl 必须满足MCPTransport协议
_impl:MCPTransport
def __init__(
self,
transport:Literal["stdio","http"]="stdio",
command:str=None,
args:list[str]=None,
env:dict=None,
url:str=None
):
"""
MCP 客户端 - 支持 stdio 与 HTTP 两种传输方式
:param transport: 传输模式 "stdio" / "http"
:param command: stdio 模式的命令 （如npx）
:param args: stdio 模式的参数
:param env: stdio 模式的环境变量
:param url: http模式的端点 URL
"""
if transport=="stdio":
if command is None:
raise ValueError("stdio 传输模式需要参数 'command'")
self._impl = StdioMCPTransport(command=command,args=args or [],env=env)
elif transport=="http":
if url is None:
raise ValueError("http 传输模式需要参数 'command'")
self._impl = HttpMCPTransport(url=url)
else:
raise ValueError(f"不支持的传输模式: {transport}")
async def connect(self):
""" 建立MCP连接(stdio或HTTP) """
await self._impl.connect()
async def list_tools(self):
"""查询工具列表，为LLM建立上下文用"""
return await self._impl.list_tools()
async def call_tool(self,name:str,args:dict):
"""调用工具（工程化：加上防御性处理）"""
return await self._impl.call_tool(name,args)
async def cleanup(self):
"""关闭MCP服务、会话和transport"""
return await self._impl.cleanup()
async def _http_request(self,method:str,params:dict=None):
""" 发送 HTTP JSON-RPC 请求 """
payload = {
"jsonrpc":"2.0",
"id":str(uuid.uuid4()),
"method":method,
}
if params:
payload["params"] = params
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
response = await self.http_client.post(
self.url,
json=payload,
headers=headers
)
if "Mcp-Session-Id" in response.headers:
self.session_id = response.headers["Mcp-Session-Id"]
response.raise_for_status()
return response.json()
m11_mcp_advanced/mcp_main.py
+127
View File
@@ -0,0 +1,127 @@
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 m11_mcp_advanced.agent_stream import run_agent_with_streaming
from m11_mcp_advanced.mcp_bridge import LangChainMCPAdapter
# ===环境配置===
# 复制当前py进程的环境变量,并在复制的环境变量里新增一条，确保安全可控
env_vars = os.environ.copy()
env_vars["AMAP_MAPS_API_KEY"] = AMAP_MAPS_API_KEY
MCP_SERVER_CONFIGS = [
# 方式1.1: 云端代理 —— stdio模式
{
"name":"高德地图", # 打印使用了什么MCP，可移除
"transport":"stdio", # 指定传输模式
"command":"npx",
"args":["-y", "@amap/amap-maps-mcp-server"],
"env":env_vars
}
# 方式1.2: 云端MCP服务 —— Streamable HTTP模式
# {
# "name":"高德地图",
# "transport":"http",
# "url": f"https://mcp.amap.com/mcp?key={AMAP_MAPS_API_KEY}"
# }
# 方式2.1: 本地工具 —— stdio模式
# {
# "name": "本地天气",
# "transport": "stdio",
# "command": "python",
# "args": ["-m", "m10_mcp_basics.stdio_server"],
# "env": None
# }
# 方式2.2:本地MCP服务 —— Streamable HTTP 模式
# {
# "name":"本地天气",
# "transport":"http",
# "url": "http://127.0.0.1:8001/mcp"
# }
# {...} 之后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()
# ===主程序===
async def main():
# 使用ExitStack统一管理所有资源的关闭
async with AsyncExitStack() as stack:
# A.插件(MCP)注入阶段 -- 允许为空
dynamic_tools = await LangChainMCPAdapter.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())
m11_mcp_advanced/transports/__init__.py
View File
m11_mcp_advanced/transports/base.py
+20
View File
@@ -0,0 +1,20 @@
from typing import Protocol,List,Dict,Any
class MCPTransport(Protocol):
"""MCP 传输层协议 —— 所有 transport必须实现如下方法"""
async def connect(self):
"""建立连接"""
...
async def list_tools(self):
"""获取工具列表"""
...
async def call_tool(self,name:str,args:dict):
"""调用工具并返回文本结果"""
...
async def cleanup(self):
"""清理资源"""
...
m11_mcp_advanced/transports/http.py
+130
View File
@@ -0,0 +1,130 @@
import json
import uuid
from typing import Optional,Literal
import httpx
class HttpMCPTransport:
def __init__(self,url:str=None):
"""
MCP 客户端 - 支持 stdio 与 HTTP 两种传输方式
:param url: http模式的端点 URL
"""
if not url:
raise ValueError("HTTP模式必须提供url参数")
self.url = url
self.session_id:Optional[str] = None
self.http_client:Optional[httpx.AsyncClient] = None
async def connect(self):
"""建立MCP长连接（一次连接，多次调用）"""
if self.http_client:
return
self.http_client = httpx.AsyncClient(timeout=60.0)
# 发送 initialize 请求
response = await self._http_request("initialize",{
"protocolVersion":"2024-11-05",
"capabilities":{},
"clientInfo":{"name":"mcp-client","version":"1.0"}
})
# 保存session ID
if response and "result" in response:
# Session ID在响应头内
pass # 已在_http_request中处理
async def list_tools(self):
"""查询工具列表，为LLM建立上下文用"""
if not self.http_client:
raise RuntimeError("未连接，请先 connect()")
result = await self._http_request("tools/list")
if result and "result" in result:
tools = result["result"].get("tools",[])
return [
{
"name":tool["name"],
"description":tool["description"],
"input_schema":tool.get("inputSchema",{})
}
for tool in tools
]
return []
async def call_tool(self,name:str,args:dict):
"""调用工具（工程化：加上防御性处理）"""
if not self.http_client:
raise RuntimeError("未连接，请先 connect()")
result = await self._http_request("tools/call",{
"name":name,
"arguments":args
})
if result and "result" in result:
content = result["result"].get("content",{})
if content and len(content):
return content[0].get("text",str(content[0]))
return "工具执行成功，但无文本返回"
async def cleanup(self):
"""关闭MCP服务、会话和transport"""
if self.http_client:
await self.http_client.aclose()
self.http_client = None
self.session_id = None
async def _http_request(self, method: str, params: dict = None):
""" 发送 HTTP JSON-RPC 请求 """
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": method,
}
if params:
payload["params"] = params
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
response = await self.http_client.post(
self.url,
json=payload,
headers=headers
)
if "Mcp-Session-Id" in response.headers:
self.session_id = response.headers["Mcp-Session-Id"]
response.raise_for_status()
# 判断响应类型
content_type = response.headers.get("Content-Type", "")
if "text/event-stream" in content_type:
# SSE 流式响应（本地服务器）
for line in response.text.split('\n'):
line = line.strip()
if line.startswith("data:"):
data_str = line[5:].strip()
try:
return json.loads(data_str)
except json.JSONDecodeError:
pass
return None
else:
# 普通 JSON 响应（云端服务）
return response.json()
m11_mcp_advanced/transports/stdio.py
+89
View File
@@ -0,0 +1,89 @@
from contextlib import AsyncExitStack
from typing import Optional,Literal
from mcp import ClientSession,StdioServerParameters
from mcp.client.stdio import stdio_client
class StdioMCPTransport:
def __init__(self,command:str=None,args:list[str]=None,env:dict=None):
"""
MCP 客户端 - 支持 stdio 与 HTTP 两种传输方式
:param transport: 传输模式 "stdio" / "http"
:param command: stdio 模式的命令 （如npx）
:param args: stdio 模式的参数
:param env: stdio 模式的环境变量
"""
# 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