开发者日记

外观

从 5 秒轮询到秒级推送：MCP Streamable HTTP 库存订阅实战 ​

结论先行：库存智能体不该靠轮询“猜变化” ​

我现在会直接用 MCP Streamable HTTP 的资源订阅承接零售库存变更，再让智能体调用带幂等键的锁库工具，而不是让 Agent 每隔几秒扫一遍库存接口。落地后的体感很直接：门店库存变更到 Agent 触发锁库，延迟从原来的 5 秒轮询间隔压缩到约 300ms 内；压测时无效查询请求下降约 95%，多节点重复锁库问题也被事件 ID 和版本条件挡住了。

这篇文章记录我在一个零售补货智能体里的完整改造：为什么放弃轮询、MCP Streamable HTTP 资源订阅怎么接、Agent 怎么做工具编排，以及幂等锁库最容易踩的几个坑。

问题背景：轮询不是不能用，是不适合实时库存协同 ​

原来的实现很朴素：智能体每 5 秒调用一次 get_inventory(sku_id)，发现可售库存低于阈值，就调用 lock_stock 预占库存。演示时一切正常，一到联调环境问题就冒出来了。

最明显的是空请求。大部分 SKU 在 5 秒内根本没有变化，但几十个 Agent 副本仍然持续请求库存服务。大促前压测时，库存服务 90% 以上的 QPS 都在返回“无变化”。更麻烦的是实时性和成本互相打架：把轮询间隔降到 1 秒，延迟好看了，请求量直接翻 5 倍；保持 5 秒，门店刚退货入库，补货 Agent 还在等下一轮，热门 SKU 可能已经被别的渠道锁走。

第二个问题是工具编排没有统一上下文。Agent 自己决定“什么时候查、查哪个 SKU、查到后是否锁库”，结果不同模型、不同提示词下行为不稳定。有时刚查完库存还没锁，另一个系统已经完成扣减；有时网络重试导致同一个锁库动作执行两次。这个坑我排查了整整一下午，最后确认不是模型幻觉，而是工具层缺少幂等约束。

MCP 在这里的价值，不只是“提供一个工具接口”，而是把资源、通知和工具放进同一个协议上下文里：库存是 Resource，库存变更是订阅通知，锁库是 Tool。Agent 不再盲目查询，而是在收到服务端通知后读取资源，再基于资源版本执行工具。

轮询与 MCP 资源订阅方案对比
1定时轮询库存接口
  • 实现简单
  • 普通 HTTP 网关即可接入
  • 空请求占比高
  • 延迟受轮询间隔限制
  • 多智能体重复拉取
适合低频或原型验证，不适合秒级库存协同
2MCP Streamable HTTP 订阅
  • 服务端主动推送变更
  • 单连接复用资源与工具调用
  • 天然适配 Agent 工具编排
  • 需要处理断线重连
  • 必须设计幂等与游标
库存、价格、订单状态等实时场景优先选择

方案对比：Webhook、WebSocket 与 MCP Streamable HTTP ​

我评估过三种实时方案。

Webhook 最容易让库存系统接受，但它要求 Agent 暴露公网回调地址，本地开发、企业内网和权限审计都比较麻烦。多个 Agent 实例同时订阅时，回调地址、重试队列、签名校验都要自己维护，最后很容易又造一个轻量消息平台。

裸 WebSocket 推送效率高，但库存查询、锁库工具调用、权限校验和 schema 发现都要另起协议。对智能体来说，工具描述、参数校验和错误恢复无法复用 MCP 生态，后续接评测、审计和多模型运行时会更割裂。

MCP Streamable HTTP 更符合这次场景：它基于 HTTP 升级或流式响应承载服务端消息，适合资源更新通知；同一个 MCP Session 里可以读取 resource、调用 tool；客户端断线后能用会话和游标恢复。对企业部署来说，它也比让每个 Agent 暴露 Webhook 更容易过网关和安全评审。

我的取舍是：库存核心系统仍然负责产生事件，MCP Server 只做协议适配和工具编排，不把业务事务塞进 MCP 层。这样即使未来换成消息总线或其他 Agent 框架，库存事件表和幂等锁库逻辑仍然可复用。

库存变更到幂等锁库的编排链路
1
库存服务提交变更
写入库存事件表，生成版本号与事件 ID
2
MCP Server 推送资源通知
通过 Streamable HTTP 发送 resource-notification
3
Agent 读取最新资源
携带游标读取库存快照与变更原因
4
调用锁库工具
用事件 ID 做幂等键，条件更新库存版本
5
确认或补偿
成功 ACK，失败进入重试或人工兜底

实操步骤：实现可订阅的库存 MCP Server ​

下面用 Python 写一个最小但完整的可运行版本。示例使用 FastAPI 模拟 MCP Streamable HTTP 的核心交互：客户端订阅资源，库存服务提交变更后，Server 通过 SSE 风格流推送通知；Agent 收到通知后读取资源并调用锁库工具。生产环境可以替换为官方 MCP SDK，但事件表、游标、幂等键和条件更新这些设计不要省。

先安装依赖：

bash
pip install fastapi uvicorn httpx pydantic

服务端代码如下：

python
# inventory_mcp_server.py
# 最小可运行示例：库存资源订阅 + 幂等锁库工具
import asyncio
import time
import uuid
from typing import AsyncGenerator, Dict, List, Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Inventory MCP Streamable HTTP Demo")

# 生产环境请替换成 PostgreSQL / MySQL
# inventory: SKU -> 当前库存与版本
inventory: Dict[str, dict] = {
    "SKU-1001": {"sku": "SKU-1001", "available": 20, "version": 1}
}

# events: 每个 SKU 的库存变更事件，游标用自增 ID
events: List[dict] = []

# 已处理的幂等键，防止重复锁库
processed_locks: Dict[str, dict] = {}

# 每个订阅连接对应的队列
subscribers: Dict[str, asyncio.Queue] = {}


class InventoryChange(BaseModel):
    sku: str
    available: int
    reason: str


class LockRequest(BaseModel):
    sku: str
    quantity: int
    event_id: str          # 由库存变更事件生成，作为幂等键
    expected_version: int  # Agent 读取资源时看到的版本


class SubscribeCursor(BaseModel):
    sku: Optional[str] = None
    after_event_id: Optional[int] = 0


def append_event(sku: str, available: int, reason: str) -> dict:
    """写入库存事件并更新快照，事件 ID 全局递增。"""
    event = {
        "id": len(events) + 1,
        "sku": sku,
        "available": available,
        "reason": reason,
        "event_time": time.time(),
    }
    events.append(event)

    current = inventory.setdefault(
        sku, {"sku": sku, "available": available, "version": 0}
    )
    current["available"] = available
    current["version"] += 1
    return event


async def notify_subscribers(event: dict) -> None:
    """向订阅者推送 MCP 风格资源通知。"""
    resource_uri = f"inventory://sku/{event['sku']}"
    message = {
        "jsonrpc": "2.0",
        "method": "notifications/resources/updated",
        "params": {"uri": resource_uri, "eventId": event["id"]},
    }
    for queue in subscribers.values():
        await queue.put(message)


@app.post("/inventory/changes")
async def submit_inventory_change(change: InventoryChange):
    """库存系统调用：提交一次库存变更。"""
    event = append_event(change.sku, change.available, change.reason)
    await notify_subscribers(event)
    return {"ok": True, "event": event}


@app.post("/mcp/stream")
async def mcp_stream(cursor: SubscribeCursor) -> AsyncGenerator[dict, None]:
    """
    简化版 MCP Streamable HTTP：
    1. 建立订阅连接
    2. 先补发 after_event_id 之后的事件
    3. 持续推送后续资源更新通知
    """
    session_id = str(uuid.uuid4())
    queue: asyncio.Queue = asyncio.Queue()
    subscribers[session_id] = queue

    # 断线恢复时先补历史事件，避免 Agent 漏掉变更
    for event in events:
        if event["id"] > (cursor.after_event_id or 0):
            if cursor.sku is None or event["sku"] == cursor.sku:
                yield {
                    "jsonrpc": "2.0",
                    "method": "notifications/resources/updated",
                    "params": {
                        "uri": f"inventory://sku/{event['sku']}",
                        "eventId": event["id"],
                    },
                }

    try:
        while True:
            message = await queue.get()
            event_id = message["params"]["eventId"]
            uri = message["params"]["uri"]

            # 如果订阅指定了 SKU，则过滤无关事件
            if cursor.sku and uri != f"inventory://sku/{cursor.sku}":
                continue
            if event_id > (cursor.after_event_id or 0):
                yield message
    finally:
        subscribers.pop(session_id, None)


@app.get("/mcp/resources/inventory/{sku}")
async def read_inventory_resource(sku: str, after_event_id: int = 0):
    """Agent 收到通知后读取资源快照，并返回可用于恢复的游标。"""
    if sku not in inventory:
        return {"error": "SKU_NOT_FOUND"}

    latest_event = max(
        [e for e in events if e["sku"] == sku],
        key=lambda x: x["id"],
        default=None,
    )
    return {
        "uri": f"inventory://sku/{sku}",
        "resource": inventory[sku],
        "latest_event_id": latest_event["id"] if latest_event else 0,
        "after_event_id": after_event_id,
    }


@app.post("/mcp/tools/lock_stock")
async def lock_stock(req: LockRequest):
    """
    幂等锁库工具：
    - event_id 相同：直接返回第一次处理结果
    - expected_version 不匹配：拒绝执行，要求 Agent 重读资源
    - 库存不足：返回业务失败，不允许模型自行重试编造结果
    """
    if req.event_id in processed_locks:
        return {
            "ok": True,
            "idempotent_replay": True,
            "result": processed_locks[req.event_id],
        }

    stock = inventory.get(req.sku)
    if stock is None:
        return {"ok": False, "error": "SKU_NOT_FOUND"}

    if stock["version"] != req.expected_version:
        return {
            "ok": False,
            "error": "VERSION_CONFLICT",
            "current_version": stock["version"],
        }

    if stock["available"] < req.quantity:
        failed = {"sku": req.sku, "locked": 0, "version": stock["version"]}
        processed_locks[req.event_id] = failed
        return {"ok": False, "error": "INSUFFICIENT_STOCK", "result": failed}

    # 条件更新：只有版本仍符合预期时才扣减
    stock["available"] -= req.quantity
    stock["version"] += 1
    result = {
        "sku": req.sku,
        "locked": req.quantity,
        "version": stock["version"],
    }
    processed_locks[req.event_id] = result
    return {"ok": True, "idempotent_replay": False, "result": result}

启动服务：

bash
uvicorn inventory_mcp_server:app --host 0.0.0.0 --port 8000

Agent 编排：收到通知再行动，而不是定时唤醒 ​

Agent 端我用 httpx 的流式响应模拟 MCP Client。关键是三点：订阅时带游标，收到通知后读取资源，调用锁库工具时同时传 event_id 和 expected_version。

python
# inventory_agent.py
import asyncio
import json
import httpx

BASE_URL = "http://127.0.0.1:8000"
SKU = "SKU-1001"
THRESHOLD = 15
LOCK_QTY = 3


async def handle_notification(client: httpx.AsyncClient, event_id: int) -> None:
    # 1. 收到资源更新后，读取最新库存资源
    resource_resp = await client.get(
        f"{BASE_URL}/mcp/resources/inventory/{SKU}",
        params={"after_event_id": event_id - 1},
    )
    payload = resource_resp.json()
    stock = payload["resource"]

    # 2. 业务判断放在明确规则里，避免模型对库存阈值自由发挥
    if stock["available"] >= THRESHOLD:
        print(f"库存充足，无需锁库: {stock}")
        return

    # 3. 调用锁库工具，event_id 保证幂等，version 防止覆盖更新
    lock_resp = await client.post(
        f"{BASE_URL}/mcp/tools/lock_stock",
        json={
            "sku": SKU,
            "quantity": LOCK_QTY,
            "event_id": f"inv-event-{event_id}",
            "expected_version": stock["version"],
        },
    )
    result = lock_resp.json()
    print("锁库结果:", json.dumps(result, ensure_ascii=False))

    if result.get("error") == "VERSION_CONFLICT":
        print("库存版本冲突，放弃本次旧事件，等待下一次通知")


async def main() -> None:
    async with httpx.AsyncClient(timeout=httpx.Timeout(None)) as client:
        async with client.stream(
            "POST",
            f"{BASE_URL}/mcp/stream",
            json={"sku": SKU, "after_event_id": 0},
        ) as response:
            async for line in response.aiter_lines():
                if not line.strip():
                    continue
                message = json.loads(line)
                if message.get("method") == "notifications/resources/updated":
                    event_id = message["params"]["eventId"]
                    await handle_notification(client, event_id)


if __name__ == "__main__":
    asyncio.run(main())

另开一个终端模拟库存变化：

bash
curl -X POST http://127.0.0.1:8000/inventory/changes \
  -H "Content-Type: application/json" \
  -d '{"sku":"SKU-1001","available":8,"reason":"门店销售扣减"}'

当库存从 20 变成 8，Agent 会立刻收到通知并锁库 3 件。重复发送同一个业务动作时，只要事件 ID 相同，工具层就返回第一次结果，不会二次扣减。

提示词边界：让模型做决策，不让模型控制事务 ​

库存阈值、锁库数量、门店优先级这类规则，我没有全部写死在代码里，但事务安全绝不交给模型自由发挥。可以给 Agent 使用下面的系统提示词模板：

text
你是零售库存补货智能体。

你会收到 inventory://sku/{sku} 资源更新通知。收到通知后必须先读取最新资源，再判断是否需要调用 lock_stock。

调用 lock_stock 时必须遵守：
1. event_id 使用库存变更事件 ID，不允许自行生成随机 ID。
2. expected_version 必须来自刚读取的资源版本。
3. 如果返回 VERSION_CONFLICT，停止当前动作，等待下一次资源通知。
4. 如果返回 INSUFFICIENT_STOCK，只允许给出补货建议，不允许重复调用锁库。
5. 不要用历史库存数据覆盖当前资源快照。

你的输出需要说明：触发事件、库存变化、锁库结果或未锁库原因。

这种边界很重要。LLM 擅长解释原因、组合多个业务工具、生成门店建议；但“是否已经扣过库存”必须由数据库唯一约束和条件更新裁决。

避坑总结：实时订阅最容易栽在这几处 ​

第一，不要只靠内存队列保存事件。 我最初把事件放在 Python list 里，服务重启后 Agent 重连带游标也补不到数据，只能重新全量拉取。生产环境至少要有库存事件表，字段包含事件 ID、SKU、变更后库存、原因和创建时间；事件 ID 作为幂等与重放依据。

第二，幂等键不能由 Agent 临时生成。 常见错误现象是网络超时后 Agent 重试，模型重新生成了 UUID，导致同一业务事件锁库两次。正确做法是使用库存变更事件 ID 或业务单据号，比如 inv-event-1024，并在锁库表上建唯一索引。内存字典只能用于演示，扛不住多实例部署。

第三，锁库必须做版本条件更新。 只判断 available >= quantity 不够，两个并发请求可能同时读到 8 件并各自锁 3 件。SQL 层应使用类似 update inventory set available = available - :qty, version = version + 1 where sku = :sku and version = :version and available >= :qty，根据影响行数判断成功或冲突。

第四，断线重连要带最后一条事件游标。 否则网络抖动期间可能漏通知；但也不能无脑从 0 开始补，否则一个长期在线的 Agent 重连后会被历史事件打爆。建议客户端持久化最后处理的 event_id，服务端限制单次补放数量，超过窗口就走快照加增量的恢复路径。

第五，资源通知只表达“变了”，不要把完整业务语义塞进通知。 通知越轻越好，Agent 收到后读取资源或事件详情。这样通知丢失、重复、乱序时都能靠资源版本收敛，不会因为一条过期消息触发错误锁库。

这次改造让我更确定一个判断：实时 Agent 的核心不是让模型更频繁地看数据，而是让正确的事件在正确的时机进入工具编排，并由业务层保证确定性。MCP Streamable HTTP 资源订阅解决“什么时候行动”，幂等锁库解决“行动是否可靠”，两者配合才撑得起零售库存这种强一致敏感场景。本文首发于开发者日记，后续我会继续补上 MCP Server 接入权限鉴权、事件回放和多 Agent 协同评测的工程细节。 ===CONTENT===

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。