From 972203c38fcee106212a3eed2947ad84dfdcb2e7 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 01:41:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(cloud):=20MCP=20=E4=BA=91=E7=AB=AF?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E5=85=A5=E5=8F=A3=20+=20=E4=B8=8A=E6=B8=B8?= =?UTF-8?q?=200.164=20=E5=AF=B9=E9=BD=90=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - cloud/mcp_server.py：stdio MCP 服务，把云端任务 REST 翻成 tools/list / tools/call（health、list_projects、list_tasks、 get_task、create_task、cancel_task、get_task_cost、 get_task_trace），底层只走已有 REST，不另开任务逻辑； 鉴权只认调用方自己的 CODEX_CLOUD_API_KEY（缺了启动失败， 绝不打印），服务端权限 403 原样透传；create_task 只传已知 字段，未知字段丢弃；REST 报错时 isError=true。 单测 test_mcp_server 9 项（假 transport，不联网、不碰真 key）。 - 上游对齐 #80 的可合部分：config.example.toml 加 [features] daemon_auto_start = false（防 managed daemon 绕过 版本 pin）+ docs/upstream-164-alignment.md 对齐表；换稳定版 二进制是另一次主机侧部署动作，本批不做、不谎称已升级。 --- cloud/mcp_server.py | 231 +++++++++++++++++++++++++++++++++ cloud/tests/test_mcp_server.py | 133 +++++++++++++++++++ config.example.toml | 10 ++ docs/upstream-164-alignment.md | 80 ++++++++++++ 4 files changed, 454 insertions(+) create mode 100644 cloud/mcp_server.py create mode 100644 cloud/tests/test_mcp_server.py create mode 100644 docs/upstream-164-alignment.md diff --git a/cloud/mcp_server.py b/cloud/mcp_server.py new file mode 100644 index 0000000..dabd46c --- /dev/null +++ b/cloud/mcp_server.py @@ -0,0 +1,231 @@ +"""云端任务管理的 MCP 入口：把 REST API 翻成 MCP tools。 + +方向说明（别跟任务内 MCP 搞混）：项目内 agent 用的 MCP 是 +`[mcp_servers.*]`（给 agent 接外部工具）；本文件是反方向—— +让外部 MCP 客户端（编辑器、agent）用 tools/list / tools/call +管云端任务，底层一律走已有的 REST API，不另开一套任务逻辑。 + +协议：stdio 上每行一个 JSON-RPC 2.0 请求，支持 initialize、 +ping、tools/list、tools/call；notifications/* 不回包（按 MCP +惯例）。tools/call 的结果一律是 content=[{type: text, +text: }]，REST 报错时 isError=true，文本里 +含 HTTP 状态与服务端 error，不伪造成功。 + +鉴权：只认调用方自己的 X-API-Key，从环境变量 +CODEX_CLOUD_API_KEY 读（缺了就启动失败，绝不读文件、 +绝不打印 key）；BASE 从 CODEX_CLOUD_BASE_URL 读，缺省 +https://api.diante.me。MCP 服务本身不存任何凭证、不建 +用户、不绕过服务端权限——服务端按 key 归属做 viewer/ +contributor 等权限校验，403 原样透传。 + +工具（只包最常用的七个，不包管理端）： + health、list_projects、list_tasks、get_task、create_task、 + cancel_task、get_task_cost、get_task_trace。 +create_task 只传 project_id + prompt + 可选 executor/model/ +effort/approval/priority，其余字段走服务端缺省；cancel_task +只做取消，不做真 kill 之外的任何事（真 kill 由 worker 的 +kill_requests 机制处理，不在这里另起一套）。 +""" + +import json +import os +import sys +import urllib.error +import urllib.parse +import urllib.request + +__all__ = ["MCPServer", "main"] + + +def _tool(name, description, properties, required): + return { + "name": name, + "description": description, + "inputSchema": { + "type": "object", + "properties": properties, + "required": required, + }, + } + + +TOOLS = [ + _tool("health", "云端控制面健康检查（不含任务数据）。", + {}, []), + _tool("list_projects", "列出当前 key 可见的项目。", + {}, []), + _tool("list_tasks", "列出某项目的任务（需 project_id）。", + {"project_id": {"type": "string"}}, ["project_id"]), + _tool("get_task", "取单个任务详情（状态、结果、成本字段等）。", + {"task_id": {"type": "string"}}, ["task_id"]), + _tool("create_task", "在某项目建一个任务（需 project_id 与 prompt）。", + {"project_id": {"type": "string"}, + "prompt": {"type": "string"}, + "executor": {"type": "string"}, + "model": {"type": "string"}, + "effort": {"type": "string"}, + "approval": {"type": "string"}, + "priority": {"type": "integer"}}, + ["project_id", "prompt"]), + _tool("cancel_task", "取消一个任务（标 interrupted，容器真杀由 worker 处理）。", + {"task_id": {"type": "string"}}, ["task_id"]), + _tool("get_task_cost", "取单个任务的成本明细。", + {"task_id": {"type": "string"}}, ["task_id"]), + _tool("get_task_trace", "取单个任务的链路条目（摘要，完整结果在任务详情）。", + {"task_id": {"type": "string"}}, ["task_id"]), +] + + +class MCPServer: + """无状态的 REST→MCP 翻译层；transport 可注入，单测用假 transport。""" + + def __init__(self, base_url, api_key, transport=None): + if not api_key: + raise ValueError("CODEX_CLOUD_API_KEY 未设置") + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.transport = transport or self._http + + def _http(self, method, path, body=None): + url = self.base_url + path + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request(url, data=data, method=method) + req.add_header("X-API-Key", self.api_key) + if data is not None: + req.add_header("Content-Type", "application/json") + try: + with urllib.request.urlopen(req, timeout=60) as resp: + raw = resp.read().decode("utf-8") or "{}" + return resp.status, json.loads(raw) + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8") or "{}" + try: + payload = json.loads(raw) + except ValueError: + payload = {"raw": raw[:500]} + return exc.code, payload + except Exception as exc: # noqa: BLE001 - 网络层一律转错误结果 + return 0, {"error": str(exc)[:500]} + + def call_tool(self, name, args): + """返回 (is_error, payload_dict)。未知工具、缺参一律 is_error。""" + args = args or {} + if name == "health": + return self._wrap(*self.transport("GET", "/health")) + if name == "list_projects": + return self._wrap(*self.transport("GET", "/projects")) + if name == "list_tasks": + if not args.get("project_id"): + return True, {"error": "project_id required"} + q = urllib.parse.urlencode({"project_id": args["project_id"]}) + return self._wrap(*self.transport("GET", "/tasks?" + q)) + if name == "get_task": + if not args.get("task_id"): + return True, {"error": "task_id required"} + return self._wrap(*self.transport( + "GET", "/tasks/" + args["task_id"])) + if name == "create_task": + if not args.get("project_id") or not args.get("prompt"): + return True, {"error": "project_id and prompt required"} + body = {"project_id": args["project_id"], + "prompt": args["prompt"]} + for key in ("executor", "model", "effort", "approval", + "priority"): + if args.get(key) is not None: + body[key] = args[key] + return self._wrap(*self.transport("POST", "/tasks", body)) + if name == "cancel_task": + if not args.get("task_id"): + return True, {"error": "task_id required"} + return self._wrap(*self.transport( + "POST", "/tasks/" + args["task_id"] + "/cancel", {})) + if name == "get_task_cost": + if not args.get("task_id"): + return True, {"error": "task_id required"} + return self._wrap(*self.transport( + "GET", "/tasks/" + args["task_id"] + "/cost")) + if name == "get_task_trace": + if not args.get("task_id"): + return True, {"error": "task_id required"} + return self._wrap(*self.transport( + "GET", "/tasks/" + args["task_id"] + "/trace")) + return True, {"error": "unknown tool: %s" % name} + + @staticmethod + def _wrap(status, payload): + if status == 0 or status >= 400: + return True, {"http_status": status, "error": payload} + return False, payload + + def handle(self, request): + """处理一个 JSON-RPC 请求，返回响应 dict 或 None（notification）。""" + if not isinstance(request, dict): + return {"jsonrpc": "2.0", "id": None, + "error": {"code": -32600, "message": "invalid request"}} + method = request.get("method", "") + req_id = request.get("id") + if method.startswith("notifications/"): + return None + if method == "initialize": + return {"jsonrpc": "2.0", "id": req_id, "result": { + "protocolVersion": request.get("params", {}).get( + "protocolVersion", "2024-11-05"), + "capabilities": {"tools": {}}, + "serverInfo": {"name": "codex-cloud-mcp", + "version": "0.1.0"}}} + if method == "ping": + return {"jsonrpc": "2.0", "id": req_id, "result": {}} + if method == "tools/list": + return {"jsonrpc": "2.0", "id": req_id, + "result": {"tools": TOOLS}} + if method == "tools/call": + params = request.get("params", {}) or {} + is_error, payload = self.call_tool( + params.get("name"), params.get("arguments") or {}) + return {"jsonrpc": "2.0", "id": req_id, "result": { + "content": [{"type": "text", + "text": json.dumps(payload, + ensure_ascii=False)}], + "isError": is_error}} + return {"jsonrpc": "2.0", "id": req_id, + "error": {"code": -32601, + "message": "method not found: %s" % method}} + + def serve(self, instream, outstream): + """stdio 主循环：每行一个请求，EOF 结束；坏行回 -32700 不炸循环。""" + for line in instream: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except ValueError: + outstream.write(json.dumps( + {"jsonrpc": "2.0", "id": None, + "error": {"code": -32700, + "message": "parse error"}}) + "\n") + outstream.flush() + continue + response = self.handle(request) + if response is not None: + outstream.write(json.dumps(response, + ensure_ascii=False) + "\n") + outstream.flush() + + +def main(argv=None): + argv = list(sys.argv[1:] if argv is None else argv) + base_url = os.environ.get("CODEX_CLOUD_BASE_URL", + "https://api.diante.me") + api_key = os.environ.get("CODEX_CLOUD_API_KEY", "") + try: + server = MCPServer(base_url, api_key) + except ValueError as exc: + sys.stderr.write("mcp_server: %s\n" % exc) + return 2 + server.serve(sys.stdin, sys.stdout) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cloud/tests/test_mcp_server.py b/cloud/tests/test_mcp_server.py new file mode 100644 index 0000000..2e8dc0a --- /dev/null +++ b/cloud/tests/test_mcp_server.py @@ -0,0 +1,133 @@ +"""cloud mcp_server 单测：协议与 REST 翻译，不联网、不碰真实 key。 + +transport 用假函数注入，返回 (status, payload)；测试里出现的 +key、URL 均为占位值。 +""" + +import io +import json +import sys +import unittest +from pathlib import Path + +try: + from cloud.mcp_server import MCPServer, TOOLS +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.mcp_server import MCPServer, TOOLS + + +class FakeTransport: + def __init__(self): + self.calls = [] + + def __call__(self, method, path, body=None): + self.calls.append((method, path, body)) + if path == "/health": + return 200, {"ok": True, "commit": "test"} + if path == "/projects": + return 200, {"projects": [{"id": "p1"}]} + if path == "/tasks?project_id=p1": + return 200, {"tasks": [{"id": "t1", "status": "completed"}]} + if path == "/tasks/t1": + return 200, {"id": "t1", "status": "completed"} + if path == "/tasks" and method == "POST": + return 201, {"id": "t2", "status": "queued"} + if path == "/tasks/t1/cancel": + return 200, {"id": "t1", "status": "interrupted"} + if path == "/tasks/t1/cost": + return 200, {"total_tokens": 10} + if path == "/tasks/t1/trace": + return 200, {"entries": []} + if path == "/tasks/nope": + return 404, {"error": "not found"} + return 400, {"error": "bad request"} + + +class MCPServerTestCase(unittest.TestCase): + def setUp(self): + self.transport = FakeTransport() + self.server = MCPServer("https://example.invalid", "test-key", + transport=self.transport) + + def test_requires_api_key(self): + with self.assertRaises(ValueError): + MCPServer("https://example.invalid", "") + + def test_initialize_and_tools_list(self): + resp = self.server.handle({"jsonrpc": "2.0", "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2024-11-05"}}) + self.assertEqual(resp["result"]["serverInfo"]["name"], + "codex-cloud-mcp") + resp = self.server.handle({"jsonrpc": "2.0", "id": 2, + "method": "tools/list"}) + names = [t["name"] for t in resp["result"]["tools"]] + self.assertEqual(names, [t["name"] for t in TOOLS]) + self.assertIn("create_task", names) + self.assertIn("cancel_task", names) + + def test_notification_returns_none(self): + self.assertIsNone(self.server.handle( + {"jsonrpc": "2.0", "method": "notifications/initialized"})) + + def test_unknown_method_and_tool(self): + resp = self.server.handle({"jsonrpc": "2.0", "id": 1, + "method": "resources/list"}) + self.assertEqual(resp["error"]["code"], -32601) + is_error, payload = self.server.call_tool("nope", {}) + self.assertTrue(is_error) + self.assertIn("unknown tool", payload["error"]) + + def test_list_and_get_translate_to_rest(self): + is_error, payload = self.server.call_tool( + "list_tasks", {"project_id": "p1"}) + self.assertFalse(is_error) + self.assertEqual(payload["tasks"][0]["id"], "t1") + self.assertEqual(self.transport.calls[0][:2], + ("GET", "/tasks?project_id=p1")) + is_error, payload = self.server.call_tool( + "get_task", {"task_id": "t1"}) + self.assertFalse(is_error) + self.assertEqual(payload["status"], "completed") + + def test_create_only_passes_known_fields(self): + is_error, payload = self.server.call_tool("create_task", { + "project_id": "p1", "prompt": "do it", "model": "m1", + "priority": 2, "injected": "drop-me"}) + self.assertFalse(is_error) + self.assertEqual(payload["id"], "t2") + method, path, body = self.transport.calls[-1] + self.assertEqual((method, path), ("POST", "/tasks")) + self.assertNotIn("injected", body) + self.assertEqual(body["priority"], 2) + + def test_rest_error_becomes_is_error(self): + resp = self.server.handle({"jsonrpc": "2.0", "id": 9, + "method": "tools/call", + "params": {"name": "get_task", + "arguments": {"task_id": "nope"}}}) + self.assertTrue(resp["result"]["isError"]) + payload = json.loads(resp["result"]["content"][0]["text"]) + self.assertEqual(payload["http_status"], 404) + + def test_missing_arg_is_error_not_rest_call(self): + before = len(self.transport.calls) + is_error, payload = self.server.call_tool("list_tasks", {}) + self.assertTrue(is_error) + self.assertEqual(len(self.transport.calls), before) + + def test_serve_loop_handles_bad_line(self): + instream = io.StringIO( + "not json\n" + '{"jsonrpc": "2.0", "id": 1, "method": "ping"}\n') + outstream = io.StringIO() + self.server.serve(instream, outstream) + lines = [json.loads(line) for line in outstream.getvalue().splitlines()] + self.assertEqual(lines[0]["error"]["code"], -32700) + self.assertEqual(lines[1]["id"], 1) + self.assertEqual(lines[1]["result"], {}) + + +if __name__ == "__main__": + unittest.main() diff --git a/config.example.toml b/config.example.toml index d255ca5..fae8c8e 100644 --- a/config.example.toml +++ b/config.example.toml @@ -17,6 +17,16 @@ approval_policy = "on-request" # [features] # multi_agent = true +# 宿主托管的 app-server 必须关掉上游的常驻后台服务自更新： +# 自 0.157 起交互式 codex 默认启动 managed daemon（daemon_auto_start +# 默认 true），它会把整包拷到 ~/.codex/packages/app-server-daemon/、 +# 5 分钟后首次检查、之后每小时检查更新并重启服务——这会绕过宿主 +# 对 codex 二进制版本的显式 pin，甚至在回合进行中中断任务。 +# 本项目由 start.sh / worker 显式拉起 app-server，不走 managed daemon， +# 所以这里必须显式关掉；只在你手动想用交互式 daemon 时再打开。 +[features] +daemon_auto_start = false + # MCP：给 agent 接外部工具（数据库、内部 API 等），按需取消注释。 # [mcp_servers.my-tool] # command = "npx" diff --git a/docs/upstream-164-alignment.md b/docs/upstream-164-alignment.md new file mode 100644 index 0000000..1d555ee --- /dev/null +++ b/docs/upstream-164-alignment.md @@ -0,0 +1,80 @@ +# 上游对齐：openai/codex 稳定版 0.164.0（核查日 2026-10-27） + +结论先说：**有更新，但本项目不追 alpha，只锚定稳定版。** + +## 基线与上游位置 + +- 本项目此前锚定：`codex-cli 0.164.0-alpha.9`（2026-10-25 实测，`protocol/` 的 + schema 就是从它用 `codex app-server generate-json-schema` 导出的）。 +- 上游稳定版：`0.164.0`，2026-10-23 发布，由 `0.164.0-alpha.11` 加一个发布提交 + 切出，相对 0.163.0 约 330 个提交。也就是说本地 alpha.9 比稳定版落后 + alpha.10、alpha.11 两个 alpha 切片加稳定提交，不是落后一个大版本。 +- 上游 alpha 线（不作为目标）：0.165.0-alpha.1（2026-10-23，含 + `reasoning_effort_override` 转 stable）→ 0.165.0-alpha.3 → 0.166.0-alpha.1。 + alpha 线 4 天内连跳两个小版本，追它等于每几天重 pin 一次， + 与宿主“版本由部署显式控制”的原则冲突，故只记录、不跟进。 +- 上游来源：https://github.com/openai/codex/releases + +## 稳定版 0.164.0 改动 → 本项目状态 + +| 上游改动 | 对宿主层的影响 | 结论 | +|---|---|---| +| `/presets` 十个可复用提示槽、`/model` 与 `/permissions` 行内参数、终端图片预览 | 都是 TUI 交互面，app-server 协议与宿主层不依赖它们；控制台已有模型/强度下拉（#59） | 不相关，无需改代码 | +| 恢复会话时找回压缩历史、恢复已保存的线程设置（不再被全局默认覆盖） | 宿主每次回合都显式传 model/effort/sandbox，`thread/resume` 本就存在；上游修的是“恢复后被重置”这一坑 | 已兼容，换稳定版二进制后需重测 resume 一项，不改代码 | +| 排队提示等挂起的权限选择解决后再继续 | 宿主后续消息本来就是排队语义（#65：排队不打断当前轮，跑完按序再跑；direct 优先于排队） | 已覆盖，语义与上游一致 | +| 后台服务磁盘清理：剪掉未用的 daemon 发布、给 daemon 日志设上限、修 daemon 更新器 socket | 本项目不用 managed daemon：`start.sh` 直接拉起 app-server，worker 也是显式拉起 | 需防“被动启用”：见下文配置改动 | +| 远程 Bash / Zsh 会话中的凭据掩码、企业 MCP 认证 | 宿主走中继 provider，不用企业 MCP；远程 exec-server 不是生产路径 | 不相关，列为观察项 | +| 沙箱加固：macOS 大小写变体元数据路径、可替换 symlink 定位可执行文件 | 生产是 Linux + Docker 外层隔离 + 容器内 danger-full-access，macOS 沙箱路径不适用 | 不相关 | +| `worktrees` / `realtime_conversation` 等 feature flag 被移除、功能保持开启 | 宿主的 git worktree 编排不依赖这些 flag | 无需改代码 | +| Bedrock 设置向导默认开启 | 宿主不走 Bedrock | 不相关 | +| Windows HTTPS / WebSocket 代理认证、CLI 多层 `-c` 覆盖保留 | 宿主跑在 Linux，Windows 路径不适用 | 不相关 | + +## 更早已覆盖的上游能力（不是本次新发现的缺口） + +- 0.159：app-server 客户端可从指定 item 分页拉线程历史。 + alpha.9 的 schema 里已有 `thread/turns/list`、`thread/items/list`、 + `ThreadItemsListCursor`，且 `ThreadReadParams` 明确写着“完整历史 + hydration 已废弃，优先元数据读取 + 分页”。宿主用的是实时事件流 + + `thread/list`，不做全量历史 hydration，这是有意的设计，不是缺口。 +- 0.163：运行中发后续消息 steering、中断同时停 Code Mode 单元、 + 并行只读工具。宿主 #65 的 followup/direct、#65 的 fork、host 的 + `turn/interrupt` 跨进程中断已覆盖对应语义。 +- 0.165 alpha：`reasoning_effort_override` 转 stable。宿主每任务已有 + effort 档位校验（low/medium/high/xhigh/max）并透传给 turn， + 不需要为这个 flag 改代码。 + +## 本批实际改动（只改宿主侧，不动官方二进制） + +1. `config.example.toml`：新增 `[features] daemon_auto_start = false`， + 并写明原因。自 0.157 起交互式 codex 默认启动 managed daemon， + 5 分钟后首次检查、每小时检查更新并重启服务，可在回合进行中 + 中断任务，还会把版本控制权从宿主手里拿走。宿主托管的机器上 + 绝不能让一次交互式 `codex` 启动悄悄换掉 pin 住的版本。 + 这是本次对齐里唯一“不改就会在未来某天出事”的项。 +2. `README.md` / `CHANGELOG.md`：把“锚定 alpha.9”改成“锚定稳定版 + 0.164.0，alpha 线只记录不追”，并说明 `protocol/` 的 schema + 仍以 alpha.9 导出为准——没有新二进制就声称 schema 已更新是假话。 +3. 本文件：作为 0.164.0 的对齐记录，后续每次上游稳定版发布时 + 按同一张表复核。 + +## 明确不做的事（诚实边界） + +- **不把 bin/ 或生产主机的 codex 二进制换成稳定版 0.164.0。** + bin/ 在 `.gitignore` 里，生产主机的 codex 是主机侧 npm 安装， + 都不在本仓库的提交里；换二进制是部署动作，需要在主机侧 + 显式执行并重跑 resume / goal / 成本实弹，不是改几行文档就能宣称的。 +- **不追 0.165 / 0.166 alpha。** 等它们出稳定版再按本文件复核。 +- **不重导 `protocol/` schema。** 本地二进制仍是 alpha.9， + 用它重导只会得到同一份 schema；等稳定版二进制到位后再重导并 + diff，diff 为空或只有新增字段时才能合并。 +- **不合并、不部署。** 本批留在隔离 worktree 的 `feat-upstream-164` + 分支上；主 checkout 当时是脏的 feat-repo-picker 分支，绝不能 + 从它推；生产仍在 0a862c1（#74），#75/#76/#77 也还没上生产， + 上游对齐与生产部署是两件分开的决策。 + +## 下一步（需要用户定） + +1. 是否把本批提 PR（只含文档 + 配置示例 + 本文件，不含二进制变更）？ +2. 是否另起一次主机侧部署，把 codex 从 alpha.9 升到稳定版 0.164.0， + 并重跑 resume / goal / 成本实弹？那一步需要主机部署凭证， + 且应与 #75/#76/#77 的生产部署放在同一批决策里看，而不是悄悄混进去。