CLI Agent MCP

Instructions

Implementation Reference

• src/cli_agent_mcp/server.py:104-112 (registration)

  Registration of the "opencode" tool in the MCP server's @server.list_tools() handler, dynamically adding it to the available tools list if allowed.
  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), ) )
• src/cli_agent_mcp/handlers/cli.py:76-234 (handler)

  CLIHandler class: the MCP ToolHandler implementation for "opencode" and other CLI tools. The handle() method (109-234) executes the tool by validating args, building OpencodeParams, creating OpencodeInvoker, calling execute(), and formatting the response.
  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:87-106 (schema)

  TOOL_DESCRIPTIONS["opencode"]: The detailed description used for the tool schema in MCP registration.
  "opencode": """Run OpenCode CLI agent (full-stack development). NO SHARED MEMORY: - Cannot see messages/outputs from codex/gemini/claude. - 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: - Excellent at rapid prototyping and development tasks - Good at working with multiple frameworks and tools - Supports multiple AI providers (Anthropic, OpenAI, Google, etc.) BEST PRACTICES: - Specify agent type for specialized tasks (e.g., --agent build) - Use file attachments for context-heavy tasks Supports: file attachments, multiple agents (build, plan, etc.).""",
• src/cli_agent_mcp/tool_schema.py:278-298 (schema)

  OPENCODE_PROPERTIES: Input schema properties specific to the opencode tool (file attachments and agent selection). Added in create_tool_schema at lines 583-584.
  OPENCODE_PROPERTIES = { "file": { "type": "array", "items": {"type": "string"}, "default": [], "description": ( "Absolute paths to files to attach to the message. " "Use for: Source code files, configuration files, documentation. " "Example: ['/path/to/main.py', '/path/to/config.json']" ), }, "agent": { "type": "string", "default": "build", "description": ( "Agent type to use for the task. " "Common agents: 'build' (default, general development), 'plan' (planning). " "Example: 'build'" ), }, }
• src/cli_agent_mcp/shared/invokers/opencode.py:55-262 (helper)

  OpencodeInvoker: Core helper that constructs and executes the 'opencode run' CLI command, handles permissions, and processes output/errors specific to OpenCode.
  class OpencodeInvoker(CLIInvoker): """OpenCode CLI 调用器。 封装 OpenCode CLI 的调用逻辑，包括： - 命令行参数构建 - Permission 到环境变量映射 - 支持文件附加和 agent 选择 - 特殊的错误处理（stdout 输出，退出码 0） Example: invoker = OpencodeInvoker() result = await invoker.execute(OpencodeParams( prompt="Analyze this project", workspace=Path("/path/to/repo"), )) """ def __init__( self, opencode_path: str = "opencode", event_callback: EventCallback | None = None, parser: Any | None = None, ) -> None: """初始化 OpenCode 调用器。 Args: opencode_path: opencode 可执行文件路径，默认 "opencode" event_callback: 事件回调函数 parser: 自定义解析器 """ super().__init__(event_callback=event_callback, parser=parser) self._opencode_path = opencode_path @property def cli_type(self) -> CLIType: return CLIType.OPENCODE def build_command(self, params: CommonParams) -> list[str]: """构建 OpenCode CLI 命令。 Args: params: 调用参数 Returns: 命令行参数列表 """ cmd = [self._opencode_path, "run"] # JSON 输出格式（JSONL 流式输出） cmd.extend(["--format", "json"]) # 可选：模型（格式为 provider/model） if params.model: cmd.extend(["--model", params.model]) # 会话恢复 if params.session_id: cmd.extend(["--session", params.session_id]) # OpenCode 特有参数 if isinstance(params, OpencodeParams): # Agent 选择 if params.agent: cmd.extend(["--agent", params.agent]) # 附加文件 for file_path in params.file: cmd.extend(["--file", str(file_path.absolute())]) # Prompt 作为位置参数 cmd.append(params.prompt) return cmd def get_env(self, params: CommonParams) -> dict[str, str] | None: """获取环境变量覆盖。 OpenCode 使用环境变量 OPENCODE_PERMISSION 来设置权限。 注意：返回的 env 会完全覆盖子进程的环境变量，所以需要继承系统环境。 Args: params: 调用参数 Returns: 环境变量字典（包含系统环境），或 None 使用默认 """ # 继承系统环境变量 env = dict(os.environ) # Permission 映射到 OPENCODE_PERMISSION 环境变量 # OpenCode 的权限模型与其他 CLI 不同，使用 JSON 格式的配置 permission_config = self._build_permission_config(params.permission) if permission_config: env["OPENCODE_PERMISSION"] = json.dumps(permission_config) return env def _build_permission_config(self, permission: Permission) -> dict[str, Any]: """构建 OpenCode 权限配置。 Args: permission: 权限级别 Returns: OpenCode 权限配置字典 """ if permission == Permission.READ_ONLY: # 只读模式：禁止编辑和执行 return { "edit": "deny", "bash": "deny", "webfetch": "deny", } elif permission == Permission.WORKSPACE_WRITE: # 工作区写入模式：允许编辑，bash 需要确认 return { "edit": "allow", "bash": "ask", "webfetch": "ask", } else: # UNLIMITED # 无限制模式：允许所有操作 return { "edit": "allow", "bash": "allow", "webfetch": "allow", "external_directory": "allow", } @property def uses_stdin_prompt(self) -> bool: """OpenCode 使用位置参数而非 stdin 传递 prompt。""" return False def _extract_error_from_line(self, line: str) -> tuple[str, str] | None: """从非 JSON 行中提取 OpenCode 错误信息。 OpenCode 的错误以堆栈跟踪格式输出到 stdout。 我们识别主错误行（如 ProviderModelNotFoundError: ...）并返回。 Args: line: 非 JSON 行内容 Returns: (error_type, error_message) 元组，如果不是错误行则返回 None """ # 检查是否是主错误行（如 ProviderModelNotFoundError: ...） match = re.match(r'^(\w+Error):\s*(.*)$', line) if match: error_name = match.group(1) error_msg = match.group(2) or error_name return (error_name, error_msg) return None def _process_event(self, event: Any, params: CommonParams) -> None: """处理 OpenCode 特有的事件。 OpenCode 的 session_id 在事件的 sessionID 字段中。 """ super()._process_event(event, params) # 从事件中提取 session_id if not self._session_id: raw = event.raw session_id = raw.get("sessionID", "") if session_id: self._session_id = session_id def _check_execution_errors(self, stderr_content: str = "") -> None: """检查 OpenCode 特有的错误情况。 OpenCode 的特殊行为： - 错误输出到 stdout 或 stderr（取决于错误类型） - 退出码通常为 0（即使发生错误） - 错误格式是堆栈跟踪，不是 JSON 如果捕获到了错误但 _exit_error 为空（返回码为 0）， 则从 stderr 或 _captured_errors 中构建错误信息。 Args: stderr_content: 子进程的 stderr 输出内容 """ # 如果已经有错误（返回码非 0），不需要额外处理 if self._exit_error: return # 优先检查 stderr（opencode 的某些错误输出到 stderr） if stderr_content.strip(): error_msg = f"OpenCode error (exit code 0):\n{stderr_content.strip()}" self._exit_error = error_msg # 向 GUI 发送错误事件 if self._event_callback: self._send_error_event(error_msg, error_type="opencode_error") return # 如果 stderr 为空，检查 stdout 中捕获的错误 if self._captured_errors: error_msg = f"OpenCode error (exit code 0):\n" error_msg += "\n".join(self._captured_errors[-5:]) # 取最后 5 条 self._exit_error = error_msg # 向 GUI 发送错误事件 if self._event_callback: self._send_error_event(error_msg, error_type="opencode_error")