Merge pull request #159 from openqj/feat-cloud-mcp-server

feat(cloud): MCP 云端任务入口 + 上游 0.164 对齐配置

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 响应 JSON>}]，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())

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()

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"

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 的生产部署放在同一批决策里看，而不是悄悄混进去。