CLI Agent MCP

Instructions

Implementation Reference

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

  Registers the 'claude' tool alongside other CLI agents in the MCP server's list_tools() method.
  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 provides the core execution logic for the 'claude' tool, building ClaudeParams, creating ClaudeInvoker, executing the CLI process, processing events via callback, formatting response, and handling save_file.
  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:249-276 (schema)

  Defines Claude-specific input schema properties (system_prompt, append_system_prompt, agent) used in create_tool_schema('claude').
  CLAUDE_PROPERTIES = { "system_prompt": { "type": "string", "default": "", "description": ( "Complete replacement for the default system prompt. " "Use only when you need full control over agent behavior. " "Prefer append_system_prompt for most cases." ), }, "append_system_prompt": { "type": "string", "default": "", "description": ( "Additional instructions appended to the default system prompt. " "Recommended way to customize behavior. " "Example: 'Focus on performance optimization, avoid adding new dependencies'" ), }, "agent": { "type": "string", "default": "", "description": ( "Specify an agent for the current session (overrides the default agent setting). " "Use predefined agent names configured in Claude Code settings." ), }, }
• src/cli_agent_mcp/tool_schema.py:66-85 (schema)

  Tool description for 'claude' used in MCP Tool registration.
  "claude": """Run Anthropic Claude CLI agent (code implementation). NO SHARED MEMORY: - Cannot see messages/outputs from codex/gemini/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 code writing and implementation abilities - Excellent at translating requirements into working code - Good at following patterns and conventions BEST PRACTICES: - Be explicit about target: "Replace old implementation completely" - Specify cleanup: "Remove deprecated code paths" Supports: system_prompt, append_system_prompt, agent parameter.""",
• src/cli_agent_mcp/shared/invokers/claude.py:38-154 (helper)

  ClaudeInvoker implements the low-level CLI invocation for 'claude': command building, parameter validation, stdin prompt feeding, stdout parsing, session management.
  class ClaudeInvoker(CLIInvoker): """Claude CLI 调用器。 封装 Claude CLI 的调用逻辑，包括： - 命令行参数构建 - Permission 到 --tools 参数映射 - system_prompt / append_system_prompt 支持 Example: invoker = ClaudeInvoker() result = await invoker.execute(ClaudeParams( prompt="Review authentication module", workspace=Path("/path/to/repo"), append_system_prompt="Focus on OWASP Top 10 vulnerabilities.", )) """ def __init__( self, claude_path: str = "claude", event_callback: EventCallback | None = None, parser: Any | None = None, ) -> None: """初始化 Claude 调用器。 Args: claude_path: claude 可执行文件路径，默认 "claude" event_callback: 事件回调函数 parser: 自定义解析器 """ super().__init__(event_callback=event_callback, parser=parser) self._claude_path = claude_path @property def cli_type(self) -> CLIType: return CLIType.CLAUDE def validate_params(self, params: CommonParams) -> None: """验证 Claude 特有参数。""" super().validate_params(params) if isinstance(params, ClaudeParams): # 不能同时指定 system_prompt 和 append_system_prompt if params.system_prompt and params.append_system_prompt: raise ValueError( "Cannot specify both system_prompt and append_system_prompt. " "Use system_prompt to completely override, or append_system_prompt to add to existing." ) def build_command(self, params: CommonParams) -> list[str]: """构建 Claude CLI 命令。 Args: params: 调用参数 Returns: 命令行参数列表 """ cmd = [self._claude_path] # 硬编码：非交互模式 cmd.append("-p") # 跳过权限确认（MCP 服务器场景下必需） cmd.append("--dangerously-skip-permissions") # 硬编码：流式 JSON 输出（需要 --verbose） cmd.extend(["--output-format", "stream-json"]) cmd.append("--verbose") # stream-json 在 -p 模式下需要 --verbose # 工作目录 cmd.extend(["--add-dir", str(params.workspace.absolute())]) # Permission 映射到 tools 参数 # read-only: 只允许读取类工具 # workspace-write: 允许编辑和 Bash # unlimited: 使用默认工具集 tools_value = PERMISSION_MAP_CLAUDE.get(params.permission, "Read,Grep,Glob") cmd.extend(["--tools", tools_value]) # 可选：模型 if params.model: cmd.extend(["--model", params.model]) # Claude 特有：系统提示词 if isinstance(params, ClaudeParams): if params.system_prompt: cmd.extend(["--system-prompt", params.system_prompt]) elif params.append_system_prompt: cmd.extend(["--append-system-prompt", params.append_system_prompt]) # Agent 选择 if params.agent: cmd.extend(["--agent", params.agent]) # 会话恢复 if params.session_id: cmd.extend(["--resume", params.session_id]) # Prompt 通过 stdin 传递，不作为命令行参数 return cmd def _process_event(self, event: Any, params: CommonParams) -> None: """处理 Claude 特有的事件。 Claude 的 session_id 在 system/init 事件中。 """ super()._process_event(event, params) if not self._session_id: raw = event.raw if raw.get("type") == "system" and raw.get("subtype") == "init": session_id = raw.get("session_id", "") if session_id: self._session_id = session_id