给语音智能体接入 MCP 工具：让它真的去查订单

用户问「我的订单到哪了」，模型只能猜——它不知道你库里有什么。而模型猜出来的物流状态，比不回答更糟。

MCP（Model Context Protocol）解决的就是这件事：用一套统一协议让模型去发现和调用你自己的工具，不必为每个模型各写一遍胶水代码。接上之后，涉及真实数据的问题会走到你的接口上，答案有据可查，出错也能定位到具体哪个工具。

下面搭一个公网 Streamable HTTP 的 MCP 服务，提供「获取北京时间」和「查询订单」两个工具，然后挂到语音智能体的 DeepSeek 上。做完之后，用户问「SW1001 到哪了」，智能体会真的去查库，而不是编一个物流状态出来。

一. 架构与准备工作

用户：“帮我查一下 SW1001”
       ↓
凤鸣 ASR → DeepSeek ─MCP Streamable HTTP→ lookup_order
       ↑             ←─“已发货，预计明天送达”─┘
MiniMax TTS

除了快速开始那四个密钥，还需要一个声网云端能访问的 HTTPS 地址。

二. 环境变量

MCP_ENDPOINT=https://REPLACE_WITH_PUBLIC_HOST/mcp
MCP_SHARED_SECRET=REPLACE_WITH_RANDOM_SECRET

MCP_SHARED_SECRET 只放服务端，同时配到创建智能体请求的 MCP Header 里。千万别写进浏览器代码。

三. 最小 MCP 服务

requirements.txt：

fastapi==0.116.1
mcp==1.12.3
uvicorn==0.35.0

mcp_service.py：

import os
from contextlib import asynccontextmanager
from datetime import datetime
from zoneinfo import ZoneInfo
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("shengwang-business-tools")

@mcp.tool()
def get_beijing_time() -> str:
    """返回北京时间。"""
    return datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")

@mcp.tool()
def lookup_order(order_id: str) -> str:
    """查询订单状态；生产环境请替换为真实业务 API。"""
    demo = {"SW1001": "已发货，预计明天送达", "SW1002": "待付款"}
    return demo.get(order_id.upper(), "没有找到这个订单")

@asynccontextmanager
async def lifespan(_: FastAPI):
    async with mcp.session_manager.run():
        yield

app = FastAPI(lifespan=lifespan)

@app.middleware("http")
async def require_bearer(request: Request, call_next):
    expected = os.getenv("MCP_SHARED_SECRET")
    if not expected or request.headers.get("authorization") != f"Bearer {expected}":
        return JSONResponse({"detail": "invalid credential"}, status_code=401)
    return await call_next(request)

app.mount("/", mcp.streamable_http_app())

运行：

pip install -r requirements.txt
uvicorn mcp_service:app --host 0.0.0.0 --port 9100

部署完用 MCP Inspector 或者任意 Streamable HTTP 客户端跑一下 tools/list 和 tools/call。别用普通的 curl GET 判断 MCP 是否正常——它是有会话语义的协议，GET 打过去看着不对劲，不代表服务真有问题。

四. 把 MCP 挂到 DeepSeek 上

在快速开始的 DeepSeekLLM 构造器里加上 mcp_servers，同时打开工具能力：

llm = DeepSeekLLM(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com/chat/completions",
    model="deepseek-v4-flash",
    system_messages=[{
        "role": "system",
        "content": (
            "你是中文语音助手。用户询问时间或订单时必须调用工具，"
            "不能猜测结果。最终只播报简短结论。"
        ),
    }],
    greeting_message="你好，我可以帮你查时间或订单。",
    mcp_servers=[{
        "name": "business-tools",
        "endpoint": os.environ["MCP_ENDPOINT"],
        "transport": "streamable_http",
        "headers": {
            "Authorization": f"Bearer {os.environ['MCP_SHARED_SECRET']}",
        },
        "allowed_tools": ["get_beijing_time", "lookup_order"],
        "timeout_ms": 10000,
    }],
)

agent = Agent(
    client=client,
    advanced_features={"enable_rtm": True, "enable_tools": True},
    parameters={"data_channel": "rtm", "enable_error_message": True},
).with_stt(FengmingSTT()).with_llm(llm).with_tts(tts)

这里有两个字符串长得几乎一样，但不能混用。FastMCP 的运行传输写作 streamable-http（中划线），而 mcp_servers 配置里要写 streamable_http（下划线）。写错了不会报明显的错，只会表现为「模型死活不调工具」。

五. 应该看到什么

分别说三句话试试：「现在几点」「查一下订单 SW1001」「查一下订单 SW9999」（不存在的单号）。

正确的表现是：前两句智能体真去调了工具再播报结果；第三句返回一个用户能听懂的错误说明，而不是硬编一个订单状态出来。

六. 故障排查

• 模型只回答不调工具：先查 enable_tools 有没有开、MCP 工具发现是否成功，再把「遇到订单问题必须调用工具」这类要求明确写进 system prompt。
• MCP 连接失败：从公网环境验证 DNS、TLS 和 /mcp 路径，光在本机测通不算数。
• MCP 返回 401：mcp_servers.headers.Authorization 和服务端的 MCP_SHARED_SECRET 要完全一致。
• 工具返回了敏感信息：内部字段不要直接抛出去。权限校验、脱敏、审计都应该在 MCP 服务内部做完。
• 工具超时：返回一段能播报的错误摘要，并在业务层设好超时和重试上限。别让用户对着空气等。

七. 下一步

工具能调了，接着要管的是「什么时候不该让它调」，按场景收窄工具集。

《动态工具集：按场景切换语音智能体的可用函数》