实战踩坑：用 uv + FastMCP + vLLM 搭建生产级大模型流式网关

12次阅读

共计 4379 个字符，预计需要花费 11 分钟才能阅读完成。

最近团队在把内部的多 Agent 调度系统与本地私有化部署的开源大模型（基于 vLLM）打通，中间要过一层统一的 API 路由和 MCP（Model Context Protocol）服务。上线第一天就在线上吃了大亏：几千个流式 SSE 请求一进来，Python 网关内存一路狂飙到 OOM，客户端断开连接后后端推理卡依然在吭哧吭哧空跑吐 Token，吞吐断崖式下跌。

折腾了两天，彻底重构了依赖链和协程生命周期管理。今天这篇文章把整套架构方案全盘托出：如何彻底抛弃臃肿缓慢的 Poetry/Conda 换上 Rust 驱动的 uv，如何用 httpx.AsyncClient 优雅处理高并发 SSE 流与客户端取消背压，以及如何用 FastMCP 构建生产级 Agent 工具调用通道。

很多同学在写大模型代理网关时，直觉上觉得用 FastAPI + StreamingResponse 包一层就行：

# 典型的错误示范：死循环与连接悬挂陷阱
@app.post("/v1/chat/completions")
async def chat_proxy(request: Request):
    async def event_generator():
        client = httpx.AsyncClient()
        async with client.stream("POST", upstream_url, json=payload) as response:
            async for chunk in response.aiter_text():
                yield chunk
    return StreamingResponse(event_generator(), media_type="text/event-stream")

在本地跑一两个请求没毛病，但在真实并发流量下，这套写法有三个致命硬伤：

客户端断连感知失效（Client Disconnect Leak）： 当用户关闭网页或 App 点击“停止生成”，前端 TCP 挥手断开，而 event_generator 依然在后台消费 upstream 的 SSE 流，vLLM 算力被白白浪费。
HTTP 连接池滥用与耗尽： 每个请求里 httpx.AsyncClient() 实例化一次，不仅无法复用底层 TCP/TLS 握手，还会瞬间把宿主机 TIME_WAIT 状态的端口撑满。
流控背压（Backpressure）缺失： upstream 吐 Token 太快，客户端网络差消费慢，中间网关的缓冲区就会无限吃内存。

搞 Python AI 工程的都知道环境管理有多折磨。现在新项目我全部统一用 Astral 家的 uv，依赖解析速度比 pip 快几十倍，而且锁定文件跨平台丝滑无比。

# 安装并初始化现代 Python 项目
curl -LsSf https://astral.sh/uv/install.sh | sh
uv init llm-mcp-gateway --python 3.12
cd llm-mcp-gateway

# 一键添加高性能依赖链
uv add "fastapi[standard]" httpx mcp orjson uvloop

注意这里锁定了 Python 3.12，并引入了 uvloop 和 orjson，这是把 FastAPI 吞吐压榨到极限的基石。

话不多说，直接上核心网关代码。我们通过监听 request.is_disconnected()，在客户端单方面切断连接时立即取消 upstream 请求，并结合单例 httpx.AsyncClient 复用连接池：

import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import httpx
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse

# 建议在服务生命周期管理单例客户端
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 配置长连接池参数
    limits = httpx.Limits(max_keepalive_connections=500, max_connections=1000)
    timeout = httpx.Timeout(connect=5.0, read=120.0, write=10.0, pool=10.0)
    app.state.http_client = httpx.AsyncClient(limits=limits, timeout=timeout)
    yield
    await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)
UPSTREAM_VLLM_URL = "http://10.0.0.101:8000/v1/chat/completions"

async def stream_with_cancellation(
    request: Request,
    payload: dict
) -> AsyncGenerator[bytes, None]:
    client: httpx.AsyncClient = request.app.state.http_client

    try:
        async with client.stream(
            "POST",
            UPSTREAM_VLLM_URL,
            json=payload,
            headers={"Authorization": "Bearer local-token"}
        ) as upstream_resp:
            if upstream_resp.status_code != 200:
                error_body = await upstream_resp.aread()
                yield f"data: {{\"error\": \"Upstream error: {upstream_resp.status_code}\"}}\n\n".encode()
                return

            # 边读边推，实时监控客户端网络状态
            async for raw_chunk in upstream_resp.aiter_raw():
                # 关键排障点：检测客户端是否断开
                if await request.is_disconnected():
                    # 抛出 CancelledError 或直接退出，触发上下文管理器断开 upstream
                    break
                yield raw_chunk

    except asyncio.CancelledError:
        # 当 FastAPI 判定请求任务被取消时触发
        pass
    except httpx.RequestError as exc:
        yield f"data: {{\"error\": \"Transport failure: {str(exc)}\"}}\n\n".encode()

@app.post("/v1/chat/completions")
async def chat_proxy_endpoint(request: Request):
    payload = await request.json()
    # 强制流式标记
    payload["stream"] = True

    return StreamingResponse(stream_with_cancellation(request, payload),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no" # 极其重要！禁止 Nginx 缓冲区攒数据
        }
    )

避坑细节： 代码里的 X-Accel-Buffering: no 救过无数老哥的命。如果前面套了 Nginx 反向代理，Nginx 默认会把几十 K 缓冲区填满才往下推数据，导致原本一个字一个字蹦的流式打字机效果直接变成“卡死 5 秒然后啪一下全出来”。加上这个响应头可以强行关掉 Nginx 缓冲。

现在 Claude 引领的 MCP 协议已经成为 Agent 领域连接外部工具的标准。我们直接基于 mcp 官方 SDK 在同一套网关内挂载 MCP Server，让 Agent 能够动态发现内部工具：

from mcp.server.fastmcp import FastMCP

# 初始化 MCP 实例
mcp = FastMCP("Ops-Agent-Toolbox")

@mcp.tool()
async def query_cluster_metrics(node_ip: str) -> str:
    """查询指定内网计算节点的 GPU 与显存状态"""
    # 真实业务下调用 Prometheus 或节点 Agent
    return f"Node {node_ip}: GPU 0 - 82C, VRAM 92% (Allocated), Active Contexts: 14"

@mcp.tool()
async def restart_vllm_engine(node_ip: str) -> str:
    """紧急重启指定节点的推理服务"""
    return f"Restart signal sent to {node_ip}."

# FastMCP 支持直接挂载为 ASGI 路由，或者以独立 stdio/SSE 协议进程运行

我们在单台 32 核 64G 内存、千兆内网的服务器上，使用 locust 对 vLLM 后端直连与该代理网关进行了对比压测（模型：Qwen2.5-72B-Instruct-AWQ，模拟 300 持续并发用户，平均输入 500 Tokens，输出 300 Tokens）：

指标项	直连 vLLM (8000 端口)	原始 FastAPI 代理方案	uv+ 连接池优化网关
首字延迟 (TTFT P95)	380 ms	620 ms	392 ms
网关平均内存占用	–	4.2 GB (线性上涨)	280 MB (恒定)
异常中断卡顿率	0%	28% (连接泄漏卡死)	< 0.1%
每秒吞吐 (Tokens/s)	2450	1620	2410 (损耗 < 2%)

从实测数据能清晰看到，正确的连接池复用 + 断连检测，把代理层带来的网络额外开销压缩到了 15ms 以内，且彻底解决了长时间运行内存泄漏的问题。

在处理异步大模型流式代理时，千万别把 asyncio 当成银弹。高并发网络 I/O 场景下，核心关键只有三点： 严格管理 HTTPClient 连接池生命周期、必须主动监听断连以实现背压、基础设施层必须关闭反向代理的缓存机制 。采用 uv + FastAPI + uvloop 这套轻量利落的组合，比起动辄上重量级网关框架（如 Kong、Envoy 二开），不仅排障简单、代码一目了然，性能也完全足够扛起中大型团队的日常推理吞吐。

正文完