#!/usr/bin/env python3 # MCP stdio server exposing one tool ("ask") that hands a prompt to a peer # workspace's Codex CLI over TCP and returns its final result — the Codex # counterpart to kilo/peer-mcp-bridge.py and claude/peer-mcp-bridge-claude.py. # # Unlike Claude Code, Codex actually has a native MCP server mode # (`codex mcp-server`) that speaks real, tool-capable MCP over stdio — the # closest of the three to kilo acp's ACP session, and simpler to bridge than # Claude Code's headless -p (no custom length-prefix framing needed: this is # already ordinary line-delimited JSON-RPC, same as ACP). So this script is a # genuine MCP *client* against the peer's codex-peer-listen.sh socat # listener: send the token, do a normal MCP initialize handshake, then call # the "codex" tool with approval-policy "never" and sandbox # "danger-full-access" — this workspace's own container is already the # isolation boundary, same justification as Claude Code's bypassPermissions # and kilo.jsonc's "permission": "allow" (see README). Model is deliberately # left unset in the tool call so it falls back to config.toml's own # model/model_provider default (MiniMax M3) rather than duplicating it here. # # The "codex" tool call can, in principle, still send server->client # approval requests (execCommandApproval / applyPatchApproval) mid-call — # not expected to actually fire with approval-policy "never", but handled # defensively (auto-"allow") the same way kilo's session/request_permission # is, rather than assuming it never will. import argparse import json import os import socket import sys import time TOOL_NAME = "ask" def read_stdin_messages(): for line in sys.stdin: line = line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError: continue def write_message(obj): sys.stdout.write(json.dumps(obj) + "\n") sys.stdout.flush() def send_response(msg_id, result=None, error=None): msg = {"jsonrpc": "2.0", "id": msg_id} if error is not None: msg["error"] = error else: msg["result"] = result write_message(msg) class PeerConnectionError(Exception): pass class CodexPeer: def __init__(self, host, port, timeout, token): self.host = host self.port = port self.timeout = timeout self.token = token def ask(self, prompt_text): deadline = time.time() + self.timeout try: sock = socket.create_connection((self.host, self.port), timeout=15) except OSError as e: raise PeerConnectionError(f"could not reach peer Codex bridge {self.host}:{self.port}: {e}") with sock: sock.settimeout(self.timeout) rf = sock.makefile("r") wf = sock.makefile("w") wf.write(self.token + "\n") wf.flush() def send(msg): wf.write(json.dumps(msg) + "\n") wf.flush() def recv(): line = rf.readline() if not line: return None return json.loads(line) def await_response(expected_id): while True: if time.time() > deadline: raise TimeoutError(f"peer did not finish within {self.timeout}s") obj = recv() if obj is None: raise PeerConnectionError("peer connection closed unexpectedly") # A genuine response to our own request has this id and # no "method" (requests/notifications always carry one, # even if a numeric id happens to collide with ours). if obj.get("id") == expected_id and "method" not in obj: if "error" in obj: raise PeerConnectionError(f"peer error: {obj['error']}") return obj if obj.get("method") in ("execCommandApproval", "applyPatchApproval"): send({"jsonrpc": "2.0", "id": obj["id"], "result": {"decision": "allow"}}) # else: ignore interleaved notifications (codex/event, etc.) send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "codex-peer-bridge", "version": "1.0.0"}}}) await_response(1) send({"jsonrpc": "2.0", "method": "notifications/initialized"}) send({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "codex", "arguments": { "prompt": prompt_text, "approval-policy": "never", "sandbox": "danger-full-access", }}}) resp = await_response(2) result = resp.get("result", {}) structured = result.get("structuredContent") or {} if structured.get("content"): return structured["content"] text_parts = [ block.get("text", "") for block in result.get("content", []) if block.get("type") == "text" ] reply = "".join(text_parts).strip() return reply or "(peer produced no text response)" def main(): ap = argparse.ArgumentParser() ap.add_argument("--host", required=True) ap.add_argument("--port", type=int, default=4297) ap.add_argument("--peer-name", required=True) ap.add_argument("--timeout", type=int, default=900) args = ap.parse_args() # Read from the environment (same as kilo's/Claude's peer-mcp-bridge # scripts) rather than a CLI flag, so the shared secret never ends up in # config.toml's mcp_servers.*.args array or in `ps` output. token = os.environ.get("PEER_BRIDGE_TOKEN", "") if not token: print(json.dumps({"jsonrpc": "2.0", "error": {"code": -32000, "message": "PEER_BRIDGE_TOKEN not set"}})) sys.exit(1) peer = CodexPeer(args.host, args.port, args.timeout, token) for msg in read_stdin_messages(): method = msg.get("method") msg_id = msg.get("id") if method == "initialize": client_version = msg.get("params", {}).get("protocolVersion", "2024-11-05") send_response(msg_id, result={ "protocolVersion": client_version, "capabilities": {"tools": {}}, "serverInfo": {"name": f"codex-peer-bridge-{args.peer_name}", "version": "1.0.0"}, }) elif method == "notifications/initialized": pass elif method == "tools/list": send_response(msg_id, result={"tools": [{ "name": TOOL_NAME, "description": ( f"Ask the '{args.peer_name}' workspace's Codex agent a question about its " "project, or hand it a task to work on. It runs with full autonomy in its " "own project directory (reads/edits/writes files, runs commands) and returns " "its final response once done. Each call starts a fresh Codex thread on the " "peer's side — include any needed context in the prompt itself." ), "inputSchema": { "type": "object", "properties": { "prompt": {"type": "string", "description": "The question or task for the peer agent."}, }, "required": ["prompt"], }, }]}) elif method == "tools/call": params = msg.get("params", {}) if params.get("name") != TOOL_NAME: send_response(msg_id, result={ "content": [{"type": "text", "text": f"Unknown tool: {params.get('name')}"}], "isError": True, }) continue prompt_text = params.get("arguments", {}).get("prompt", "") try: reply = peer.ask(prompt_text) send_response(msg_id, result={"content": [{"type": "text", "text": reply}], "isError": False}) except Exception as e: send_response(msg_id, result={ "content": [{"type": "text", "text": f"Peer request failed: {e}"}], "isError": True, }) elif msg_id is not None: send_response(msg_id, error={"code": -32601, "message": f"Method not found: {method}"}) # else: unrecognized notification, ignore if __name__ == "__main__": main()