在 Codex 里接 MCP

先确认配置被读取,再验证初始化、工具目录与一次真实结果。用明确的权限范围和可复现的离线实验,缩小每次排错的范围。

你能带走什么
  • Codex 直接 MCP 配置使用 config.toml,先确认主机与项目作用域
  • list 与 get 只能证明配置存在,实际连接和工具调用要分别验证
  • OAuth、环境变量和服务端授权各有边界,秘密不进入示例
  • 离线 mock 帮助复现故障,真实服务与模型行为仍需单独验收

1. 先确定服务运行在哪里

给 Codex 接入 MCP,先选一个能独立核对结果的只读任务。例如查询开发文档,或读取一条已知测试记录。接通后保存服务名、工具名、输入、结果来源和客户端版本。以后出现故障,这些记录能帮助你判断变化发生在哪一层。

本文面向 Codex CLI 与同一主机上的本地客户端,讲直接配置和排错。当前官方文档列出 STDIO 与 Streamable HTTP:前者由宿主启动一个进程,后者连接已经部署的端点。换到远程执行环境或托管插件时,需要重新确认连接位置和认证状态,不能假设本机能访问的地址在云端也能访问。支持范围见官方 MCP 文档。

图 01 / 概念示意
STDIO 由 Codex host 启动进程;Streamable HTTP 连接已部署端点。
两种传输承担不同的连接方式,认证和权限还要另行核对。
查看图示文字说明

STDIO 由 Codex host 启动进程;Streamable HTTP 连接已部署端点。

STDIO 适合维护方提供启动命令的服务。它是否能运行,取决于对应主机上的可执行文件、依赖和工作目录。Streamable HTTP 适合服务方提供 HTTPS 地址的情况,客户端需要能到达该端点,并满足服务端的认证要求。URL 应来自服务维护方;网页首页、REST API 地址和 MCP 端点可能各不相同。

本篇不展开 Events、任务订阅或 Skill 编写。文末的本地实验只处理初始化、工具发现和一次确定性查询,用来观察常见故障。它不会访问真实账号,也不会调用模型。

2. 从官方文档服务建立一个最小配置

OpenAI Docs MCP 提供开发文档的搜索和正文读取,官方给出了公开的 Streamable HTTP 地址。下面是可复制的 Codex 配置命令;需要本机已经安装 Codex,且当前网络能够访问该官方端点。示例会新增用户级配置,执行前先看已有服务,避免覆盖同名项。官方 Docs MCP 快速开始。

终端 · 先查看,再添加
codex --version
codex mcp list
codex mcp add openaiDeveloperDocs   --url https://developers.openai.com/mcp
codex mcp get openaiDeveloperDocs --json

也可以把下面这段写入用户配置文件 ~/.codex/config.toml。命令方式和手写方式任选一种;不要重复声明同名 TOML 表。这里无需为了教程准备 API key,Docs 服务用于查文档,不会代你调用 OpenAI API。

~/.codex/config.toml · 官方端点
[mcp_servers.openaiDeveloperDocs]
url = "https://developers.openai.com/mcp"

配置保存后,打开新的 Codex 会话,用 /mcp 查看活动服务,再要求它通过这个服务查询一个具体文档问题,并返回来源链接。核对实际工具调用和来源页面。list 与 get 展示配置,不能证明端点成功初始化,也不能证明模型已经调用工具。本文未执行这条远程连接或真实模型查询。

普通 Codex 直接连接使用 config.toml。其他编辑器的 mcp.json,以及插件包中的 .mcp.json,属于各自的配置入口。迁移时要查目标客户端文档,不能只改文件名后继续使用原来的 JSON 字段。官方目前也支持插件提供 MCP 服务,但插件的启动配置与用户侧工具策略有各自的位置。

3. 把 STDIO 的进程与输出检查清楚

STDIO 配置把程序和参数分开。command 指向可执行文件,args 按参数逐项排列;cwd 指定服务进程的工作目录。不要依赖一个终端里临时生效的别名,也不要默认图形客户端拥有相同 PATH。排错时用绝对路径更容易判断实际启动的是哪份程序。

教学 mock 配置 · 替换成本机真实绝对路径后可用
[mcp_servers.offline_catalog]
command = "/absolute/path/to/python3"
args = ["-u", "/absolute/path/to/mock_server.py"]
cwd = "/absolute/path/to/example-folder"
startup_timeout_sec = 10
tool_timeout_sec = 15
enabled_tools = ["catalog_lookup"]
default_tools_approval_mode = "prompt"

上面的路径都是占位路径,原样执行会失败。示例文件在本文后半部分,先用模拟客户端完成离线验证,再考虑把真实路径加入 Codex。若使用 CLI,形式是 codex mcp add offline_catalog -- python3 /实际路径/mock_server.py;存在空格的路径要按当前 shell 的规则引用。本文代码只要求 Python 3.9 或更高版本。

图 02 / 概念示意
stdin 承载请求,stdout 承载 JSON-RPC 响应,stderr 独立记录日志。
协议输出和诊断日志分流,客户端可同时保留两份证据。
查看图示文字说明

stdin 承载请求,stdout 承载 JSON-RPC 响应,stderr 独立记录日志。

MCP 的 STDIO 传输使用 UTF-8 JSON-RPC 消息,每条消息占一行。stdout 只能写协议消息;启动横幅、调试信息和堆栈诊断应写 stderr。JSON 字符串中的换行要由 JSON 编码器转义,不能把一个 JSON 对象排成多行后当作 STDIO 帧。传输规范。

程序启动后安静等待 stdin 是正常状态。手动运行时没有输出,不能直接判断服务挂了;反过来,看到一行 started 也不能证明完成初始化。下面的实验会等待响应,遇到超时或提前关闭管道时明确失败,并把 stderr 单独存档。

4. 管理命令与配置作用域

本次在 codex-cli 0.159.2 上逐项读取了 add、list、get、remove、login、logout 的帮助。list 与 get 支持 --json;add 的远程入口是 --url,本地进程入口位于 -- 之后。--env KEY=VALUE 只用于 STDIO;--bearer-token-env-var 只用于 Streamable HTTP。遇到较旧安装版本,先运行对应子命令的 --help,再按它支持的参数调整。

按需执行 · NAME 换成已有服务名
codex mcp list --json
codex mcp get NAME --json
codex mcp login NAME
codex mcp logout NAME
codex mcp remove NAME

上述命令承担不同动作,不能整段当作安装脚本运行。login 用于支持 OAuth 的服务;logout 清理该服务的本地登录状态;remove 删除配置。删除本地配置不等于在服务提供方撤销授权,停用账号或撤销服务端令牌需按对方的管理流程处理。想暂时停用,可在对应表中设置 enabled = false。

本次还使用独立的临时 CODEX_HOME,实际执行了本地配置的 add、list、get、remove 往返,返回均为成功;add 明确报告添加的是 global 服务。没有向真实用户目录写配置,也没有由 Codex 启动该 mock。需要项目级设置时,按官方文档编辑受信项目的 .codex/config.toml,不要给 add 补上未核实的作用域参数。

图 03 / 概念示意
CLI、受信项目、profile、用户、云默认、系统、内置依次取值,组织 requirements 另行约束。
诊断时同时记录当前目录、CODEX_HOME、profile 与受信状态。
查看图示文字说明

CLI、受信项目、profile、用户、云默认、系统、内置依次取值,组织 requirements 另行约束。

当前官方配置优先级从高到低为:CLI 覆盖、受信项目配置、选中的 profile 文件、用户配置、云托管默认值、系统配置、内置默认值。项目内离当前目录更近的配置优先。组织 requirements 属于约束,不能靠 -c 绕过。这个完整顺序依据本次官方文档,未在本机重建所有管理层进行集成测试。配置层级与信任前提。

排查同一个仓库在不同终端表现不同,先记录启动目录、CODEX_HOME 和所选 profile。项目未被信任时,项目配置不会加载。审阅项目内启动命令和脚本后再决定是否信任;陌生仓库里的 MCP 配置可能启动代码。CODEX_HOME 默认是 ~/.codex,设置自定义目录时目录应已经存在。环境变量参考。

5. 认证与超时按服务单独配置

OAuth 适合服务提供方支持浏览器授权的场景。先核对授权域名、登录账号和申请范围,再使用 codex mcp login NAME。本机帮助还列出 --scopes、--no-browser 与 --oauth-client-registration;这些高级选项用于具体服务的兼容问题,不应在没有服务要求时随意拼上。Codex 登录状态也不会自动成为所有第三方 MCP 的授权。

若提供方给的是 Bearer token,配置中只保留环境变量名。下面使用 example.invalid 明确表示不可连接的占位地址,须换成经过核实的服务端点。秘密值由受控启动环境注入,不写进 TOML、shell 历史或截图。STDIO 可用 env_vars 指定要转发的变量;env 与 --env 保存的是具体值,适合非敏感开关,放秘密会增加落盘风险。

HTTP 认证结构示例 · 占位 URL 不可运行
[mcp_servers.team_docs]
url = "https://example.invalid/mcp"
bearer_token_env_var = "TEAM_DOCS_TOKEN"
startup_timeout_sec = 20
tool_timeout_sec = 45
enabled = true
图 04 / 概念示意
公开服务、OAuth 服务和环境变量凭据三条路线,各自需要服务端支持。
用环境变量引用避免把秘密写入示例,运行时仍需保护进程与日志。
查看图示文字说明

公开服务、OAuth 服务和环境变量凭据三条路线,各自需要服务端支持。

环境变量只是传值方式,不提供加密保证。MCP 子进程、调试日志和错误报告都可能接触秘密;应缩小令牌权限、有效期和可读取进程范围。也不要把 env 文件交给模型总结后贴到工单。OAuth 凭据另由客户端的凭据存储设置管理;没有解析到凭据时,客户端可能尝试匿名连接,需要授权的服务仍会拒绝。

startup_timeout_sec 控制服务启动等待,官方默认十秒;tool_timeout_sec 控制单次工具执行,默认六十秒。required = true 会让已启用服务初始化失败成为启动或恢复失败。可选服务的初始目录等待还受 mcp_optional_startup_grace_ms 影响,当前默认一千毫秒,设为零时按各服务启动超时等待。字段定义与默认值。

增加等待时间之前,先分清进程启动、认证、工具调用和上游业务服务哪一步耗时。涉及写操作时,超时之后远端可能已完成动作;先查询目标状态,再决定是否重试。对不支持幂等性的接口,盲目重试可能重复创建或发送。

6. 让工具范围与权限保持可检查

服务连接后会声明工具及其输入 schema。工具名、描述、参数约束和来源都值得检查。本文 mock 只暴露 catalog_lookup,一个名称对应一项固定查询。真实集成可采用清晰、稳定的命名,避免几个服务都用没有上下文的 run 或 execute;不同界面可能显示带服务前缀的名称,配置筛选时应依据服务实际公布的工具名。

enabled_tools 是允许列表,disabled_tools 在它之后应用;同一工具若同时出现于两者,会被禁用。新接入时可以先开放最少的只读工具,敏感工具单独要求确认。当前配置支持服务级 default_tools_approval_mode 和单工具 approval_mode;本文用 prompt 保留审阅机会。服务升级增加新工具后,应重新检查目录与允许范围。

图 05 / 概念示意
Codex 工具筛选、服务端授权、外部内容信任分别检查;远程副作用需明确授权。
按层设置限制,可以避免把成功连接误当作完整授权。
查看图示文字说明

Codex 工具筛选、服务端授权、外部内容信任分别检查;远程副作用需明确授权。

readOnlyHint 等注解是服务方提供的提示。客户端不能据此替服务端完成账号、租户、资源与写入权限校验。远程服务持有的账号权限也不会因为本地命令使用只读沙箱就自动缩小;官方明确把命令沙箱、MCP 连接和连接器列为分别控制的通道。工具规范;Codex 权限边界。

提示词注入可能出现在工具描述、初始化说明或返回的文档正文里。例如检索结果要求把环境文件上传到另一个域名,这已经超出查询任务。保留来源,把相关文字当作外部内容处理;不要因它声称是管理员要求就扩大权限。关闭保护无法修复这个信任问题。服务侧还需使用独立授权、参数校验和审计,减少一次错误调用能造成的影响。安全与审批说明。

7. 按阶段排查并保留足够的日志

先用 list 和 get 确认目标配置,再检查 /mcp 中的活动状态。如果配置存在但没有工具,继续看初始化、服务 capabilities、工具筛选和组织策略。只有目录出现还不够,要执行一次已知输入的只读调用,并验证结果;鉴权成功、协议成功和业务成功是三个需要分别记录的条件。

图 06 / 概念示意
六阶段诊断路径:配置、进程、初始化、工具目录、调用、结果核对。
先区分配置、连接、工具目录和业务调用,再决定修改位置。
查看图示文字说明

六阶段诊断路径:配置、进程、初始化、工具目录、调用、结果核对。

STDIO 问题先复核 command、args、cwd、运行时版本和 stderr。HTTP 问题先确认维护方公布的端点,再看网络可达性、证书、代理与认证。401 或 403 的含义要结合服务返回内容判断,scope 不足与令牌失效需要不同处理;不要把更换账号或扩大授权当成通用修复。

Streamable HTTP 的规范允许 JSON 响应,也允许 SSE 承载响应流;看到 SSE 字样不能直接判断它使用旧传输。2025-11-25 规范说明 Streamable HTTP 替代了 2024-11-05 的 HTTP+SSE,并给出迁移建议。这能解释协议差异,不能据此宣称每个 Codex 版本都自动兼容每个旧 SSE 端点。遇到旧服务,应核对维护方的传输说明和双方版本。

当前官方日志说明中,明文 TUI 日志需要显式设置 log_dir 才启用。下面命令将诊断写入指定目录;RUST_LOG=debug 增加日志细节。先复现一次最小问题,提取服务名、失败阶段、时间及已脱敏错误。日志可能包含业务内容和认证信息,发送前要逐段检查。日志与环境变量。

一次诊断会话 · 会启动真实 Codex
RUST_LOG=debug codex -c log_dir=./.codex-log

服务暂时不可用时,记录原始错误,保持已核实的配置。若任务依赖该服务,应停止相关步骤,说明缺失的数据会影响哪些结论。若只影响可选查询,可以继续不依赖它的工作,但不要用模型记忆伪装成最新工具结果。恢复后用同一只读测试复核,再恢复依赖这项数据的流程。

不要把整份配置和日志直接发到公开 issue。保留一份删除私密数据的最小配置,注明 CLI 版本、操作系统、服务版本及稳定复现步骤;若没有服务端日志权限,把请求时间和脱敏错误交给维护方即可。本文没有读取真实用户日志。

8. 用无网络实验观察协议与失败路径

下面三个完整文件放在同一目录。mock_server.py 提供一个虚构库存工具,simulated_client.py 通过子进程管道调用它,test_offline.py 检查正例和反例。只用 Python 标准库,没有网络端口、第三方依赖、模型请求或生产认证。模拟客户端对子进程使用空环境,不传入宿主已有令牌。

实验固定使用 MCP 2025-11-25,只实现教学需要的子集:initialize 响应之后发送 notifications/initialized,再请求 tools/list 和 tools/call。请求与响应 ID 对应,通知没有响应;结果中的 isError 还要额外检查。生命周期规范与JSON-RPC 2.0 规范给出了消息语义。

2026-10-20 更正:先区分请求结构与工具输入校验。未知工具、缺少 name 或 arguments 为数组/null,仍返回 JSON-RPC 错误;已找到工具但缺少 sku、sku 类型错误或多了字段,则返回带说明的 result.isError = true。协议允许省略 arguments,但本工具要求 sku,所以省略后仍是工具输入错误。依据 错误处理规则与 CallToolRequestParams,不能把所有参数问题都归为 -32602。

在三个文件所在目录执行
python3 simulated_client.py --transcript transcript.jsonl
python3 -m unittest -v test_offline.py

# 人为制造一次 stdout 污染,预期退出码为 1
python3 simulated_client.py --fault stdout

正常运行输出 ok 为 true、frames 为七,固定业务结果为 DEMO-1: 3 units。transcript.jsonl 记录两个方向的协议消息,transcript.stderr.txt 保存服务诊断。这里的七帧来自三次请求及响应,加一次 initialized 通知;这个固定计数仅适用于本实验。

mock_server.py
"""Offline teaching subset of MCP 2025-11-25; not a production SDK."""
import argparse
import json
import sys
import time

VERSION = "2025-11-25"
TOOL = {
    "name": "catalog_lookup",
    "description": "Read a fixed fictional product catalog; no network or files.",
    "inputSchema": {
        "type": "object",
        "properties": {"sku": {"type": "string"}},
        "required": ["sku"],
        "additionalProperties": False,
    },
    "annotations": {"readOnlyHint": True, "openWorldHint": False},
}


def handle(message, state):
    if not isinstance(message, dict) or message.get("jsonrpc") != "2.0":
        return error(None, -32600, "Invalid Request")
    method = message.get("method")
    if not isinstance(method, str):
        return error(message.get("id"), -32600, "Invalid Request")
    if "id" not in message:
        if method == "notifications/initialized" and state[0] == "negotiated":
            state[0] = "ready"
        return None
    rid = message["id"]
    if type(rid) not in (str, int):
        return error(None, -32600, "Invalid request ID")
    params = message.get("params", {})
    if not isinstance(params, dict):
        return error(rid, -32602, "Expected object parameters")
    if method == "initialize" and state[0] == "new":
        if not isinstance(params.get("protocolVersion"), str):
            return error(rid, -32602, "Missing protocolVersion")
        state[0] = "negotiated"
        result = {
            "protocolVersion": VERSION,
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "offline-catalog", "version": "0.1.0"},
        }
    elif method == "ping":
        result = {}
    elif state[0] != "ready":
        return error(rid, -32000, "Initialize this teaching server first")
    elif method == "tools/list":
        result = {"tools": [TOOL]}
    elif method == "tools/call":
        if params.get("name") != TOOL["name"]:
            return error(rid, -32602, "Unknown tool")
        args = params.get("arguments", {})
        if not isinstance(args, dict):
            return error(rid, -32602, "Expected object arguments")
        if set(args) != {"sku"}:
            return tool_error(rid, "Provide exactly one argument: sku")
        if not isinstance(args["sku"], str):
            return tool_error(rid, "sku must be a string")
        if args["sku"] != "DEMO-1":
            result = {
                "content": [{"type": "text", "text": "Fictional SKU missing"}],
                "isError": True,
            }
        else:
            result = {
                "content": [{"type": "text", "text": "DEMO-1: 3 units"}],
                "isError": False,
            }
    else:
        return error(rid, -32601, "Method not found")
    return {"jsonrpc": "2.0", "id": rid, "result": result}


def tool_error(rid, message):
    return {
        "jsonrpc": "2.0", "id": rid,
        "result": {
            "content": [{"type": "text", "text": message}],
            "isError": True,
        },
    }


def error(rid, code, message):
    return {
        "jsonrpc": "2.0", "id": rid,
        "error": {"code": code, "message": message},
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--fault", default="none", choices=[
        "none", "stdout", "id", "version", "both", "tool", "slow", "eof",
    ])
    fault = parser.parse_args().fault
    print("offline teaching server started", file=sys.stderr, flush=True)
    state = ["new"]
    for line in sys.stdin:
        try:
            message = json.loads(line)
        except json.JSONDecodeError:
            response = error(None, -32700, "Parse error")
            message = {}
        else:
            response = handle(message, state)
        if response is None:
            continue
        method = message.get("method") if isinstance(message, dict) else None
        if method == "initialize":
            if fault == "slow":
                time.sleep(10)
            if fault == "eof":
                return
            if fault == "stdout":
                print("DEBUG: not a protocol message", flush=True)
            if fault == "id":
                response["id"] = 999
            if fault == "version":
                response["result"]["protocolVersion"] = "1900-01-01"
            if fault == "both":
                response["error"] = {"code": -32603, "message": "test"}
        if method == "tools/call" and fault == "tool":
            response["result"]["isError"] = True
        print(json.dumps(response, ensure_ascii=False), flush=True)


if __name__ == "__main__":
    main()

故障开关仅服务于实验:stdout 混入普通文字,id 返回错误请求号,version 返回不支持的版本,both 同时出现 result 与 error,tool 返回业务失败,slow 延迟,eof 提前结束。真实服务出现未知消息时应按官方 SDK 和协商能力处理,不要把本客户端的严格单请求假设扩展到生产。

simulated_client.py
"""Small deterministic client for mock_server.py, not Codex or a full MCP client."""
import argparse
import json
from pathlib import Path
import queue
import subprocess
import sys
import threading

VERSION = "2025-11-25"
MAX_LINE = 65536


class ProtocolError(Exception):
    pass


def validate_response(message, expected_id):
    if not isinstance(message, dict) or message.get("jsonrpc") != "2.0":
        raise ProtocolError("Expected a JSON-RPC 2.0 object")
    if type(message.get("id")) is not int or message["id"] != expected_id:
        raise ProtocolError("Response ID mismatch")
    if "method" in message or (("result" in message) == ("error" in message)):
        raise ProtocolError("Expected exactly one result or error")
    if "error" in message:
        raise ProtocolError("JSON-RPC error: " + str(message["error"]))
    if not isinstance(message["result"], dict):
        raise ProtocolError("This teaching client expects object results")
    return message["result"]


def exercise(fault="none", timeout=2.0, transcript=None):
    server = Path(__file__).with_name("mock_server.py")
    proc = subprocess.Popen(
        [sys.executable, "-u", str(server), "--fault", fault],
        stdin=subprocess.PIPE, stdout=subprocess.PIPE,
        stderr=subprocess.PIPE, text=True, encoding="utf-8", env={},
    )
    incoming = queue.Queue()
    stderr_lines = []
    records = []

    def read_stdout():
        try:
            while True:
                line = proc.stdout.readline(MAX_LINE + 1)
                if not line:
                    incoming.put(None)
                    return
                if len(line) > MAX_LINE or not line.endswith("\n"):
                    incoming.put(ProtocolError("Oversized or incomplete frame"))
                    return
                incoming.put(line)
        except Exception as exc:
            incoming.put(ProtocolError(str(exc)))

    def read_stderr():
        for line in proc.stderr:
            if len(stderr_lines) < 20:
                stderr_lines.append(line.rstrip())

    readers = [threading.Thread(target=f, daemon=True) for f in (
        read_stdout, read_stderr,
    )]
    for thread in readers:
        thread.start()

    def send(message):
        records.append({"direction": "client", "message": message})
        proc.stdin.write(json.dumps(message) + "\n")
        proc.stdin.flush()

    def request(rid, method, params):
        send({"jsonrpc": "2.0", "id": rid, "method": method, "params": params})
        try:
            line = incoming.get(timeout=timeout)
        except queue.Empty as exc:
            raise ProtocolError("Response timeout") from exc
        if line is None:
            raise ProtocolError("Server closed stdout early")
        if isinstance(line, Exception):
            raise line
        try:
            response = json.loads(line)
        except json.JSONDecodeError as exc:
            raise ProtocolError("Non-JSON stdout") from exc
        records.append({"direction": "server", "message": response})
        return validate_response(response, rid)

    try:
        init = request(1, "initialize", {
            "protocolVersion": VERSION,
            "capabilities": {},
            "clientInfo": {"name": "offline-check", "version": "0.1.0"},
        })
        if init.get("protocolVersion") != VERSION:
            raise ProtocolError("Unsupported negotiated protocol version")
        if "tools" not in init.get("capabilities", {}):
            raise ProtocolError("No tools capability")
        send({"jsonrpc": "2.0", "method": "notifications/initialized"})
        listing = request(2, "tools/list", {})
        tools = listing.get("tools")
        if not isinstance(tools, list) or len(tools) != 1:
            raise ProtocolError("Unexpected fictional catalog tool list")
        tool = tools[0]
        if tool.get("name") != "catalog_lookup":
            raise ProtocolError("Tool name mismatch")
        if tool.get("inputSchema", {}).get("required") != ["sku"]:
            raise ProtocolError("Tool schema mismatch")
        result = request(3, "tools/call", {
            "name": tool["name"], "arguments": {"sku": "DEMO-1"},
        })
        if result.get("isError", False) is not False:
            raise ProtocolError("Tool-level error")
        expected = [{"type": "text", "text": "DEMO-1: 3 units"}]
        if result.get("content") != expected:
            raise ProtocolError("Fictional business result mismatch")
        return {"ok": True, "frames": len(records), "result": expected[0]["text"]}
    finally:
        if proc.stdin:
            proc.stdin.close()
        try:
            proc.wait(timeout=0.5)
        except subprocess.TimeoutExpired:
            proc.terminate()
            try:
                proc.wait(timeout=0.5)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait()
        for thread in readers:
            thread.join(timeout=0.5)
        proc.stdout.close()
        proc.stderr.close()
        if transcript:
            path = Path(transcript)
            rows = [json.dumps(r, ensure_ascii=False) for r in records]
            path.write_text("\n".join(rows) + "\n", encoding="utf-8")
            path.with_suffix(".stderr.txt").write_text(
                "\n".join(stderr_lines) + "\n", encoding="utf-8",
            )


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--fault", default="none")
    parser.add_argument("--timeout", type=float, default=2.0)
    parser.add_argument("--transcript")
    args = parser.parse_args()
    try:
        print(json.dumps(exercise(args.fault, args.timeout, args.transcript)))
    except ProtocolError as error:
        print(str(error), file=sys.stderr)
        raise SystemExit(1)

客户端设响应截止时间并在结束时关闭子进程,区分非 JSON 输出、响应 ID 不匹配、协议错误与工具错误。它还核对一项已知业务事实,避免仅凭收到 JSON 就宣布成功。它不实现服务器主动请求、分页、取消、进度、并发、HTTP 或完整 schema 验证,也未做恶意服务的资源耗尽防护,不能用作生产客户端。

test_offline.py
"""Run with: python3 -m unittest -v test_offline.py"""
import copy
import unittest
from mock_server import handle
from simulated_client import ProtocolError, exercise, validate_response


class OfflineTests(unittest.TestCase):
    def test_success(self):
        result = exercise()
        self.assertEqual(result["frames"], 7)
        self.assertEqual(result["result"], "DEMO-1: 3 units")

    def test_faults(self):
        cases = {
            "stdout": "Non-JSON stdout",
            "id": "Response ID mismatch",
            "version": "Unsupported negotiated",
            "both": "exactly one",
            "tool": "Tool-level error",
            "slow": "Response timeout",
            "eof": "closed stdout early",
        }
        for fault, message in cases.items():
            with self.subTest(fault=fault):
                with self.assertRaisesRegex(ProtocolError, message):
                    exercise(fault, timeout=0.3 if fault == "slow" else 2.0)

    def test_response_envelopes(self):
        good = {"jsonrpc": "2.0", "id": 1, "result": {}}
        self.assertEqual(validate_response(good, 1), {})
        mutations = [
            {"jsonrpc": "1.0"}, {"id": True}, {"result": []},
            {"error": {"code": -32603, "message": "test"}},
            {"method": "unexpected"},
        ]
        for update in mutations:
            with self.subTest(update=update):
                message = copy.deepcopy(good)
                message.update(update)
                with self.assertRaises(ProtocolError):
                    validate_response(message, 1)

    def test_notification_has_no_response(self):
        state = ["negotiated"]
        message = {"jsonrpc": "2.0", "method": "notifications/initialized"}
        self.assertIsNone(handle(message, state))
        self.assertEqual(state[0], "ready")

    def test_request_before_initialization(self):
        request = {"jsonrpc": "2.0", "id": 1, "method": "tools/list"}
        self.assertEqual(handle(request, ["new"])["error"]["code"], -32000)

    def test_unknown_method(self):
        request = {"jsonrpc": "2.0", "id": 1, "method": "unknown"}
        self.assertEqual(handle(request, ["ready"])["error"]["code"], -32601)

    def test_invalid_arguments(self):
        invalid_inputs = [
            {"name": "catalog_lookup"},
            {"name": "catalog_lookup", "arguments": {}},
            {"name": "catalog_lookup", "arguments": {"sku": 42}},
            {"name": "catalog_lookup", "arguments": {"sku": "DEMO-1", "extra": True}},
        ]
        for params in invalid_inputs:
            with self.subTest(params=params):
                request = {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
                           "params": params}
                response = handle(request, ["ready"])
                self.assertNotIn("error", response)
                result = validate_response(response, 1)
                self.assertIs(result["isError"], True)
                self.assertEqual(result["content"][0]["type"], "text")
                self.assertTrue(result["content"][0]["text"])
        protocol_errors = [
            {"name": "unknown", "arguments": {}},
            {"name": "catalog_lookup", "arguments": []},
            {"name": "catalog_lookup", "arguments": None},
            {},
        ]
        for params in protocol_errors:
            with self.subTest(params=params):
                request = {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
                           "params": params}
                response = handle(request, ["ready"])
                self.assertNotIn("result", response)
                self.assertEqual(response["error"]["code"], -32602)

    def test_business_error(self):
        request = {
            "jsonrpc": "2.0", "id": 1, "method": "tools/call",
            "params": {"name": "catalog_lookup", "arguments": {"sku": "X"}},
        }
        self.assertTrue(handle(request, ["ready"])["result"]["isError"])


if __name__ == "__main__":
    unittest.main()

阅读 transcript 时先按 direction 区分发送方,再用 id 对齐请求和响应。不要按日志相邻位置推断生产系统中的调用关系:并发服务可能交错返回。实验里只有一个未完成请求,因此顺序可以保持简单;迁移到真实客户端时必须保留请求关联与取消语义。

把故障定位到一层后,再设计最小复现。例如已经成功返回 tools/list,随后调用报参数错误,就先检查 schema 和 arguments;此时重装运行时通常没有帮助。如果同一账号只能读取部分资源,应拿一条已知允许和一条已知拒绝的记录比较,避免把资源权限问题误判为连接故障。测试资料不要包含生产客户信息。

2026-10-20 用修正后的完整文件重新运行八组 unittest,普通模式与 -O 模式均通过;另经真实本地子进程管道核对十种工具调用输入,并保持正常流程七帧的 transcript。输入测试区分工具校验失败与协议错误,原有七种故障仍被覆盖。只有 slow 用例使用 0.3 秒截止时间,其他故障保留 2 秒等待,避免进程启动较慢时误报超时。测试仅证明这份教学 mock 的行为,不代表真实 Codex、SDK 或外部服务已互通,也不是完整 MCP 合规验证。

9. 接入完成后保留一份可重复的验收记录

给正式接入留一个短记录:使用哪个客户端和版本、从哪个目录启动、配置属于哪个范围、服务端点由谁维护、授权账号有哪些权限,以及哪次只读调用验证了什么事实。以后升级 CLI、调整代理或更换服务端版本,用同样的输入重复验证,差异就有可比较的依据。

图 07 / 概念示意
离线证据包含帮助核对、配置往返、8 组测试和 transcript;真实服务另行验收。
每条结论对应一份证据;实验通过的范围保持明确。
查看图示文字说明

离线证据包含帮助核对、配置往返、8 组测试和 transcript;真实服务另行验收。

离线部分可以帮助发现路径、stdout 和消息结构错误。真实环境还要核对工具目录、账号范围、组织限制与服务端行为,写操作必须有明确授权和失败恢复方式。首次接入不宜同时开放多个来源不明的服务;先把一个可解释、可撤销的连接维护好,再逐个增加。

本文按 2026-10-05 官方资料与本机 CLI 0.159.2 核对。没有执行 OAuth、创建生产凭据、连接真实 MCP 端点或运行模型任务。本页图示用于解释连接、权限和诊断流程。本页完整代码可以直接保存为对应文件;正式服务应使用维护中的官方 SDK,并针对实际业务完成验收。

10. 长结果返回成功后,还要核对哪一层完整

补充核对日期为 2026-10-14。前面的短库存结果只有一句话,容易逐字验收。换成一份长报告时,要把三份证据分开:服务端原始返回、宿主实际提供给模型的工具内容、最终回答。工具没有报错,只能说明这次调用没有以工具错误结束;回答语气肯定,也不能证明模型看到报告末尾。

先问一个能回查的小问题:合成报告 DEMO-STOCK 的 r1 版本里,SKU-120 的 pending_count 是多少?独立预期是 7,完整报告共有 120 条 SKU 记录。前 119 条都为 0,目标故意放在末尾。若回答只依据前几条就说“全部为零”,它没有完成这个问题的验收。

先查看当前客户端能够展示或导出的实际调用输入与输出,再与有权限读取的源记录对照。界面折叠的预览不一定等于模型收到的内容。若拿不到服务端原始返回或进入上下文的准确内容,就把那一层记为未核实,不能仅凭最终回答遗漏就判定 Codex 截断。服务端本身返回片段、查询范围不对、宿主输出预算和模型遗漏,都需要不同证据。

下面的手工片段仅用于训练这个核对动作;没有声称它来自真实工具调用,也没有模拟 Codex 的截断算法。120 行是教学数据规模,不是任何产品的 token 阈值。报告尾标和行数也是本例约定,不是 MCP 标准字段。

11. 确认输出预算后,只调整需要的工具

当前官方 MCP 文档列出 tools.<tool>.output_token_limit。它是单个工具输出的正整数 token 预算,覆盖模型默认的该工具输出截断预算,并位于标准 20% 序列化余量之前。配置参考没有给出适用于所有模型的统一数字默认值。这里的 6000 只是配置写法示例,不保证某份报告一定完整。demo_docs 与 read_report 都是虚构名称,不对应本次已接入的服务或工具。

output-budget.example.toml · 占位服务,仅展示字段位置
# 结构示例,不能原样连接;请核对实际服务名、工具名和端点
[mcp_servers.demo_docs]
url = "https://example.invalid/mcp"
enabled_tools = ["read_report"]
default_tools_approval_mode = "prompt"

[mcp_servers.demo_docs.tools.read_report]
output_token_limit = 6000

把服务名与工具名换成真实目录中核对过的名称,再编辑已有配置中对应的表;不要把同名表重复粘贴进去。如果服务已存在,只需要评估并修改它的单工具设置。保存后按客户端的正常流程重新启动或建立新会话,再以相同版本、相同输入核对实际输出。本文没有执行这一步,也没有验证某个已安装 Codex 版本如何裁剪内容。

token 不能直接换算成中文字数、UTF-8 字节数或记录条数,20% 余量也不是“额外保住某几行”的保证。tool_timeout_sec 控制等待时间,不能补回缺失正文。如果服务端原始返回已经少了目标记录,提高宿主预算也无法恢复它。只在证据指向输出预算且任务确实需要较长内容时评估该设置;调整后仍须核对完整性和上下文开销。

12. 用完整报告、手工片段和定向结果做一次对照

以下三个文本文件与 expected.json 放在同一空目录。它们全部是固定合成材料,不含账号、真实库存或凭据。report-full.txt 是完整输入;report-excerpt.txt 是人为编写、只有开头三条的缺尾片段;report-targeted.txt 只保留要回答的 SKU-120。expected.json 单独保存问题的已知事实,不从待验内容或模型回答反推。

report-full.txt
report=DEMO-STOCK;revision=r1;rows=120
SKU-001|pending_count=0
SKU-002|pending_count=0
SKU-003|pending_count=0
SKU-004|pending_count=0
SKU-005|pending_count=0
SKU-006|pending_count=0
SKU-007|pending_count=0
SKU-008|pending_count=0
SKU-009|pending_count=0
SKU-010|pending_count=0
SKU-011|pending_count=0
SKU-012|pending_count=0
SKU-013|pending_count=0
SKU-014|pending_count=0
SKU-015|pending_count=0
SKU-016|pending_count=0
SKU-017|pending_count=0
SKU-018|pending_count=0
SKU-019|pending_count=0
SKU-020|pending_count=0
SKU-021|pending_count=0
SKU-022|pending_count=0
SKU-023|pending_count=0
SKU-024|pending_count=0
SKU-025|pending_count=0
SKU-026|pending_count=0
SKU-027|pending_count=0
SKU-028|pending_count=0
SKU-029|pending_count=0
SKU-030|pending_count=0
SKU-031|pending_count=0
SKU-032|pending_count=0
SKU-033|pending_count=0
SKU-034|pending_count=0
SKU-035|pending_count=0
SKU-036|pending_count=0
SKU-037|pending_count=0
SKU-038|pending_count=0
SKU-039|pending_count=0
SKU-040|pending_count=0
SKU-041|pending_count=0
SKU-042|pending_count=0
SKU-043|pending_count=0
SKU-044|pending_count=0
SKU-045|pending_count=0
SKU-046|pending_count=0
SKU-047|pending_count=0
SKU-048|pending_count=0
SKU-049|pending_count=0
SKU-050|pending_count=0
SKU-051|pending_count=0
SKU-052|pending_count=0
SKU-053|pending_count=0
SKU-054|pending_count=0
SKU-055|pending_count=0
SKU-056|pending_count=0
SKU-057|pending_count=0
SKU-058|pending_count=0
SKU-059|pending_count=0
SKU-060|pending_count=0
SKU-061|pending_count=0
SKU-062|pending_count=0
SKU-063|pending_count=0
SKU-064|pending_count=0
SKU-065|pending_count=0
SKU-066|pending_count=0
SKU-067|pending_count=0
SKU-068|pending_count=0
SKU-069|pending_count=0
SKU-070|pending_count=0
SKU-071|pending_count=0
SKU-072|pending_count=0
SKU-073|pending_count=0
SKU-074|pending_count=0
SKU-075|pending_count=0
SKU-076|pending_count=0
SKU-077|pending_count=0
SKU-078|pending_count=0
SKU-079|pending_count=0
SKU-080|pending_count=0
SKU-081|pending_count=0
SKU-082|pending_count=0
SKU-083|pending_count=0
SKU-084|pending_count=0
SKU-085|pending_count=0
SKU-086|pending_count=0
SKU-087|pending_count=0
SKU-088|pending_count=0
SKU-089|pending_count=0
SKU-090|pending_count=0
SKU-091|pending_count=0
SKU-092|pending_count=0
SKU-093|pending_count=0
SKU-094|pending_count=0
SKU-095|pending_count=0
SKU-096|pending_count=0
SKU-097|pending_count=0
SKU-098|pending_count=0
SKU-099|pending_count=0
SKU-100|pending_count=0
SKU-101|pending_count=0
SKU-102|pending_count=0
SKU-103|pending_count=0
SKU-104|pending_count=0
SKU-105|pending_count=0
SKU-106|pending_count=0
SKU-107|pending_count=0
SKU-108|pending_count=0
SKU-109|pending_count=0
SKU-110|pending_count=0
SKU-111|pending_count=0
SKU-112|pending_count=0
SKU-113|pending_count=0
SKU-114|pending_count=0
SKU-115|pending_count=0
SKU-116|pending_count=0
SKU-117|pending_count=0
SKU-118|pending_count=0
SKU-119|pending_count=0
SKU-120|pending_count=7
END report=DEMO-STOCK;revision=r1;rows=120
report-excerpt.txt
report=DEMO-STOCK;revision=r1;rows=120
SKU-001|pending_count=0
SKU-002|pending_count=0
SKU-003|pending_count=0
report-targeted.txt
report=DEMO-STOCK;revision=r1;rows=1
SKU-120|pending_count=7
END report=DEMO-STOCK;revision=r1;rows=1
expected.json
{
  "report": "DEMO-STOCK",
  "revision": "r1",
  "full_rows": 120,
  "target_sku": "SKU-120",
  "target_pending_count": 7
}

完整报告应通过 120 条记录与末尾事实的核对;手工片段应报告资料不完整。若服务提供按标识读取或过滤参数,恢复时可在它支持的 schema 范围内缩小请求:

面向真实服务的请求意图,实际参数依工具 schema 填写
只读取 DEMO-STOCK 的 r1 版本中 SKU-120 这一条记录。返回报告标识、版本、SKU 和 pending_count,并给出对应原文。若工具没有提供该记录,就写“未核实”,不要从其他记录推断。

report-targeted.txt 展示这个缩小范围的合成预期:一条 SKU-120,值为 7,报告和版本保持不变。它足以核验这一个 SKU,不能据此宣布整份库存都已审阅。若服务没有定向查询能力,就按其文档采用其他受支持的读取方式;无法得到所需证据时保留“未核实”。不要自造参数,也不要把另一版报告当作同一次恢复。

13. 把缺尾当作未完成,再检查恢复后的原文

保存下面的 verify_report.py,使用 Python 3.9 或更高版本运行。它只读取刚保存的四份合成材料,不启动 MCP、Codex、服务器或模型,也不读取用户配置。校验器要求本例的首尾标记、记录集合与目标事实同时匹配;补一个假的 END 或返回错误数值也不能通过。

verify_report.py · 完整离线核对文件
"""Validate synthetic report evidence; no MCP, Codex, tokenizer, network or model."""
import json
from pathlib import Path
import re

ROOT = Path(__file__).resolve().parent
EXPECTED = json.loads((ROOT / "expected.json").read_text(encoding="utf-8"))


def validate(text, mode):
    if mode not in ("full", "targeted"):
        raise ValueError("unknown mode")
    count = EXPECTED["full_rows"] if mode == "full" else 1
    tag = "report={};revision={};rows={}".format(
        EXPECTED["report"], EXPECTED["revision"], count)
    lines = text.splitlines()
    if not lines or lines[0] != tag or lines[-1] != "END " + tag:
        raise ValueError("report boundary mismatch")
    records = {}
    for line in lines[1:-1]:
        match = re.fullmatch(r"(SKU-[0-9]{3})\|pending_count=([0-9]+)", line)
        if not match or match[1] in records:
            raise ValueError("invalid or duplicate record")
        records[match[1]] = int(match[2])
    wanted = {"SKU-{:03d}".format(i) for i in range(1, count + 1)}
    if mode == "targeted":
        wanted = {EXPECTED["target_sku"]}
    if set(records) != wanted:
        raise ValueError("record set mismatch")
    if records[EXPECTED["target_sku"]] != EXPECTED["target_pending_count"]:
        raise ValueError("target fact mismatch")
    return len(records)


def main():
    full = (ROOT / "report-full.txt").read_text(encoding="utf-8")
    excerpt = (ROOT / "report-excerpt.txt").read_text(encoding="utf-8")
    targeted = (ROOT / "report-targeted.txt").read_text(encoding="utf-8")
    # Expectations are fixed separately; never inferred from the text under test.
    cases = [
        ("full", full, "full", 120),
        ("manual_excerpt", excerpt, "full", "report boundary mismatch"),
        ("targeted", targeted, "targeted", 1),
        ("wrong_target_fact", targeted.replace("pending_count=7", "pending_count=8"),
         "targeted", "target fact mismatch"),
        ("forged_footer", excerpt + "END report=DEMO-STOCK;revision=r1;rows=120\n",
         "full", "record set mismatch"),
    ]
    results = []
    for name, text, mode, expected in cases:
        try:
            actual = validate(text, mode)
        except ValueError as exc:
            actual = str(exc)
        if actual != expected:
            raise RuntimeError("{}: expected {!r}; got {!r}".format(name, expected, actual))
        results.append({"case": name, "observed": actual})
    print(json.dumps({"passed": len(results), "cases": results}, indent=2))


if __name__ == "__main__":
    main()
在这些文件所在目录运行
python3 verify_report.py
python3 -O verify_report.py

本次两种运行方式均完成 5 个固定对照:完整资料返回 120;手工片段报 report boundary mismatch;定向资料返回 1;目标值改成 8 报 target fact mismatch;缺记录却补上尾标时报 record set mismatch。这些错误是预期拒绝,不是脚本运行失败。检查使用显式条件,-O 不会移除它们。

这个离线结果只证明合成资料的核对规则有效,没有测出真实 Codex 的 token 阈值,也没有证明真实宿主把哪些字送进模型。真实恢复应保留本次调用的输入、服务与工具名、报告版本,以及能够获得的实际返回证据;重新核对 SKU-120 的原文后,才能回答该条记录的问题。若只能看到最终回答,完整性仍不能由回答自证。相关日志只截取必要的非敏感内容,分享前先脱敏。

14. 参考资料

  1. Codex MCP 配置与支持范围
  2. OpenAI Docs MCP 快速开始
  3. 配置优先级与项目信任
  4. 完整配置字段参考
  5. 环境变量与日志
  6. Codex 权限边界
  7. 审批与安全
  8. MCP 2025-11-25 传输
  9. MCP 2025-11-25 生命周期
  10. MCP 2025-11-25 工具
  11. JSON-RPC 2.0

官方页面会持续更新。复制配置之前,核对你正在使用的 CLI 帮助与服务端文档;本文标明的版本和离线测试边界不随页面更新自动扩大。

Codex 要创建 mcp.json 吗?

普通直接连接写入 config.toml。其他编辑器和插件包可能使用 mcp.json 或 .mcp.json,迁移时按对应入口的官方格式处理。

list 有服务,为什么会话里没有工具?

继续检查当前主机和目录、信任状态、初始化、认证、工具筛选及组织策略。配置存在尚不能证明服务已返回可用工具。

HTTP 服务是否都需要 codex mcp login?

取决于服务认证方式。公开服务可匿名访问;支持 OAuth 的服务可用 login;有些服务要求从环境变量提供 Bearer token。

本地只读沙箱会限制远程 MCP 写入吗?

不能据此作保证。MCP 连接、服务端账号和工具权限需要分别配置与检查,远程操作仍须在授权范围内执行。

本文 mock 可以直接用于生产吗?

不能。它是固定协议版本的教学子集,缺少完整协议、认证、并发与安全防护。生产集成应使用维护中的官方 SDK 并针对业务验收。