CLI Agent MCP

Instructions

Implementation Reference

• src/cli_agent_mcp/server.py:401-419 (registration)

  MCP server registration of the 'opencode' tool via @server.list_tools(), including conditional enabling and schema generation.
  @server.list_tools() async def list_tools() -> list[Tool]: """列出可用工具。""" tools = [] for cli_type in ["codex", "gemini", "claude", "opencode"]: if config.is_tool_allowed(cli_type): tools.append( Tool( name=cli_type, description=TOOL_DESCRIPTIONS[cli_type], inputSchema=create_tool_schema(cli_type), ) ) # DEBUG: 记录工具列表请求（通常是客户端初始化后的第一个调用） logger.debug( f"[MCP] list_tools called, returning {len(tools)} tools: " f"{[t.name for t in tools]}" ) return tools
• shared/invokers/opencode.py:55-287 (handler)

  Primary handler: OpencodeInvoker class implements the execute() logic by building 'opencode run' CLI command, setting OPENCODE_PERMISSION env, parsing JSONL stream with special error handling from stdout.
  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 # 错误累积器：用于收集多行堆栈跟踪 self._error_accumulator: list[str] = [] self._in_error_block = False @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 """ # 累积所有非 JSON 行（可能是错误堆栈跟踪的一部分） self._error_accumulator.append(line) # 检查是否是主错误行（如 ProviderModelNotFoundError: ...） match = re.match(r'^(\w+Error):\s*(.*)$', line) if match: error_name = match.group(1) error_msg = match.group(2) or error_name self._in_error_block = True return (error_name, error_msg) # 检查 throw 语句 if 'throw new' in line: self._in_error_block = True # 不返回，等待主错误行 # 检查是否是源代码行（带行号），表示错误开始 if re.match(r'^\d+\s*\|', line): self._in_error_block = True # 不返回，等待主错误行 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 # 检查是否有累积的错误（来自 stdout） if self._error_accumulator: # 合并累积的错误 full_error = '\n'.join(self._error_accumulator) self._captured_errors.append(full_error) self._error_accumulator = [] # 优先检查 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")
• src/cli_agent_mcp/server.py:116-132 (schema)

  Tool description used in MCP Tool registration for 'opencode'.
  "opencode": """Invoke OpenCode CLI agent for full-stack development. CAPABILITIES: - Excellent at rapid prototyping and development tasks - Good at working with multiple frameworks and tools - Supports multiple AI providers (Anthropic, OpenAI, Google, etc.) LIMITATIONS: - May need explicit model selection for best results - Permission system differs from other CLI agents BEST PRACTICES: - Use for: Rapid prototyping, multi-framework projects - 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/server.py:250-270 (schema)

  JSON Schema properties specific to 'opencode' tool: file attachments and agent selection.
  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'" ), }, }
• shared/parsers/opencode.py:39-199 (helper)

  Helper: OpencodeParser for parsing JSONL stream events from opencode CLI into unified event format for GUI and processing.
  class OpencodeParser: """OpenCode CLI 事件解析器。 维护解析状态，支持流式事件的 ID 关联。 Example: parser = OpencodeParser() for line in stream: event = parser.parse(json.loads(line)) if event: gui.push_event(event) """ def __init__(self) -> None: self.session_id: str | None = None self.model: str | None = None def parse(self, data: dict[str, Any]) -> UnifiedEvent: """解析单个 OpenCode 事件。 Args: data: 原始事件字典 Returns: 统一事件实例 """ event_type = data.get("type", "") timestamp = data.get("timestamp", time.time() * 1000) # OpenCode 使用毫秒时间戳，转换为秒 if timestamp > 10000000000: # 如果是毫秒 timestamp = timestamp / 1000 # 从事件中提取 sessionID session_id = data.get("sessionID", "") if session_id and not self.session_id: self.session_id = session_id base_kwargs = { "source": CLISource.OPENCODE, "timestamp": timestamp, "raw": data, "session_id": self.session_id, } # 分发到具体的解析方法 if event_type == "tool_use": return self._parse_tool_use(data, base_kwargs) elif event_type == "step_start": return self._parse_step_start(data, base_kwargs) elif event_type == "step_finish": return self._parse_step_finish(data, base_kwargs) elif event_type == "text": return self._parse_text(data, base_kwargs) elif event_type == "error": return self._parse_error(data, base_kwargs) else: # Fallback: 未识别的事件类型 return make_fallback_event(CLISource.OPENCODE, data) def _parse_tool_use( self, data: dict[str, Any], base: dict[str, Any] ) -> OperationEvent: """解析 tool_use 事件。""" part = data.get("part", {}) tool_name = part.get("tool", "unknown") state = part.get("state", {}) # 获取工具输入参数 input_data = state.get("input", {}) try: input_str = json.dumps(input_data, ensure_ascii=False, indent=2) except (TypeError, ValueError): input_str = str(input_data) # 获取工具输出 output = state.get("output", "") title = state.get("title", "") # 确定状态 status_str = state.get("status", "completed") if status_str == "completed": status = Status.SUCCESS elif status_str == "running": status = Status.RUNNING elif status_str == "failed" or status_str == "error": status = Status.FAILED else: status = Status.SUCCESS return OperationEvent( event_id=make_event_id("opencode", f"tool_{tool_name}"), operation_type=OperationType.TOOL, name=tool_name, input=input_str, output=output or title, status=status, metadata={"state": state, "title": title}, **base, ) def _parse_step_start( self, data: dict[str, Any], base: dict[str, Any] ) -> LifecycleEvent: """解析 step_start 事件。""" return LifecycleEvent( event_id=make_event_id("opencode", "step_start"), lifecycle_type="turn_start", status=Status.RUNNING, **base, ) def _parse_step_finish( self, data: dict[str, Any], base: dict[str, Any] ) -> LifecycleEvent: """解析 step_finish 事件。""" return LifecycleEvent( event_id=make_event_id("opencode", "step_finish"), lifecycle_type="turn_end", status=Status.SUCCESS, **base, ) def _parse_text( self, data: dict[str, Any], base: dict[str, Any] ) -> MessageEvent: """解析 text 事件。""" part = data.get("part", {}) text = part.get("text", "") time_info = part.get("time", {}) # 如果有 end 时间，说明是完整消息 is_delta = not time_info.get("end") return MessageEvent( event_id=make_event_id("opencode", "text"), content_type=ContentType.TEXT, role="assistant", text=text, is_delta=is_delta, **base, ) def _parse_error( self, data: dict[str, Any], base: dict[str, Any] ) -> SystemEvent: """解析 error 事件。""" error = data.get("error", {}) if isinstance(error, dict): message = error.get("message", error.get("name", "Unknown error")) if "data" in error and isinstance(error["data"], dict): message = error["data"].get("message", message) else: message = str(error) return SystemEvent( event_id=make_event_id("opencode", "error"), severity="error", message=message, **base, )