CLI Agent MCP

Instructions

Implementation Reference

• src/cli_agent_mcp/server.py:101-144 (registration)

  Registers the MCP tool named 'codex' (and 'codex_parallel') in the server's list_tools() method if enabled via config. Uses TOOL_DESCRIPTIONS and create_tool_schema.
  async def list_tools() -> list[Tool]: """列出可用工具。""" tools = [] for cli_type in ["codex", "gemini", "claude", "opencode", "banana", "image"]: if config.is_tool_allowed(cli_type): tools.append( Tool( name=cli_type, description=TOOL_DESCRIPTIONS[cli_type], inputSchema=create_tool_schema(cli_type), ) ) # 追加 *_parallel 工具（仅支持的 CLI 工具） if cli_type in PARALLEL_SUPPORTED_TOOLS: parallel_name = f"{cli_type}_parallel" parallel_desc = ( f"Run multiple {cli_type} tasks in parallel. " f"All tasks share workspace/permission/save_file. " f"Results are appended to save_file with XML wrappers " f"(<agent-output agent=... continuation_id=... task_note=... task_index=... status=...>). " f"Max 100 tasks. Model can be array: single element shared by all, or one per task." ) tools.append( Tool( name=parallel_name, description=parallel_desc, inputSchema=create_tool_schema(cli_type, is_parallel=True), ) ) # 添加 get_gui_url 工具 if gui_manager: tools.append( Tool( name="get_gui_url", description="Get the GUI dashboard URL. Returns the HTTP URL where the live event viewer is accessible.", inputSchema={"type": "object", "properties": {}, "required": []}, ) ) # DEBUG: 记录工具列表请求（通常是客户端初始化后的第一个调用） logger.debug( f"[MCP] list_tools called, returning {len(tools)} tools: " f"{[t.name for t in tools]}" ) return tools
• src/cli_agent_mcp/handlers/cli.py:76-234 (handler)

  CLIHandler class provides the core execution logic for the 'codex' tool (dispatched from server.call_tool). Builds CodexParams, creates CodexInvoker, executes the CLI process, formats response, handles save_file and debug.
  class CLIHandler(ToolHandler): """CLI 工具处理器（codex, gemini, claude, opencode）。""" def __init__(self, cli_type: str): """初始化 CLIHandler。 Args: cli_type: CLI 类型（codex, gemini, claude, opencode） """ self._cli_type = cli_type @property def name(self) -> str: return self._cli_type @property def description(self) -> str: from ..tool_schema import TOOL_DESCRIPTIONS return TOOL_DESCRIPTIONS.get(self._cli_type, "") def get_input_schema(self) -> dict[str, Any]: from ..tool_schema import create_tool_schema return create_tool_schema(self._cli_type) def validate(self, arguments: dict[str, Any]) -> str | None: prompt = arguments.get("prompt") workspace = arguments.get("workspace") if not prompt or not str(prompt).strip(): return "Missing required argument: 'prompt'" if not workspace: return "Missing required argument: 'workspace'" return None async def handle( self, arguments: dict[str, Any], ctx: ToolContext, ) -> list[TextContent]: """处理 CLI 工具调用。""" # 校验 error = self.validate(arguments) if error: return format_error_response(error) task_note = arguments.get("task_note", "") prompt = arguments.get("prompt", "") # 创建 invoker（per-request 隔离） event_callback = ctx.make_event_callback(self._cli_type, task_note, None) if ctx.gui_manager else None invoker = create_invoker(self._cli_type, event_callback=event_callback) # 立即推送用户 prompt 到 GUI ctx.push_user_prompt(self._cli_type, prompt, task_note) # 使用 helper 注入 report_mode 和 context_paths report_mode = arguments.get("report_mode", False) context_paths = arguments.get("context_paths", []) injected_prompt = inject_context_and_report_mode(prompt, context_paths, report_mode) arguments = {**arguments, "prompt": injected_prompt} # 构建参数 params = build_params(self._cli_type, arguments) try: # 执行（取消异常会直接传播，不会返回） result = await invoker.execute(params) # 获取参数 debug_enabled = ctx.resolve_debug(arguments) save_file_path = arguments.get("save_file", "") # 构建 debug_info（当 debug 开启时始终构建，包含 log_file） debug_info = None if debug_enabled: debug_info = FormatterDebugInfo( model=result.debug_info.model if result.debug_info else None, duration_sec=result.debug_info.duration_sec if result.debug_info else 0.0, message_count=result.debug_info.message_count if result.debug_info else 0, tool_call_count=result.debug_info.tool_call_count if result.debug_info else 0, input_tokens=result.debug_info.input_tokens if result.debug_info else None, output_tokens=result.debug_info.output_tokens if result.debug_info else None, cancelled=result.cancelled, log_file=ctx.config.log_file if ctx.config.log_debug else None, ) # 构建 ResponseData（直接使用 invoker 提取的统一数据） # 错误时也尽力返回已收集的内容和 session_id，方便客户端发送"继续" response_data = ResponseData( answer=result.agent_messages, # 即使失败也返回已收集的内容 session_id=result.session_id or "", thought_steps=result.thought_steps if not result.success else [], debug_info=debug_info, success=result.success, error=result.error, ) # 格式化响应 formatter = get_formatter() response = formatter.format( response_data, debug=debug_enabled, ) # DEBUG: 记录响应摘要 logger.debug( f"[MCP] call_tool response:\n" f" Tool: {self._cli_type}\n" f" Success: {result.success}\n" f" Response length: {len(response)} chars\n" f" Duration: {result.debug_info.duration_sec:.3f}s" if result.debug_info else "" ) # 保存到文件（如果指定） # NOTE: save_file 是权限限制的例外，它仅用于落盘分析记录结果， # 而非通用的文件写入能力。CLI agent 的实际文件操作仍受 permission 参数控制。 # 这是一个便捷功能，让编排器无需单独写文件来保存分析结果。 if save_file_path and result.success: try: file_content = formatter.format_for_file(response_data) # 添加 XML wrapper（如果启用） if arguments.get("save_file_with_wrapper", False): continuation_id = result.session_id or "" file_content = ( f'<agent-output agent="{self._cli_type}" continuation_id="{continuation_id}">\n' f'{file_content}\n' f'</agent-output>\n' ) # 追加或覆盖 file_path = Path(save_file_path) file_path.parent.mkdir(parents=True, exist_ok=True) if arguments.get("save_file_with_append_mode", False) and file_path.exists(): with file_path.open("a", encoding="utf-8") as f: f.write("\n" + file_content) logger.info(f"Appended output to: {save_file_path}") else: file_path.write_text(file_content, encoding="utf-8") logger.info(f"Saved output to: {save_file_path}") except Exception as e: logger.warning(f"Failed to save output to {save_file_path}: {e}") return [TextContent(type="text", text=response)] except anyio.get_cancelled_exc_class() as e: # 取消通知已由 invoker._send_cancel_event() 推送到 GUI # 直接 re-raise 让 MCP 框架处理 logger.info(f"Tool '{self._cli_type}' cancelled (type={type(e).__name__})") raise except asyncio.CancelledError as e: # 捕获 asyncio.CancelledError（可能与 anyio 不同） logger.info(f"Tool '{self._cli_type}' cancelled via asyncio.CancelledError") raise except Exception as e: logger.error(f"Tool '{self._cli_type}' error: {e}") return format_error_response(str(e))
• src/cli_agent_mcp/tool_schema.py:25-46 (schema)

  Detailed human-readable description for the 'codex' tool used during MCP tool registration.
  TOOL_DESCRIPTIONS = { "codex": """Run OpenAI Codex CLI agent (deep analysis / critical review). NO SHARED MEMORY: - Cannot see messages/outputs from gemini/claude/opencode. - Only sees: (1) this prompt, (2) files in context_paths, (3) its own history via continuation_id. CROSS-AGENT HANDOFF: - Small data: paste into prompt. - Large data: save_file -> context_paths -> prompt says "Read <file>". CAPABILITIES: - Strongest deep analysis and reflection abilities - Excellent at finding issues, edge cases, and potential bugs - Good at critical code review and architectural assessment BEST PRACTICES: - Be explicit about scope: "Only fix X, don't refactor Y" - Specify constraints: "Keep it simple, no new abstractions" Supports: image attachments.""",
• src/cli_agent_mcp/tool_schema.py:236-247 (schema)

  Codex-specific input schema properties (supports image array for visual context), merged into create_tool_schema('codex').
  CODEX_PROPERTIES = { "image": { "type": "array", "items": {"type": "string"}, "default": [], "description": ( "Absolute paths to image files for visual context. " "Use for: UI screenshots, error dialogs, design mockups. " "Example: ['/path/to/screenshot.png']" ), }, }
• src/cli_agent_mcp/shared/invokers/codex.py:37-175 (helper)

  CodexInvoker executes the actual 'codex exec' CLI subprocess, builds command line args, handles permissions/images/session resume, processes stdout events, invoked by CLIHandler.
  class CodexInvoker(CLIInvoker): """Codex CLI 调用器。 封装 Codex CLI 的调用逻辑，包括： - 命令行参数构建 - Permission 到 --sandbox 参数映射 - 图片附件支持 Example: invoker = CodexInvoker() result = await invoker.execute(CodexParams( prompt="Review this code", workspace=Path("/path/to/repo"), image=[Path("screenshot.png")], )) """ # Codex 特有：可忽略的警告消息（如果有实际内容则不视为错误） _IGNORABLE_WARNINGS = [ "Long conversations and multiple compactions", ] # Codex 特有：可忽略的错误消息（不触发致命错误） _IGNORABLE_ERRORS = [ "Reconnecting", # 网络重连消息 ] def __init__( self, codex_path: str = "codex", event_callback: EventCallback | None = None, parser: Any | None = None, ) -> None: """初始化 Codex 调用器。 Args: codex_path: codex 可执行文件路径，默认 "codex" event_callback: 事件回调函数 parser: 自定义解析器 """ super().__init__(event_callback=event_callback, parser=parser) self._codex_path = codex_path @property def cli_type(self) -> CLIType: return CLIType.CODEX def validate_params(self, params: CommonParams) -> None: """验证 Codex 特有参数。""" super().validate_params(params) if isinstance(params, CodexParams): # 验证图片路径 for img_path in params.image: if not Path(img_path).exists(): raise ValueError(f"Image file does not exist: {img_path}") def build_command(self, params: CommonParams) -> list[str]: """构建 Codex CLI 命令。 Args: params: 调用参数 Returns: 命令行参数列表 """ cmd = [self._codex_path, "exec"] # 工作目录 cmd.extend(["--cd", str(params.workspace.absolute())]) # Permission 映射到 sandbox 参数 sandbox_value = PERMISSION_MAP_CODEX.get(params.permission, "read-only") cmd.extend(["--sandbox", sandbox_value]) # 硬编码参数 cmd.append("--skip-git-repo-check") cmd.append("--json") # 可选：模型 if params.model: cmd.extend(["--model", params.model]) # Codex 特有：图片附件 if isinstance(params, CodexParams): for img_path in params.image: cmd.extend(["--image", str(Path(img_path).absolute())]) # 会话恢复 if params.session_id: cmd.append("resume") cmd.append(params.session_id) # Prompt 通过 stdin 传递（使用 -- 分隔） cmd.append("--") return cmd def _process_event(self, event: Any, params: CommonParams) -> None: """处理 Codex 特有的事件。 Codex 的 session_id 可能在 thread.started 事件中。 """ super()._process_event(event, params) if not self._session_id: raw = event.raw if raw.get("type") == "thread.started": thread_id = raw.get("thread_id", "") if thread_id: self._session_id = thread_id def _is_ignorable_error(self, error_msg: str) -> bool: """检查错误消息是否可忽略。 Codex 的重连消息等不应触发致命错误。 """ for pattern in self._IGNORABLE_ERRORS: if pattern in error_msg: return True return False def _check_execution_errors(self, stderr_content: str = "") -> None: """检查 Codex 执行错误，处理可忽略的警告。 Codex 有时会输出 "Long conversations and multiple compactions" 警告， 这不是真正的错误。如果有实际内容，应该返回内容而不是错误。 """ if not self._exit_error: return # 检查是否是可忽略的警告 for warning in self._IGNORABLE_WARNINGS: if warning in self._exit_error: # 如果有实际内容，清除错误 if self._final_answer and self._final_answer.strip(): self._exit_error = None return