MBE Tool 使用指南

面向集成开发者的 MCP 工具调用完整指南 — 从连接到自定义工具开发

1. 快速开始

1.1 最简调用

import httpx
import json

MCP_URL = "https://your-domain.com/mcp/message"

async def call_tool(tool_name: str, arguments: dict):
    async with httpx.AsyncClient() as client:
        resp = await client.post(MCP_URL, json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments,
            }
        })
        return resp.json()["result"]

# 咨询法律专家
result = await call_tool("ask_expert", {
    "question": "劳动合同到期不续签需要赔偿吗？",
    "device_id": "my-device-001",
})
print(result["content"][0]["text"])

1.2 连接流程

1. 初始化连接  →  initialize
2. 获取工具列表  →  tools/list
3. 调用工具  →  tools/call
4. 处理结果  →  解析 content

2. 连接 MCP 服务器

2.1 初始化

# 发送初始化请求
init_resp = await client.post(MCP_URL, json={
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
        "clientInfo": {
            "name": "my-app",
            "version": "1.0.0"
        }
    }
})

# 响应包含服务器能力
server_info = init_resp.json()["result"]
# {
#   "serverInfo": {"name": "mbe-mcp-server", "version": "1.0.0"},
#   "capabilities": {"tools": {}}
# }

2.2 获取工具列表

tools_resp = await client.post(MCP_URL, json={
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list",
    "params": {}
})

tools = tools_resp.json()["result"]["tools"]
for tool in tools:
    print(f"{tool['name']}: {tool['description'][:50]}...")

2.3 SSE 流式连接

from httpx_sse import aconnect_sse

async with httpx.AsyncClient() as client:
    async with aconnect_sse(client, "GET", "/mcp/sse") as event_source:
        async for event in event_source.aiter_sse():
            data = json.loads(event.data)
            # 处理流式事件

3. 内置工具详解

3.1 mises_analyze — 米塞斯行为分析

适用场景：用户表达不舒适、困惑、决策困难等情绪或行为问题。

result = await call_tool("mises_analyze", {
    "user_input": "最近工作压力很大，不知道该怎么办",
    "device_id": "device_001",
    "user_id": "user_001"           # 可选，用于记忆
})

响应结构：

{
  "content": [{
    "type": "text",
    "text": "我能感受到你现在面临的压力...(分析回答)"
  }]
}

注意：支持多轮对话，自动管理会话状态和米塞斯序列流转。

3.2 ask_expert — 咨询 AI 专家

适用场景：用户询问专业知识（法律、金融、医疗、技术等）。

result = await call_tool("ask_expert", {
    "question": "股票回购对公司股价有什么影响？",
    "device_id": "device_001",
    "preferred_expert": "finance"    # 可选，偏好专家类型
})

系统会自动匹配最合适的专家。如果问题涉及多个领域，可能触发多智能体编排。

3.3 list_experts — 列出可用专家

result = await call_tool("list_experts", {
    "device_id": "device_001"
})
# 返回所有可用专家的列表及其领域

3.4 mises_feedback — 反馈记录

适用场景：用户报告之前决策的执行结果。

result = await call_tool("mises_feedback", {
    "device_id": "device_001",
    "action_result": "我按照建议跟老板沟通了，效果不错",
    "satisfaction": 8                # 满意度 1-10
})

3.5 capture_and_analyze — 拍照分析

result = await call_tool("capture_and_analyze", {
    "question": "这是什么建筑风格？",
    "device_id": "device_001",
    "camera_id": "default"           # 可选
})

3.6 analyze_image — 分析上传图片

import base64

with open("photo.jpg", "rb") as f:
    image_b64 = base64.b64encode(f.read()).decode()

result = await call_tool("analyze_image", {
    "question": "帮我分析这张图片中的食物营养",
    "device_id": "device_001",
    "image_base64": image_b64        # 可选，不传则用最近上传的
})

3.7 get_camera_upload_info — 获取摄像头上传信息

result = await call_tool("get_camera_upload_info", {
    "device_id": "device_001"
})
# 返回上传地址和 token

4. 自定义工具开发

4.1 开发步骤

Step 1: 定义工具 Schema

# src/mcp/tools.py
MY_WEATHER_TOOL = {
    "name": "get_weather",
    "description": "获取指定城市的天气信息",
    "inputSchema": {
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "城市名称，如'北京'"
            },
            "device_id": {
                "type": "string",
                "description": "设备ID"
            }
        },
        "required": ["city", "device_id"]
    }
}

# 注册到工具列表
MISES_TOOLS.append(MY_WEATHER_TOOL)

Step 2: 实现处理逻辑

# src/mcp/server.py — MCPServer 类中
async def _call_get_weather(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    city = arguments["city"]
    
    # 调用天气 API（示例）
    weather_data = await self._fetch_weather(city)
    
    return {
        "content": [{
            "type": "text",
            "text": f"{city}今天：{weather_data['temp']}°C，{weather_data['desc']}"
        }]
    }

Step 3: 注册路由

# _handle_tools_call() 方法中添加
elif tool_name == "get_weather":
    return await self._call_get_weather(arguments)

4.2 工具设计规范

规范	说明
命名	使用 snake_case，动词开头（get_、analyze_、list_）
描述	清晰说明何时调用，包含示例场景
参数	`device_id` 必传（用于认证），其余按需
返回	统一使用 `{"content": [{"type": "text", "text": "..."}]}`
错误	异常时返回友好的错误信息而非堆栈
幂等	查询类工具应幂等，写入类工具做好去重

5. 外部系统集成

5.1 小智（Xiaozhi）集成

MBE 已内置小智 MCP 客户端，支持与小智智能音箱交互：

小智设备 → 小智云平台 → MCP SSE → MBE MCP Server → 工具调用 → 返回结果

配置参见：小智集成指南

5.2 QT2 量化系统集成

通过 MCP 代理模式访问远程 QT2 量化系统：

# QT2 代理工具自动注册
# 环境变量配置
QT2_MCP_URL=ws://qt2-server:9000

# 调用示例
result = await call_tool("analyze_stock", {
    "device_id": "device_001",
    "stock_code": "600519",
    "analysis_type": "fundamental"
})

5.3 第三方 LLM 集成

通过 SmartRouter 集成多个 LLM 提供商：

提供商	模型	配置
Anthropic	Claude Sonnet/Opus/Haiku 4	`ANTHROPIC_API_KEY`
OpenAI	GPT-4.1, GPT-4o, o1	`OPENAI_API_KEY`
DeepSeek	DeepSeek V3, R1	`DEEPSEEK_API_KEY`
阿里云	通义千问	`DASHSCOPE_API_KEY`
智谱	GLM-4	`ZHIPU_API_KEY`

6. 错误处理

6.1 JSON-RPC 错误码

错误码	含义	处理方式
-32600	无效请求	检查 JSON 格式
-32601	方法不存在	检查 method 名称
-32602	参数无效	检查 required 参数
-32603	内部错误	重试或联系支持

6.2 错误响应示例

{
  "jsonrpc": "2.0",
  "id": 3,
  "error": {
    "code": -32602,
    "message": "Missing required parameter: device_id"
  }
}

6.3 重试策略

import asyncio

async def call_with_retry(tool_name, arguments, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await call_tool(tool_name, arguments)
            if "error" not in result:
                return result
            if result["error"]["code"] == -32603:  # 内部错误，可重试
                await asyncio.sleep(2 ** attempt)
                continue
            raise Exception(result["error"]["message"])
        except httpx.TimeoutException:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
            else:
                raise

7. 调试技巧

7.1 查看 MCP 日志

MBE 服务器自动记录所有 MCP 请求和响应：

📥 小智请求 | 工具: ask_expert
📥 参数: {"question": "...", "device_id": "..."}
📤 MBE响应 | 工具: ask_expert
📤 内容: 根据劳动合同法第46条...
═══════════════════════════════════════════════════

7.2 使用 curl 测试

# 初始化
curl -X POST http://localhost:8001/mcp/message \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"clientInfo":{"name":"test"}}}'

# 列出工具
curl -X POST http://localhost:8001/mcp/message \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'

# 调用工具
curl -X POST http://localhost:8001/mcp/message \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"list_experts","arguments":{"device_id":"test-001"}}}'

7.3 常见问题

问题	原因	解决
工具调用超时	LLM 响应慢	增加超时时间或使用更快模型
认证失败	device_id 无效	确认设备已注册
QT2 连接失败	QT2 服务未启动	检查 `QT2_MCP_URL` 配置
返回空结果	专家未匹配	检查专家配置和知识库

📖 相关文档：Agent 框架 · API Reference · 小智集成 · Getting Started

MBE 文档中心