215 lines
8.6 KiB
Markdown
215 lines
8.6 KiB
Markdown
# MCP 集成三种用法总览
|
||
|
||
本节汇总了在 **LangGraph + Qwen 自定义模型（`ChatQwenOpenAICompat`）** 下使用 MCP 工具的三种常见模式：**ReAct 代理**、**直接工具循环**、**计划-执行（Plan & Execute）**。每种模式都支持从 **代码**、**配置文件（JSON/YAML）** 或 **环境变量** 加载 MCP 服务器；既可连接本地 `stdio`（如 `npx`/`python` 启动的 MCP server），也可连接远程 **HTTP streamable** 服务器。
|
||
|
||
---
|
||
|
||
## 配置来源与优先级
|
||
|
||
可通过任意方式声明 MCP 服务器，内部合并优先级如下（后者覆盖前者同名项）：
|
||
|
||
1. **配置文件**：`config_path` 或 `MCP_CONFIG_PATH`（JSON/YAML）
|
||
2. **环境变量**：`MCP_SERVERS_JSON`（JSON 字符串）
|
||
3. **代码传入**：`servers: Dict[str, Dict[str, Any]]`
|
||
4. **兜底**：本地 `weather` HTTP server（`http://127.0.0.1:8000/mcp/`，`transport=streamable_http`）
|
||
|
||
**示例：JSON 配置（`mcp_servers.json`）**
|
||
```json
|
||
{
|
||
"servers": {
|
||
"weather": {
|
||
"url": "http://127.0.0.1:8000/mcp/",
|
||
"transport": "streamable_http"
|
||
},
|
||
"airbnb": {
|
||
"command": "npx",
|
||
"args": ["-y", "@openbnb/mcp-server-airbnb"],
|
||
"transport": "stdio"
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**示例：YAML 配置（`mcp_servers.yaml`）**
|
||
```yaml
|
||
servers:
|
||
weather:
|
||
url: http://127.0.0.1:8000/mcp/
|
||
transport: streamable_http
|
||
playwright:
|
||
command: npx
|
||
args: ["@playwright/mcp@latest"]
|
||
transport: stdio
|
||
env:
|
||
DISPLAY: ":1"
|
||
```
|
||
|
||
> 环境变量（任选）：
|
||
> - `MCP_CONFIG_PATH=./mcp_servers.yaml`
|
||
> - `MCP_SERVERS_JSON='{"weather":{"url":"http://127.0.0.1:8000/mcp/","transport":"streamable_http"}}'`
|
||
|
||
---
|
||
|
||
## 用法一：ReAct 代理（最简单、默认推荐）
|
||
|
||
**特点**
|
||
|
||
- 一行创建：自动把 MCP 工具注入 LangGraph 的 **ReAct** 代理（`ToolNode` 自动执行工具）。
|
||
- 支持 **多次工具调用**；通过 `config={"recursion_limit": N}` **显式限制步数**。
|
||
- 适合“让模型自由决策何时调用哪个工具”的通用智能体场景。
|
||
|
||
**典型场景**
|
||
|
||
- 对话检索、计划+查证、简单自动化任务编排。
|
||
- 工具数量较多、先不想自己写调用逻辑。
|
||
|
||
**最小示例**
|
||
```python
|
||
from langchain_core.messages import HumanMessage
|
||
from langgraph_qwen.mcp import create_qwen_agent_with_mcp_async
|
||
|
||
SERVERS = {
|
||
"weather": {"url": "http://127.0.0.1:8000/mcp/", "transport": "streamable_http"},
|
||
# "airbnb": {"command":"npx","args":["-y","@openbnb/mcp-server-airbnb"],"transport":"stdio"},
|
||
}
|
||
|
||
agent = await create_qwen_agent_with_mcp_async(
|
||
servers=SERVERS, # 或者传 config_path / 用 MCP_SERVERS_JSON
|
||
tool_choice="auto", # 也可 "none" 做两阶段（先思考再注入）
|
||
)
|
||
|
||
res = await agent.ainvoke(
|
||
{"messages": [HumanMessage(content="列出可用工具，演示一次调用并总结。")]},
|
||
config={"recursion_limit": 6} # ★ 控制最大工具交互步数
|
||
)
|
||
print(res["messages"][-1].content)
|
||
```
|
||
|
||
---
|
||
|
||
## 用法二：直接工具循环（完全可控）
|
||
|
||
**特点**
|
||
|
||
- 不用 LangGraph 的 ToolNode，由你在循环里**手动执行**工具（识别 `AIMessage.tool_calls`，再调用 `tool.invoke/ainvoke`，最后用 `ToolMessage` 回传）。
|
||
- 对**每一步**是否调用工具、如何合并结果、失败如何重试，有 **100%** 控制权。
|
||
- 适合需要严格可控的业务流程、精细化容错与审计。
|
||
|
||
**典型场景**
|
||
|
||
- 合规/金融/科研等对工具副作用与审计有严格要求的系统。
|
||
- 多工具“串行+并行”的复杂编排、**阶段性切换工具集**。
|
||
|
||
**最小示例**
|
||
```python
|
||
from langchain_core.messages import HumanMessage, ToolMessage
|
||
from langgraph_qwen.chat_model import ChatQwenOpenAICompat
|
||
from langgraph_qwen.mcp import load_mcp_tools
|
||
|
||
tools = await load_mcp_tools(servers={
|
||
"math": {"command":"python","args":["/abs/path/to/math_server.py"],"transport":"stdio"},
|
||
"weather":{"url":"http://127.0.0.1:8000/mcp/","transport":"streamable_http"},
|
||
})
|
||
|
||
model = ChatQwenOpenAICompat(temperature=0).bind_tools(tools).bind(tool_choice="auto")
|
||
tool_map = {t.name: t for t in tools}
|
||
|
||
msgs = [HumanMessage(content="先算 12*(3+5)，再查北京天气，最后总结。")]
|
||
for _ in range(8): # ★ 最大步骤
|
||
ai = await model.ainvoke(msgs)
|
||
msgs.append(ai)
|
||
calls = getattr(ai, "tool_calls", []) or ai.additional_kwargs.get("tool_calls", [])
|
||
if not calls:
|
||
break
|
||
for call in calls:
|
||
name, args, call_id = call["name"], call.get("args", {}), call.get("id") or ""
|
||
tool = tool_map.get(name)
|
||
if not tool:
|
||
msgs.append(ToolMessage(tool_call_id=call_id, content=f"Unknown tool: {name}"))
|
||
continue
|
||
out = await tool.ainvoke(args) if hasattr(tool, "ainvoke") else tool.invoke(args)
|
||
msgs.append(ToolMessage(tool_call_id=call_id, content=str(out)))
|
||
|
||
final = await model.ainvoke(msgs)
|
||
print(final.content)
|
||
```
|
||
|
||
---
|
||
|
||
## 用法三：计划-执行（Plan & Execute，动态工具集）
|
||
|
||
**特点**
|
||
|
||
- 第一步用模型把任务拆解为**多个步骤**（Planner）。
|
||
- 对每个步骤，可**动态选择/切换工具集**（可按阶段加载不同的 MCP 配置）。
|
||
- 执行器部分类似“直接工具循环”，可精细控制每步最多调用几次工具。
|
||
|
||
**典型场景**
|
||
|
||
- 旅行/采购/研究类 **多阶段任务**：先找资源，再比较价格，再路线规划，再汇总。
|
||
- 不同阶段接入不同 MCP 工具（如：Airbnb → Google Maps → Weather）。
|
||
|
||
**最小示例（摘录）**
|
||
```python
|
||
from langchain_core.messages import HumanMessage
|
||
from langgraph_qwen.chat_model import ChatQwenOpenAICompat
|
||
from langgraph_qwen.mcp import load_mcp_tools
|
||
|
||
# 规划
|
||
planner = ChatQwenOpenAICompat(temperature=0)
|
||
steps_ai = await planner.ainvoke([HumanMessage(content="把任务拆成可执行步骤（每行一步）：...")])
|
||
steps = [s for s in str(steps_ai.content).splitlines() if s][:8]
|
||
|
||
# 每步执行（动态加载工具集）
|
||
for i, step in enumerate(steps, 1):
|
||
tools = await load_mcp_tools(servers=(
|
||
{"weather": {"url":"http://127.0.0.1:8000/mcp/","transport":"streamable_http"}}
|
||
if i % 2 == 0 else
|
||
{"airbnb": {"command":"npx","args":["-y","@openbnb/mcp-server-airbnb"],"transport":"stdio"}}
|
||
))
|
||
model = ChatQwenOpenAICompat(temperature=0).bind_tools(tools).bind(tool_choice="auto")
|
||
# … 按“直接工具循环”方式执行，给每步设 max_tool_steps_per_step
|
||
```
|
||
|
||
---
|
||
|
||
## 选择建议
|
||
|
||
- **先用 ReAct**：如果你需要最快跑通“让模型自己决定如何用工具”的智能体，且可接受自动化行为 → **用法一**。
|
||
- **需要强控/审计/容错**：你希望精确掌控每次工具调用、失败重试、输出格式 → **用法二**。
|
||
- **多阶段任务**：需要“先规划、再按阶段注入不同工具” → **用法三**。
|
||
|
||
---
|
||
|
||
## 常见问题与提示
|
||
|
||
- **工具是异步还是同步？**
|
||
`langchain-mcp-adapters` 返回的工具通常是 `StructuredTool`/`BaseTool` 封装，可能只实现 `ainvoke`（异步）。
|
||
调用前建议检测：`await tool.ainvoke(args) if hasattr(tool, "ainvoke") else tool.invoke(args)`。
|
||
|
||
- **连接失败 (`ConnectError`)**
|
||
大多是 HTTP MCP 服务器没启动/端口不对；请确认 `url` 可访问，或本地 `stdio` 的 `command/args` 正确。
|
||
|
||
- **工具模式兼容**
|
||
后端（如 vLLM / llama.cpp / llama-box）对 `tools/tool_choice` 的支持程度有差异。若遇到 5xx/模板错误：
|
||
1) 先用最小工具 schema（`type=object`，`properties` 简单）验证；
|
||
2) 暂时将 `tool_choice="none"` 做“两阶段”：先思考生成计划，再注入目标工具并允许调用。
|
||
|
||
- **代理与鉴权**
|
||
- 如需禁用系统代理：`QWEN_HTTP_TRUST_ENV=0`（适配器会传给 `httpx`）。
|
||
- 自定义鉴权头/前缀：`QWEN_AUTH_HEADER`（默认 `Authorization`）、`QWEN_AUTH_SCHEME`（默认 `Bearer`，设空即裸 Key）。
|
||
|
||
- **配置复用**
|
||
建议把多套 MCP 服务器写在一个 `mcp_servers.yaml`，运行时以 `config_path` 选择，或用 `MCP_SERVERS_JSON` 动态注入，配合三种模式灵活切换。
|
||
|
||
---
|
||
|
||
## 相关 API（来自 `langgraph_qwen/mcp.py`）
|
||
|
||
- `resolve_servers_config(servers=None, config_path=None) -> Dict`: 合并并解析配置。
|
||
- `load_mcp_tools(servers=None, config_path=None) -> List[Tool]`: 异步加载 MCP 工具。
|
||
- `create_qwen_agent_with_mcp_async(..., tool_choice="auto")`: **ReAct** 代理（异步）。
|
||
- `create_qwen_agent_with_mcp(..., tool_choice="auto")`: **ReAct** 代理（同步包装；在异步环境请用上面的异步接口）。
|
||
|
||
> 所有模式均依赖：`pip install langchain-mcp-adapters`。工具服务端可混合 `streamable_http` 与 `stdio`（本地 `npx/python/node` 等）。
|