在 OncallAgent 中，MCP 不是一个只负责展示“可用工具列表”的装饰层，而是聊天 Agent 和 AIOps 诊断访问外部系统的真实执行边界。用户在界面中保存连接，后端按当前用户读取启用项，向对应 MCP Server 发起初始化、工具发现和工具调用。连接检查失败时，系统把失败状态、空工具列表和安全错误保存到连接的 lastCheck；工具真正进入聊天或诊断生命周期后，调用成功与失败才会另行写入工具审计。两类记录不能混为一谈，系统也不会用模拟结果把流程伪装成成功。

📷 [图片 token=YsqSbjs6Ho7brGxkAnPcc0mbnVV（未能下载，见飞书原文）]

理解这条链路的关键，是把“连接配置”“工具发现”“运行时装配”“真实调用”和“调用审计”看成五个连续环节。任意一环缺失，Agent 都不应声称已经获得日志或完成外部操作。OncallAgent 当前支持 SSE 与 Streamable HTTP 两种 transport，并把腾讯云官方 CLS MCP Server 作为未配置用户连接时的项目默认入口。

学习目标

• 理解一条 MCP 连接如何从前端表单进入 SQLite，并被当前用户的 Agent 运行时加载。

• 区分“保存连接”“检查连接”“发现工具”和“调用工具”四种不同动作。

• 掌握多连接下的同名工具保护、超时重试和 owner 范围隔离。

• 能够沿源码和测试判断一次工具结果是否来自真实 MCP Server。

功能入口与完整调用链

用户从前端 /mcp 工作区创建或编辑连接。apps/frontend/src/stores/mcp.ts 中的 useMcpStore 负责页面状态，initialize、create、update、check 和 remove 分别对应列表、保存、连通性检查和删除动作。网络请求由 apps/frontend/src/mcp/mcpClient.ts 的 createMcpClient 发往 /mcp/connections 系列接口。

FastAPI 路由位于 apps/backend/src/super_ai/api/app.py。每个接口都先通过 _current_user 得到认证用户，再把 user.id 作为 owner_user_id 传给 McpConnectionService。服务层完成字段校验和业务判断，SQLite repository 则在查询、更新、删除和保存检查结果时再次应用 owner 条件。这样，连接 ID 即使被猜到，也不能脱离用户范围单独访问。

执行路径可以概括为：前端表单 → 共享请求契约 → FastAPI 认证路由 → McpConnectionService → owner-scoped SQLite 记录 → LocalMcpClient → MCP session 初始化 → list_tools 或 call_tool → SSE/HTTP 工具结果 → Agent 事件与 SQLite 审计。聊天和 AIOps 并不读取一个全局工具清单，而是通过 client_for_user 装配当前用户启用的连接。

📷 [图片 token=RhTvbVe4Jo7ehyxoP4XcNbudnNe（未能下载，见飞书原文）]

/mcp 页面
  → useMcpStore.check(connectionId)
  → POST /mcp/connections/{connectionId}:check
  → McpConnectionService.check(owner_user_id, connection_id)
  → LocalMcpClient.discover_tools()
  → ClientSession.initialize() + list_tools()
  → 保存 lastCheck 与真实 tools

聊天或诊断执行
  → McpConnectionService.client_for_user(owner_user_id)
  → 仅装配 enabled 连接
  → discover_tools() / call_tool()
  → 工具生命周期事件 + owner-scoped 审计

核心源码地图

📷 [图片 token=FhJDbDzJCoTTnZxjWOZcJAagnUh（未能下载，见飞书原文）]

代码调用流程图

MCP 链路要区分连接检查、聊天工具装配和 AIOps 直接调用。它们共享 owner-scoped 连接配置，但运行时调用方式并不完全相同。

关键实现拆解

连接管理不是简单的 URL 保存

McpConnectionService.create 和 update 都先调用 _validated_connection。名称不能为空且长度受限；transport 只能是 sse 或 streamable_http；URL 必须是 HTTP/HTTPS、必须有 hostname，并且禁止携带 username 或 password；超时限制在 1 到 300 秒，重试次数限制在 0 到 5。校验失败会产生 VALIDATION_INVALID_ARGUMENT，错误通过统一 API envelope 返回。

看什么：先看服务层在任何 Repository 写入之前如何一次性收紧名称、transport、URL、超时和重试范围；尤其注意 URL 的 userinfo 检查不是前端提示，而是后端保存边界。

    normalized_name = name.strip()
    normalized_url = url.strip()
    parsed = urlsplit(normalized_url)
    # 1. 名称和 transport 在落库前完成规范化与白名单校验。
    if not normalized_name or len(normalized_name) > 120:
        raise McpConnectionError("VALIDATION_INVALID_ARGUMENT", "MCP 连接名称无效。")
    if transport not in SUPPORTED_MCP_TRANSPORTS:
        raise McpConnectionError("VALIDATION_INVALID_ARGUMENT", "MCP transport 无效。")
    # 2. HTTP(S) URL 必须有 hostname，且禁止 username/password userinfo。
    if (
        parsed.scheme not in {"http", "https"}
        or not parsed.hostname
        or parsed.username
        or parsed.password
    ):
        raise McpConnectionError("VALIDATION_INVALID_ARGUMENT", "MCP URL 必须是安全的 HTTP 地址。")
    if not 1 <= timeout_seconds <= 300 or not 0 <= retries <= 5:
        raise McpConnectionError("VALIDATION_INVALID_ARGUMENT", "MCP 超时或重试参数无效。")

这段代码证明校验失败不会产生半条连接记录，且 transport 与重试策略不能由任意字符串扩展。它同时说明这里的“安全 URL”只落实到 scheme、hostname 与 userinfo；文档不能进一步声称已做内网地址阻断或完整 SSRF 防护。

📷 [图片 token=JXylbP3wvoZhALxsdQAceqpwn9f（未能下载，见飞书原文）]

list 还有一个容易忽略的默认行为：当前用户完全没有连接记录时，会为该用户创建一条“腾讯云 CLS”默认连接。只要用户已经存在记录，即使全部被禁用，也不会再次偷偷补一条启用连接。client_for_user 也遵守这个区别：没有任何记录时使用项目默认 CLS 地址；已有记录时只选取 enabled 为真的项。因而“禁用全部自定义连接”不会绕回默认连接。

看什么：把 list 的“首次可见默认记录”和 client_for_user 的“运行时回退”放在一起读，观察判断条件都是“是否存在任何记录”，而不是“是否存在启用记录”。

        records = await repository.list(owner_user_id=owner_user_id)
        if records:
            return records
        # 1. 仅当 owner 尚无任何记录时，列表入口创建默认连接。
        await repository.create(
            owner_user_id=owner_user_id,
            connection_id=f"mcp_{uuid4().hex}",
            name="腾讯云 CLS",
            transport="sse",
            url=self._default_url,
            enabled=True,
            timeout_seconds=self._default_timeout_seconds,
            retries=self._default_retries,
        )
        return await repository.list(owner_user_id=owner_user_id)

    async def client_for_user(self, *, owner_user_id: str) -> LocalMcpClient:
        records = await self._repository().list(owner_user_id=owner_user_id)
        if not records:
            # 2. 运行时只在“零记录”时使用项目默认地址。
            return LocalMcpClient(
                self._default_url,
                timeout_seconds=self._default_timeout_seconds,
                retries=self._default_retries,
            )
        return _client_from_records([record for record in records if record.enabled])

它证明“全部禁用”会装配一个零连接客户端，而不是重新启用 CLS 默认连接。边界也很清楚：默认记录创建发生在列表请求中，运行时回退本身不会写库；两条路径最终都仍受当前 owner 的记录集合控制。

📷 [图片 token=AvlobqQ0no2cTVxvGzAcU6C9nOf（未能下载，见飞书原文）]

看什么：下面的状态图专门区分零记录、已有启用记录和已有但全禁用三种状态，避免把“没有配置”与“用户主动禁用”混为一谈。

这张图证明连接状态的分支点是 owner-scoped 持久记录，而不是全局配置是否存在。失败与一致性边界是：禁用状态不会被默认回退覆盖，但零连接客户端自然无法提供 MCP 工具，调用方必须把这一事实显式呈现。

工具发现先于可信调用

LocalMcpClient.discover_tools 对每条连接执行 ClientSession.initialize 和 session.list_tools，把服务端返回的名称、描述、输入 schema 和连接标识转换成 McpToolDefinition。当前用户连接服务把持久化记录的 ID 作为这个内部 server name。这是真实网络发现，不是根据本地配置拼出一组静态工具。检查接口捕获 McpClientError 后保存失败状态和安全提示，工具列表保持为空，不会为了让页面“看起来正常”而填入预设名称。

📷 [图片 token=Z2KjbqunWooSuUxl8vpcbQsfnid（未能下载，见飞书原文）]

看什么：连接检查先按 owner 读取指定记录，再对这一条记录创建客户端；发现异常只转换为固定中文错误，最后把真实工具摘要或空列表写入 lastCheck。

        record = await repository.get(
            owner_user_id=owner_user_id,
            connection_id=connection_id,
        )
        if record is None:
            raise McpConnectionError("AUTH_FORBIDDEN", "MCP connection is not accessible.")
        client = _client_from_records([record])
        tools: list[McpToolDefinition] = []
        error: str | None = None
        try:
            # 1. 检查动作发起真实 initialize 与 list_tools。
            tools = await client.discover_tools()
        except McpClientError:
            error = "MCP Server 不可用或工具发现失败。"
        # 2. 失败时 tools 仍为空，且只保存安全错误。
        updated = await repository.save_check(
            owner_user_id=owner_user_id,
            connection_id=connection_id,
            ok=error is None,
            tools=[_tool_payload(tool) for tool in tools],
            error=error,
        )

这段代码证明检查接口不会把其他 owner 的连接拿来探测，也不会把底层异常原文保存到连接状态。它的边界是：固定错误提高了秘密安全性，却有意牺牲底层网络错误细节；排障时应结合脱敏运行日志，而不是期待 lastCheck.error 含堆栈。

多连接场景还要处理工具名冲突。discover_tools 使用 seen 集合拒绝重复名称；call_tool 在多连接时重新确认名称只能匹配一个定义，再按 server_name 找到目标连接。如果工具缺失或同名歧义，调用直接失败。这种保护比“随便选择第一个 Server”更可靠，因为它不会把日志查询发送到错误系统。

看什么：关注 seen 与 connection.name 两个字段。前者把工具名当作跨连接唯一键，后者把发现结果绑定回具体 Server。

        definitions: list[McpToolDefinition] = []
        seen: set[str] = set()
        for connection in self._connections:
            result = await self._run_connection(
                connection,
                lambda session: session.list_tools(),
            )
            for tool in result.tools:
                # 1. 任意两个连接暴露同名工具时，整次发现明确失败。
                if tool.name in seen:
                    raise McpClientError(f"Duplicate MCP tool name: {tool.name}")
                seen.add(tool.name)
                definitions.append(
                    McpToolDefinition(
                        tool.name,
                        tool.description or "MCP tool",
                        tool.inputSchema,
                        # 2. server_name 保留后续调用所需的归属。
                        connection.name,
                    )
                )
        return definitions

它证明多连接不是简单拼接清单：发现阶段就建立“工具名唯一、定义可追到 Server”的不变量。失败边界是任一连接发现失败或任意同名冲突都会使整次发现失败；当前实现没有命名空间降级或部分成功模式。

📷 [图片 token=OohEb7nEaoDoNJxPc1GcJxkAnyd（未能下载，见飞书原文）]

看什么：这张时序图把显式检查和运行时装配并列起来。两者都以真实发现为信任门槛，但只有检查路径会保存最近检查结果。

这张图证明工具定义来自 Server 会话，而不是 SQLite 中上次保存的摘要；保存的 lastCheck 是展示与诊断状态，不是运行时可信调用的静态白名单。权限边界始终在最前面的 owner 查询处。

📷 [图片 token=FFkrbOlCQoRBN1xvFQ4cztornXb（未能下载，见飞书原文）]

真实调用、重试和日志边界

单连接调用通过 session.call_tool 执行，多连接调用先完成工具归属解析。_run_connection 按 retries + 1 次尝试执行，重试间隔为递增的短等待；SSE 走 sse_client，Streamable HTTP 走 streamable_http_client。两条路径最终都进入 _run_initialized_session，在 session 初始化后用 asyncio.wait_for 施加超时。

看什么：单连接可以直接调用；多连接必须先重新发现，再要求工具名恰好匹配一个定义，并按定义中的 server_name 选择连接。

            if len(self._connections) == 1:
                connection = self._connections[0]
                # 1. 单连接无需归属消歧，直接进入统一运行器。
                result = await self._run(
                    lambda session: session.call_tool(
                        name,
                        arguments,
                        read_timeout_seconds=timedelta(
                            seconds=connection.timeout_seconds
                        ),
                    )
                )
            else:
                matching = [tool for tool in await self.discover_tools() if tool.name == name]
                # 2. 多连接只接受唯一匹配，缺失与歧义都失败。
                if len(matching) != 1:
                    raise McpClientError(f"MCP tool is unavailable or ambiguous: {name}")
                connection = next(
                    item for item in self._connections if item.name == matching[0].server_name
                )

这段代码证明调用不会在多连接之间“任选一个”。它也暴露一个一致性取舍：多连接调用前会再次做网络发现，因此能反映 Server 当前清单，但发现失败会在真正调用之前终止本次执行。

📷 [图片 token=OIu5bzQcbo2WwFxUaENczX4un9d（未能下载，见飞书原文）]

看什么：统一运行器把重试、transport 分派、session 初始化和操作超时串成一条链；重试耗尽后只抛统一的 McpClientError，原异常作为 cause 保留在服务端异常链中。

        error: Exception | None = None
        for attempt in range(connection.retries + 1):
            try:
                # 1. 每次尝试按持久化 transport 建立新的客户端会话。
                if connection.transport == "streamable_http":
                    return await self._run_streamable_http(connection, operation)
                return await self._run_sse(connection, operation)
            except Exception as exc:
                error = exc
                if attempt < connection.retries:
                    await asyncio.sleep(0.2 * (attempt + 1))
        raise McpClientError(f"MCP server unavailable at {connection.url}") from error

# … 省略两个 transport 的流创建代码
    async with ClientSession(read_stream, write_stream) as session:
        # 2. initialize 完成后，具体操作仍受 asyncio 超时保护。
        await session.initialize()
        return await asyncio.wait_for(operation(session), timeout=timeout_seconds)

它证明配置中的 retries 表示额外重试次数，总尝试数是 retries + 1。失败边界包括连接、初始化、列举与调用异常，都会进入同一重试循环；当前实现没有按异常类型区分“可重试”和“不可重试”。

📷 [图片 token=LlHqbL042oXVbtxXEkRcAccwn1g（未能下载，见飞书原文）]

当 MCP 返回 isError 时，客户端抛出 McpClientError，不会把错误内容当成成功输出。结构化运行日志只记录工具名、参数键名、结果条数、耗时和错误类别；apps/backend/tests/test_mcp_observability.py 明确断言参数值和工具输出不会进入日志。需要区分的是：运行日志为脱敏运维信号，工具审计是受 owner 保护的业务记录，两者职责不同。

看什么：最后用局部时序图观察“失败后重试”与“成功后日志”的分叉；注意 MCP 返回 isError 时不会发 completed。

这张图证明运行日志和业务返回在调用完成后才分流：前者刻意不含参数值与输出，后者保留真实 payload 供 Agent 和 owner-scoped 审计使用。安全边界不是“工具输出无敏感信息”，而是运行日志不复制这些内容；工具参数与审计仍应避免承载凭据。

📷 [图片 token=WHOYbb5hFoFPjix9Ld0cBdQHnWg（未能下载，见飞书原文）]

数据、契约与状态

McpConnectionModel 保存连接 ID、owner、名称、transport、URL、启用状态、超时、重试次数以及最近检查字段。last_check_ok、last_tool_count、last_tools、last_error 和 last_checked_at 共同表达检查结果。共享契约 McpConnectionCheck 将其整理为 ok、toolCount、tools、error、checkedAt，前端无需猜测数据库状态。

“保存成功”不等于“连接可用”。创建或更新只证明字段通过校验并进入 SQLite；只有显式 :check 调用完成真实工具发现后，页面才有最近检查结果。类似地，“发现到工具”也不等于“某次调用成功”，每次 call_tool 仍可能因网络、超时、服务端错误或工具参数失败。教学和排障时应保留这三层状态，不能把它们合并成一个绿色标记。

📷 [图片 token=UcrebRd37o46XRxXcQIcjEnyn5d（未能下载，见飞书原文）]

Agent 使用的工具定义来自运行时发现。聊天侧会把 MCP 工具与知识检索、当前时间等工具一起交给 LangChain Agent；AIOps 侧则在 Planner 中发现可用工具，在 Executor 中调用选定工具。两条链路都以当前 owner_user_id 调用 client_for_user，连接状态不会跨用户共享。

📷 [图片 token=Bf5ObvD6Xo9oluxRPeOckioznEd（未能下载，见飞书原文）]

权限、安全与失败边界

第一道边界是认证路由，第二道边界是 service/repository 的 owner 条件。更新或检查不存在于当前用户范围的连接时，服务返回 AUTH_FORBIDDEN；删除返回 false 时，API 同样映射为统一权限错误。这样不会通过“存在返回 404、不存在返回 403”的差异泄露其他用户是否拥有某个连接。

📷 [图片 token=JScWbmV0Fo94ELxfeVScVVdwnrf（未能下载，见飞书原文）]

URL 校验禁止 userinfo，但它并不等于完整的网络出口安全体系。当前实现允许用户配置 HTTP/HTTPS hostname，因此部署到更开放的环境时仍需结合网络策略考虑 SSRF 和私网访问范围。OncallAgent 的定位是本地优先工作台，文档不能把这一校验夸大成通用企业网关。

📷 [图片 token=XODNbRtt4oj8CZxkOa4cHbKUnvh（未能下载，见飞书原文）]

如果 MCP Server 不可用，连接检查保存明确失败；运行时发现或调用抛出错误；AIOps 会生成失败工具事件、失败审计和证据不足报告。代码没有“工具失败后返回一份示例日志”的兜底。项目默认连接也只是配置回退，不是可用性回退：默认 CLS Server 未启动时，readiness 和调用都必须如实失败。

📷 [图片 token=RqR3by8jaow94DxLGP2ctt6FnDj（未能下载，见飞书原文）]

阅读顺序与小结

1. 先读 packages/api-contracts/src/mcp.ts，建立页面和 API 看到的数据结构。

2. 再沿 mcpClient.ts、useMcpStore 和 api/app.py 看一次连接检查。

3. 进入 mcp_connections.py 理解校验、默认连接和启用过滤。

4. 最后读 mcp_client.py，确认工具发现、真实调用、重试、冲突和脱敏边界。

这套实现的核心不是“让 Agent 拥有更多按钮”，而是让每个外部能力都有明确来源、用户归属、真实调用和失败证据。只有沿连接记录、发现结果、工具调用、SSE 事件和审计记录逐段核对，才能判断一次外部取证是否可信。