Enhanced Interactive Feedback MCP Server

Instructions

Implementation Reference

• server.py:448-534 (handler)

  The primary handler for the 'execute_with_feedback' MCP tool. It optionally seeks user confirmation and error feedback using the interactive_feedback tool, executes shell commands via subprocess.run in the specified project directory, and returns comprehensive results including status, stdout/stderr, and any feedback.
  @self.app.tool() def execute_with_feedback( command: str, project_directory: str, require_confirmation: bool = True, language: str = "en" ) -> Dict[str, Any]: """ 执行命令并请求反馈 Args: command: 要执行的命令 project_directory: 项目目录 require_confirmation: 是否需要确认 language: 语言设置 Returns: 执行结果和反馈 """ try: if require_confirmation: # 请求执行确认 confirmation_message = self.template_manager.get_template( "confirmation", language, action=command ) confirmation_result = interactive_feedback( project_directory=project_directory, message=confirmation_message, feedback_type="confirmation", priority="medium", options=["Yes", "No", "Modify"], language=language ) if confirmation_result.get("response", "").lower() not in ["yes", "是"]: return { "status": "cancelled", "message": "Command execution cancelled by user", "feedback": confirmation_result } # 执行命令 result = subprocess.run( command, shell=True, cwd=project_directory, capture_output=True, text=True, timeout=300 ) # 如果有错误，请求错误处理反馈 if result.returncode != 0: error_message = self.template_manager.get_template( "error_handling", language, error=result.stderr ) error_feedback = interactive_feedback( project_directory=project_directory, message=error_message, feedback_type="error_report", priority="high", language=language, tags=["error", "command_execution"] ) return { "status": "error", "command": command, "return_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr, "error_feedback": error_feedback } return { "status": "success", "command": command, "return_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr } except Exception as e: logger.error(f"Error in execute_with_feedback: {e}") return {"error": str(e), "status": "error"}
• server.py:280-280 (registration)

  Calls the _register_tools method during server initialization, which contains the @app.tool() decorator registrations for all tools including execute_with_feedback.
  self._register_tools()
• server.py:287-535 (helper)

  The 'interactive_feedback' tool used internally by execute_with_feedback for confirmation and error feedback prompts.
  @self.app.tool() def interactive_feedback( project_directory: str, message: str, feedback_type: str = "question", priority: str = "medium", options: Optional[List[str]] = None, language: str = "en", timeout: int = 300, tags: Optional[List[str]] = None, user_id: str = "default" ) -> Dict[str, Any]: """ 增强型交互式反馈工具 Args: project_directory: 项目目录路径 message: 反馈消息或问题 feedback_type: 反馈类型 (question/confirmation/selection/code_review/suggestion/error_report) priority: 优先级 (low/medium/high/urgent) options: 预设选项列表 (用于选择题) language: 语言设置 (en/zh) timeout: 超时时间（秒） tags: 标签列表 user_id: 用户ID Returns: 包含用户反馈和分析信息的字典 """ try: # 验证参数 feedback_type_enum = FeedbackType(feedback_type) priority_enum = Priority(priority) # 生成唯一ID feedback_id = str(uuid.uuid4()) start_time = time.time() # 获取智能建议 patterns = self.analytics.get_feedback_patterns(project_directory) # 格式化消息 formatted_message = self._format_message( message, feedback_type_enum, language, options, patterns.get("suggestions", []) ) # 显示反馈界面 response = self._show_feedback_dialog( formatted_message, options, timeout, priority_enum, language ) response_time = time.time() - start_time # 保存反馈记录 feedback_record = FeedbackHistory( id=feedback_id, timestamp=datetime.now(), project_path=project_directory, feedback_type=feedback_type_enum, question=message, response=response, response_time=response_time, ai_summary="", # 可以通过AI生成摘要 tags=tags or [], priority=priority_enum, user_id=user_id ) self.analytics.save_feedback(feedback_record) # 生成实时建议 next_suggestions = self._generate_next_suggestions( project_directory, feedback_record ) return { "feedback_id": feedback_id, "response": response, "response_time": response_time, "patterns": patterns, "next_suggestions": next_suggestions, "status": "success" } except Exception as e: logger.error(f"Error in interactive_feedback: {e}") return { "error": str(e), "status": "error" } @self.app.tool() def get_feedback_analytics( project_directory: str, days: int = 30, user_id: str = "default" ) -> Dict[str, Any]: """ 获取反馈分析报告 Args: project_directory: 项目目录 days: 分析天数 user_id: 用户ID Returns: 分析报告 """ try: patterns = self.analytics.get_feedback_patterns(project_directory, days) return { "project": project_directory, "analysis_period_days": days, "patterns": patterns["patterns"], "suggestions": patterns["suggestions"], "generated_at": datetime.now().isoformat(), "status": "success" } except Exception as e: logger.error(f"Error in get_feedback_analytics: {e}") return {"error": str(e), "status": "error"} @self.app.tool() def create_feedback_template( template_name: str, template_content: Dict[str, str], template_type: str = "custom" ) -> Dict[str, Any]: """ 创建自定义反馈模板 Args: template_name: 模板名称 template_content: 模板内容 (多语言支持) template_type: 模板类型 Returns: 创建结果 """ try: templates_file = self.template_manager.templates_dir / f"{template_name}.yaml" template_data = { "type": template_type, "created_at": datetime.now().isoformat(), "content": template_content } with open(templates_file, 'w', encoding='utf-8') as f: yaml.dump(template_data, f, default_flow_style=False, allow_unicode=True) return { "template_name": template_name, "file_path": str(templates_file), "status": "created" } except Exception as e: logger.error(f"Error creating template: {e}") return {"error": str(e), "status": "error"} @self.app.tool() def execute_with_feedback( command: str, project_directory: str, require_confirmation: bool = True, language: str = "en" ) -> Dict[str, Any]: """ 执行命令并请求反馈 Args: command: 要执行的命令 project_directory: 项目目录 require_confirmation: 是否需要确认 language: 语言设置 Returns: 执行结果和反馈 """ try: if require_confirmation: # 请求执行确认 confirmation_message = self.template_manager.get_template( "confirmation", language, action=command ) confirmation_result = interactive_feedback( project_directory=project_directory, message=confirmation_message, feedback_type="confirmation", priority="medium", options=["Yes", "No", "Modify"], language=language ) if confirmation_result.get("response", "").lower() not in ["yes", "是"]: return { "status": "cancelled", "message": "Command execution cancelled by user", "feedback": confirmation_result } # 执行命令 result = subprocess.run( command, shell=True, cwd=project_directory, capture_output=True, text=True, timeout=300 ) # 如果有错误，请求错误处理反馈 if result.returncode != 0: error_message = self.template_manager.get_template( "error_handling", language, error=result.stderr ) error_feedback = interactive_feedback( project_directory=project_directory, message=error_message, feedback_type="error_report", priority="high", language=language, tags=["error", "command_execution"] ) return { "status": "error", "command": command, "return_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr, "error_feedback": error_feedback } return { "status": "success", "command": command, "return_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr } except Exception as e: logger.error(f"Error in execute_with_feedback: {e}") return {"error": str(e), "status": "error"}