AI Agent 技能分享｜MCP Server 查询慢怎么办：连接池、超时、缓存与熔断

MCP Server 在本机演示时，查询通常很快：一个客户端、几条测试数据、数据库也在同一台机器上。服务放到线上以后，Agent 可能为了回答一个问题连续调用多个 Tool，多个用户又会同时发起会话，原来几十毫秒的查询很快变成排队、超时和重复重试。

更麻烦的是，Agent 看到 Tool 超时后，通常会尝试换个说法或者再次调用。如果服务端没有幂等、限流和熔断，数据库越慢，重试越多；重试越多，数据库又变得更慢。

这篇不谈“换更大的数据库”这种结论，直接在 MCP Server 里处理几个最常见的问题：

连接池       避免每次查询重新建连接
超时         让慢请求尽快释放资源
结果上限     不把几十万行送回模型
缓存         减少重复读
限流与舱壁   控制并发数量
熔断         下游故障时快速失败
指标         看清时间到底耗在哪里

先把慢拆开

用户只会说“Agent 回答很慢”，但一次 Tool 调用至少包含这些时间：

排队等待连接
建立或检查连接
数据库执行 SQL
读取结果集
序列化 JSON
网络传输
模型处理 Tool 结果

如果只记录整个 Tool 的耗时，定位不出是连接池耗尽、SQL 本身慢，还是返回数据太大。

我会给每次调用生成 trace_id，分别记录：

from dataclasses import dataclass


@dataclass
class ToolTiming:
    pool_wait_ms: float = 0
    query_ms: float = 0
    serialize_ms: float = 0
    total_ms: float = 0
    rows: int = 0
    response_bytes: int = 0

先有数据，再决定加缓存还是改 SQL。没有指标时，最容易出现“缓存加了一堆，真正慢的是连接池一直在等”的情况。

连接池不是越大越好

SQLAlchemy 创建 SQL Server 连接池：

from sqlalchemy import create_engine


engine = create_engine(
    database_url,
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=5,
    pool_timeout=3,
    pool_recycle=1800,
)

这些参数分别控制：

• pool_size=10：长期保留的连接数量；
• max_overflow=5：高峰期额外允许的临时连接；
• pool_timeout=3：等不到连接时最多等待三秒；
• pool_pre_ping=True：借出前检查连接是否仍可用；
• pool_recycle=1800：定期回收长时间存活的连接。

连接池大小不能只看 MCP Server 进程。如果有四个 Worker，每个 Worker 的最大连接数是 15，理论上就可能占用 60 条数据库连接。

最大连接数 ≈ Worker 数 × (pool_size + max_overflow)

池子开得太小，请求都在应用层排队；开得太大，数据库同时执行的查询过多，同样会变慢。最终数值要结合数据库连接上限、查询耗时和真实并发压测确定。

同时设置连接等待和 SQL 执行超时

只有 HTTP 超时不够。客户端十秒后放弃，但数据库中的 SQL 可能还在继续执行。

pyodbc 可以给连接设置查询超时：

from sqlalchemy import event


@event.listens_for(engine, "connect")
def configure_connection(dbapi_connection, _connection_record) -> None:
    dbapi_connection.timeout = 5

Tool 自己也要有总时间预算：

import time


class Deadline:
    def __init__(self, timeout_seconds: float):
        self.ends_at = time.monotonic() + timeout_seconds

    def remaining(self) -> float:
        return max(0.0, self.ends_at - time.monotonic())

    def ensure_available(self) -> None:
        if self.remaining() <= 0:
            raise TimeoutError("Tool 调用已超过时间预算")

假设整个 Tool 预算八秒，可以这样分：

等待连接最多 2 秒
数据库执行最多 5 秒
序列化和网络预留 1 秒

不要让每个环节都单独等八秒，否则总耗时会远超客户端预期。

返回行数和返回字节数都要限制

只加 TOP 100 还不够。如果某个字段包含大段备注或文档，100 行仍然可能有几 MB。

import json


MAX_ROWS = 100
MAX_RESPONSE_BYTES = 256 * 1024


def bounded_result(items: list[dict]) -> dict:
    selected: list[dict] = []
    current_bytes = 2

    for item in items[:MAX_ROWS]:
        encoded = json.dumps(item, ensure_ascii=False, default=str).encode("utf-8")
        if current_bytes + len(encoded) > MAX_RESPONSE_BYTES:
            break
        selected.append(item)
        current_bytes += len(encoded)

    return {
        "items": selected,
        "count": len(selected),
        "truncated": len(selected) < len(items),
        "response_bytes": current_bytes,
    }

更好的做法是在 SQL 层就只选择需要的字段，并通过游标分页，不要先把十万行读进 Python 再截断。

结果被截断时要明确返回 truncated=true，避免 Agent 把前 100 条说成全部结果。

分页使用稳定游标

大结果查询不要使用越来越深的 OFFSET：

ORDER BY CreatedAt DESC
OFFSET 100000 ROWS FETCH NEXT 50 ROWS ONLY;

可以使用稳定排序字段做 Keyset Pagination：

SELECT TOP (:limit)
    OrderId,
    OrderNo,
    OrderStatus,
    CreatedAt
FROM dbo.v_AgentOrderSummary
WHERE TenantId = :tenant_id
  AND
  (
      :cursor_created_at IS NULL
      OR CreatedAt < :cursor_created_at
      OR (CreatedAt = :cursor_created_at AND OrderId < :cursor_order_id)
  )
ORDER BY CreatedAt DESC, OrderId DESC;

Tool 返回下一页游标：

{
  "items": [],
  "next_cursor": {
    "created_at": "2026-08-14T09:30:00",
    "order_id": 10081
  },
  "has_more": true
}

游标也要绑定租户、过滤条件和有效期。不要接受客户端随意修改游标里的租户字段。

缓存只适合可接受短暂旧数据的查询

订单列表、数据字典和统计指标通常可以缓存几秒到几分钟；支付状态、审批状态和写操作结果则不一定适合。

先定义缓存 Key：

import hashlib
import json


def cache_key(tenant_id: str, tool_name: str, params: dict) -> str:
    canonical = json.dumps(
        params,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )
    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return f"mcp:{tenant_id}:{tool_name}:{digest}"

必须包含租户，最好也包含数据版本和权限相关维度。否则两个租户使用相同参数时可能命中同一份结果。

缓存包装器：

import json


def cached_query(
    redis_client,
    key: str,
    ttl_seconds: int,
    loader,
):
    cached = redis_client.get(key)
    if cached is not None:
        return json.loads(cached), True

    result = loader()
    redis_client.setex(
        key,
        ttl_seconds,
        json.dumps(result, ensure_ascii=False, default=str),
    )
    return result, False

返回值可以带上 cached 和 data_as_of，让 Agent 知道数据新鲜度：

{
  "cached": true,
  "data_as_of": "2026-08-14T10:20:00+08:00",
  "items": []
}

防止缓存击穿

一个热门查询刚好过期，几十个请求同时落到数据库，会把缓存的收益抵消掉。

可以使用单航班锁：同一个 Key 只允许一个请求回源，其他请求短暂等待或读取旧值。

def load_with_lock(redis_client, key: str, ttl: int, loader):
    lock = redis_client.lock(
        f"lock:{key}",
        timeout=10,
        blocking_timeout=1,
    )

    if not lock.acquire(blocking=True):
        stale = redis_client.get(f"stale:{key}")
        if stale is not None:
            return json.loads(stale), True
        raise RuntimeError("查询繁忙，请稍后重试")

    try:
        cached = redis_client.get(key)
        if cached is not None:
            return json.loads(cached), True

        result = loader()
        payload = json.dumps(result, ensure_ascii=False, default=str)
        redis_client.setex(key, ttl, payload)
        redis_client.setex(f"stale:{key}", ttl * 5, payload)
        return result, False
    finally:
        lock.release()

锁超时、Redis 故障和旧值策略需要按业务确定。不能因为缓存挂了就让所有请求无限制打到数据库。

用信号量做舱壁隔离

数据库能承受 50 个并发查询，不代表单个 MCP Tool 可以全部占满。报表 Tool 如果耗尽连接，会连简单的订单状态查询也一起拖慢。

import asyncio


order_query_slots = asyncio.Semaphore(20)
report_query_slots = asyncio.Semaphore(4)


async def run_report(loader):
    try:
        await asyncio.wait_for(report_query_slots.acquire(), timeout=0.5)
    except TimeoutError as exc:
        raise RuntimeError("报表查询繁忙，请稍后重试") from exc

    try:
        return await loader()
    finally:
        report_query_slots.release()

这种做法叫 Bulkhead，意思是把不同类型的负载隔开。高成本报表最多占四个并发槽位，不能拖垮所有普通查询。

如果 Tool 是同步函数，可以使用线程信号量或在网关层做并发限制。

熔断不是遇到一次异常就关闭服务

当数据库连续超时，继续接收全部请求只会制造更多排队。熔断器可以在失败达到阈值后短时间快速失败，给数据库恢复机会。

import time


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_seconds: int = 20):
        self.failure_threshold = failure_threshold
        self.reset_seconds = reset_seconds
        self.failures = 0
        self.opened_at: float | None = None

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if time.monotonic() - self.opened_at >= self.reset_seconds:
            self.opened_at = None
            self.failures = 0
            return True
        return False

    def success(self) -> None:
        self.failures = 0
        self.opened_at = None

    def failure(self) -> None:
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.opened_at = time.monotonic()

示例只表达状态逻辑。多实例部署时，熔断状态可以放在实例本地，让每个实例独立保护自己；如果要共享状态，需要考虑 Redis 故障时的行为。

业务错误不能计入熔断。例如“订单不存在”和“参数不合法”不是数据库故障。一般只统计连接失败、超时和明确的下游不可用错误。

重试要有条件

只读查询可以对短暂连接错误重试一次，写操作不能在不知道执行结果时随便重试。

import random
import time


def retry_read(loader, attempts: int = 2):
    last_error = None

    for attempt in range(attempts):
        try:
            return loader()
        except TransientDatabaseError as exc:
            last_error = exc
            if attempt + 1 >= attempts:
                break
            delay = 0.1 * (2 ** attempt) + random.uniform(0, 0.05)
            time.sleep(delay)

    raise last_error

不要重试参数错误、权限错误和确定的业务拒绝。写操作只有在具备稳定幂等键，并能查询最终状态时才考虑重试。

一个带指标的查询 Tool

import json
import time
import uuid


@mcp.tool()
def get_recent_orders(limit: int = 20) -> dict:
    identity = get_verified_identity()
    require_scope(identity, "order.read")

    trace_id = str(uuid.uuid4())
    started_at = time.perf_counter()

    params = {"limit": min(max(limit, 1), 50)}
    key = cache_key(identity.tenant_id, "get_recent_orders", params)

    def load():
        query_started = time.perf_counter()
        rows = repository.get_recent_orders(
            tenant_id=identity.tenant_id,
            limit=params["limit"],
        )
        query_ms = (time.perf_counter() - query_started) * 1000
        return {
            "items": rows,
            "query_ms": round(query_ms, 2),
        }

    result, cache_hit = cached_query(redis_client, key, 30, load)
    response = bounded_result(result["items"])
    response["cache_hit"] = cache_hit
    response["trace_id"] = trace_id

    total_ms = (time.perf_counter() - started_at) * 1000
    logger.info(
        "trace_id=%s tool=get_recent_orders tenant=%s cache_hit=%s rows=%s bytes=%s total_ms=%.2f",
        trace_id,
        identity.tenant_id,
        cache_hit,
        response["count"],
        response["response_bytes"],
        total_ms,
    )

    return response

响应里保留 trace_id，用户报告慢请求时可以直接定位日志。详细数据库错误和连接信息仍然只进入服务端。

监控哪些指标

上线后至少观察：

指标	用途
Tool 调用次数	判断热点能力
P50/P95/P99 总耗时	观察长尾延迟
连接池等待时间	判断池是否耗尽
SQL 执行时间	定位慢查询
超时率	判断时间预算是否合理
缓存命中率	判断缓存是否有效
返回行数和字节数	防止上下文膨胀
熔断次数	判断下游是否不稳定
每租户并发数	发现单租户抢占资源

日志里有一次请求的细节，指标里有整体趋势，两者都需要。只看平均耗时通常会掩盖少量特别慢的请求。

压测要模拟 Agent 的调用方式

普通接口压测可能每个用户只调用一次，但 Agent 为回答一个问题会连续调用 search_orders、get_order_status 和 get_refund_rule。压测脚本应该模拟这种调用链，并加入超时后的重试。

我会至少测三组：

稳定负载：持续并发查询 10 分钟
突发负载：短时间放大到正常流量的 5 倍
故障负载：人为增加数据库延迟，观察超时、重试和熔断

重点看数据库变慢时，MCP Server 能不能快速失败，而不是把所有请求和连接一起堆住。

最后

MCP Server 查询慢，通常不是某一个参数能解决的。连接池控制数据库连接数量，超时控制单次占用时间，缓存减少重复读取，舱壁限制高成本 Tool 的并发，熔断在下游故障时阻止请求继续堆积。

这些措施有一个共同前提：必须知道时间花在哪里。先把连接等待、SQL 执行、返回行数、响应字节和总耗时记录清楚，再针对真正的瓶颈动手。

对 Agent 来说，快速而明确地返回“系统繁忙，请稍后重试”，通常比等待三十秒后抛出一个数据库异常更容易处理，也更不容易触发失控的重复调用。