Coverage for aceflow_mcp_server/tools.py: 80%
132 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-08-02 16:56 +0800
1"""AceFlow MCP Tools implementation."""
3try:
4 from fastmcp.tools import tool
5except ImportError:
6 # Mock implementation for testing
7 def tool(func):
8 return func
9from typing import Dict, Any, Optional, List
10import json
11import os
12import sys
13from pathlib import Path
14import shutil
15import datetime
17# Import existing AceFlow functionality
18current_dir = Path(__file__).parent
19aceflow_scripts_dir = current_dir.parent.parent / "aceflow" / "scripts"
20sys.path.insert(0, str(aceflow_scripts_dir))
22try:
23 from utils.platform_compatibility import PlatformUtils, SafeFileOperations, EnhancedErrorHandler
24except ImportError:
25 # Fallback implementations if utils are not available
26 class PlatformUtils:
27 @staticmethod
28 def get_os_type(): return "unknown"
30 class SafeFileOperations:
31 @staticmethod
32 def write_text_file(path, content, encoding="utf-8"):
33 with open(path, 'w', encoding=encoding) as f:
34 f.write(content)
36 class EnhancedErrorHandler:
37 @staticmethod
38 def handle_file_error(error, context=""): return str(error)
41class AceFlowTools:
42 """AceFlow MCP Tools collection."""
44 def __init__(self):
45 """Initialize tools with necessary dependencies."""
46 self.platform_utils = PlatformUtils()
47 self.file_ops = SafeFileOperations()
48 self.error_handler = EnhancedErrorHandler()
50 @tool
51 def aceflow_init(
52 self,
53 mode: str,
54 project_name: Optional[str] = None,
55 directory: Optional[str] = None
56 ) -> Dict[str, Any]:
57 """Initialize AceFlow project with specified mode.
59 Args:
60 mode: Workflow mode (minimal, standard, complete, smart)
61 project_name: Optional project name
62 directory: Optional target directory (defaults to current directory)
64 Returns:
65 Dict with success status, message, and project info
66 """
67 try:
68 # Validate mode
69 valid_modes = ["minimal", "standard", "complete", "smart"]
70 if mode not in valid_modes:
71 return {
72 "success": False,
73 "error": f"Invalid mode '{mode}'. Valid modes: {', '.join(valid_modes)}",
74 "message": "Mode validation failed"
75 }
77 # Determine target directory
78 if directory:
79 target_dir = Path(directory).resolve()
80 else:
81 target_dir = Path.cwd()
83 # Create directory if it doesn't exist
84 target_dir.mkdir(parents=True, exist_ok=True)
86 # Set project name
87 if not project_name:
88 project_name = target_dir.name
90 # Check if already initialized (unless forced)
91 aceflow_dir = target_dir / ".aceflow"
92 clinerules_file = target_dir / ".clinerules"
94 if aceflow_dir.exists() or clinerules_file.exists():
95 return {
96 "success": False,
97 "error": "Directory already contains AceFlow configuration",
98 "message": "Use --force flag to overwrite existing configuration"
99 }
101 # Initialize project structure
102 result = self._initialize_project_structure(target_dir, project_name, mode)
104 if result["success"]:
105 return {
106 "success": True,
107 "message": f"Project '{project_name}' initialized successfully in {mode} mode",
108 "project_info": {
109 "name": project_name,
110 "mode": mode,
111 "directory": str(target_dir),
112 "created_files": result.get("created_files", [])
113 }
114 }
115 else:
116 return result
118 except Exception as e:
119 return {
120 "success": False,
121 "error": str(e),
122 "message": "Failed to initialize project"
123 }
125 def _initialize_project_structure(self, target_dir: Path, project_name: str, mode: str) -> Dict[str, Any]:
126 """Initialize the complete project structure."""
127 created_files = []
129 try:
130 # Create .aceflow directory
131 aceflow_dir = target_dir / ".aceflow"
132 aceflow_dir.mkdir(exist_ok=True)
133 created_files.append(".aceflow/")
135 # Create aceflow_result directory
136 result_dir = target_dir / "aceflow_result"
137 result_dir.mkdir(exist_ok=True)
138 created_files.append("aceflow_result/")
140 # Create project state file
141 state_data = {
142 "project": {
143 "name": project_name,
144 "mode": mode.upper(),
145 "created_at": datetime.datetime.now().isoformat(),
146 "version": "3.0"
147 },
148 "flow": {
149 "current_stage": "user_stories" if mode != "minimal" else "implementation",
150 "completed_stages": [],
151 "progress_percentage": 0
152 },
153 "metadata": {
154 "total_stages": self._get_stage_count(mode),
155 "last_updated": datetime.datetime.now().isoformat()
156 }
157 }
159 state_file = aceflow_dir / "current_state.json"
160 with open(state_file, 'w', encoding='utf-8') as f:
161 json.dump(state_data, f, indent=2, ensure_ascii=False)
162 created_files.append(".aceflow/current_state.json")
164 # Create .clinerules file
165 clinerules_content = self._generate_clinerules(project_name, mode)
166 clinerules_file = target_dir / ".clinerules"
167 with open(clinerules_file, 'w', encoding='utf-8') as f:
168 f.write(clinerules_content)
169 created_files.append(".clinerules")
171 # Create template.yaml
172 template_content = self._generate_template_yaml(mode)
173 template_file = aceflow_dir / "template.yaml"
174 with open(template_file, 'w', encoding='utf-8') as f:
175 f.write(template_content)
176 created_files.append(".aceflow/template.yaml")
178 # Copy management scripts
179 script_files = ["aceflow-stage.py", "aceflow-validate.py", "aceflow-templates.py"]
180 for script in script_files:
181 source_path = aceflow_scripts_dir / script
182 if source_path.exists():
183 dest_path = target_dir / script
184 shutil.copy2(source_path, dest_path)
185 created_files.append(script)
187 # Create README
188 readme_content = self._generate_readme(project_name, mode)
189 readme_file = target_dir / "README_ACEFLOW.md"
190 with open(readme_file, 'w', encoding='utf-8') as f:
191 f.write(readme_content)
192 created_files.append("README_ACEFLOW.md")
194 return {
195 "success": True,
196 "created_files": created_files
197 }
199 except Exception as e:
200 return {
201 "success": False,
202 "error": str(e),
203 "message": "Failed to create project structure"
204 }
206 def _get_stage_count(self, mode: str) -> int:
207 """Get the number of stages for the given mode."""
208 stage_counts = {
209 "minimal": 3,
210 "standard": 8,
211 "complete": 12,
212 "smart": 10
213 }
214 return stage_counts.get(mode, 8)
216 def _generate_clinerules(self, project_name: str, mode: str) -> str:
217 """Generate .clinerules content."""
218 return f"""# AceFlow v3.0 - AI Agent 集成配置
219# 项目: {project_name}
220# 模式: {mode}
221# 初始化时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
223## 工作模式配置
224AceFlow模式: {mode}
225输出目录: aceflow_result/
226配置目录: .aceflow/
227项目名称: {project_name}
229## 核心工作原则
2301. 所有项目文档和代码必须输出到 aceflow_result/ 目录
2312. 严格按照 .aceflow/template.yaml 中定义的流程执行
2323. 每个阶段完成后更新项目状态文件
2334. 保持跨对话的工作记忆和上下文连续性
2345. 遵循AceFlow v3.0规范进行标准化输出
236## 质量标准
237- 代码质量: 遵循项目编码规范，注释完整
238- 文档质量: 结构清晰，内容完整，格式统一
239- 测试覆盖: 根据模式要求执行相应测试策略
240- 交付标准: 符合 aceflow-spec_v3.0.md 规范
242## 工具集成命令
243- python aceflow-validate.py: 验证项目状态和合规性
244- python aceflow-stage.py: 管理项目阶段和进度
245- python aceflow-templates.py: 管理模板配置
247记住: AceFlow是AI Agent的增强层，通过规范化输出和状态管理，实现跨对话的工作连续性。
248"""
250 def _generate_template_yaml(self, mode: str) -> str:
251 """Generate template.yaml content based on mode."""
252 templates = {
253 "minimal": """# AceFlow Minimal模式配置
254name: "Minimal Workflow"
255version: "3.0"
256description: "快速原型和概念验证工作流"
258stages:
259 - name: "implementation"
260 description: "快速实现核心功能"
261 required: true
262 - name: "test"
263 description: "基础功能测试"
264 required: true
265 - name: "demo"
266 description: "功能演示"
267 required: true
269quality_gates:
270 - stage: "implementation"
271 criteria: ["核心功能完成", "基本可运行"]
272 - stage: "test"
273 criteria: ["主要功能测试通过"]""",
275 "standard": """# AceFlow Standard模式配置
276name: "Standard Workflow"
277version: "3.0"
278description: "标准软件开发工作流"
280stages:
281 - name: "user_stories"
282 description: "用户故事分析"
283 required: true
284 - name: "task_breakdown"
285 description: "任务分解"
286 required: true
287 - name: "test_design"
288 description: "测试用例设计"
289 required: true
290 - name: "implementation"
291 description: "功能实现"
292 required: true
293 - name: "unit_test"
294 description: "单元测试"
295 required: true
296 - name: "integration_test"
297 description: "集成测试"
298 required: true
299 - name: "code_review"
300 description: "代码审查"
301 required: true
302 - name: "demo"
303 description: "功能演示"
304 required: true
306quality_gates:
307 - stage: "user_stories"
308 criteria: ["用户故事完整", "验收标准明确"]
309 - stage: "implementation"
310 criteria: ["代码质量合格", "功能完整"]
311 - stage: "unit_test"
312 criteria: ["测试覆盖率 > 80%", "所有测试通过"]""",
314 "complete": """# AceFlow Complete模式配置
315name: "Complete Workflow"
316version: "3.0"
317description: "完整企业级开发工作流"
319stages:
320 - name: "requirement_analysis"
321 description: "需求分析"
322 required: true
323 - name: "architecture_design"
324 description: "架构设计"
325 required: true
326 - name: "user_stories"
327 description: "用户故事分析"
328 required: true
329 - name: "task_breakdown"
330 description: "任务分解"
331 required: true
332 - name: "test_design"
333 description: "测试用例设计"
334 required: true
335 - name: "implementation"
336 description: "功能实现"
337 required: true
338 - name: "unit_test"
339 description: "单元测试"
340 required: true
341 - name: "integration_test"
342 description: "集成测试"
343 required: true
344 - name: "performance_test"
345 description: "性能测试"
346 required: true
347 - name: "security_review"
348 description: "安全审查"
349 required: true
350 - name: "code_review"
351 description: "代码审查"
352 required: true
353 - name: "demo"
354 description: "功能演示"
355 required: true
357quality_gates:
358 - stage: "architecture_design"
359 criteria: ["架构设计完整", "技术选型合理"]
360 - stage: "implementation"
361 criteria: ["代码质量优秀", "性能满足要求"]
362 - stage: "security_review"
363 criteria: ["安全检查通过", "无重大漏洞"]""",
365 "smart": """# AceFlow Smart模式配置
366name: "Smart Adaptive Workflow"
367version: "3.0"
368description: "AI增强的自适应工作流"
370stages:
371 - name: "project_analysis"
372 description: "AI项目复杂度分析"
373 required: true
374 - name: "adaptive_planning"
375 description: "自适应规划"
376 required: true
377 - name: "user_stories"
378 description: "用户故事分析"
379 required: true
380 - name: "smart_breakdown"
381 description: "智能任务分解"
382 required: true
383 - name: "test_generation"
384 description: "AI测试用例生成"
385 required: true
386 - name: "implementation"
387 description: "功能实现"
388 required: true
389 - name: "automated_test"
390 description: "自动化测试"
391 required: true
392 - name: "quality_assessment"
393 description: "AI质量评估"
394 required: true
395 - name: "optimization"
396 description: "性能优化"
397 required: true
398 - name: "demo"
399 description: "智能演示"
400 required: true
402ai_features:
403 - "复杂度智能评估"
404 - "动态流程调整"
405 - "自动化测试生成"
406 - "质量智能分析"
408quality_gates:
409 - stage: "project_analysis"
410 criteria: ["复杂度评估完成", "技术栈确定"]
411 - stage: "implementation"
412 criteria: ["AI代码质量检查通过", "性能指标达标"]"""
413 }
415 return templates.get(mode, templates["standard"])
417 def _generate_readme(self, project_name: str, mode: str) -> str:
418 """Generate README content."""
419 return f"""# {project_name}
421## AceFlow项目说明
423本项目使用AceFlow v3.0工作流管理系统，采用 **{mode.upper()}** 模式。
425### 项目信息
426- **项目名称**: {project_name}
427- **工作流模式**: {mode.upper()}
428- **初始化时间**: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
429- **AceFlow版本**: 3.0
431### 目录结构
432```
433{project_name}/
434├── .aceflow/ # AceFlow配置目录
435│ ├── current_state.json # 项目状态文件
436│ └── template.yaml # 工作流模板
437├── aceflow_result/ # 项目输出目录
438├── .clinerules # AI Agent工作配置
439├── aceflow-stage.py # 阶段管理脚本
440├── aceflow-validate.py # 项目验证脚本
441├── aceflow-templates.py # 模板管理脚本
442└── README_ACEFLOW.md # 本文件
443```
445### 快速开始
4471. **查看当前状态**
448 ```bash
449 python aceflow-stage.py --action status
450 ```
4522. **验证项目配置**
453 ```bash
454 python aceflow-validate.py
455 ```
4573. **推进到下一阶段**
458 ```bash
459 python aceflow-stage.py --action next
460 ```
462### 工作流程
464根据{mode}模式，项目将按以下阶段进行：
466{self._get_stage_description(mode)}
468### 注意事项
470- 所有项目文档和代码请输出到 `aceflow_result/` 目录
471- 使用AI助手时，确保.clinerules配置已加载
472- 每个阶段完成后，使用 `aceflow-stage.py` 更新状态
473- 定期使用 `aceflow-validate.py` 检查项目合规性
475### 帮助和支持
477如需帮助，请参考：
478- AceFlow官方文档
479- 项目状态文件: `.aceflow/current_state.json`
480- 工作流配置: `.aceflow/template.yaml`
482---
483*Generated by AceFlow v3.0 MCP Server*"""
485 def _get_stage_description(self, mode: str) -> str:
486 """Get stage descriptions for the mode."""
487 descriptions = {
488 "minimal": """1. **Implementation** - 快速实现核心功能
4892. **Test** - 基础功能测试
4903. **Demo** - 功能演示""",
492 "standard": """1. **User Stories** - 用户故事分析
4932. **Task Breakdown** - 任务分解
4943. **Test Design** - 测试用例设计
4954. **Implementation** - 功能实现
4965. **Unit Test** - 单元测试
4976. **Integration Test** - 集成测试
4987. **Code Review** - 代码审查
4998. **Demo** - 功能演示""",
501 "complete": """1. **Requirement Analysis** - 需求分析
5022. **Architecture Design** - 架构设计
5033. **User Stories** - 用户故事分析
5044. **Task Breakdown** - 任务分解
5055. **Test Design** - 测试用例设计
5066. **Implementation** - 功能实现
5077. **Unit Test** - 单元测试
5088. **Integration Test** - 集成测试
5099. **Performance Test** - 性能测试
51010. **Security Review** - 安全审查
51111. **Code Review** - 代码审查
51212. **Demo** - 功能演示""",
514 "smart": """1. **Project Analysis** - AI项目复杂度分析
5152. **Adaptive Planning** - 自适应规划
5163. **User Stories** - 用户故事分析
5174. **Smart Breakdown** - 智能任务分解
5185. **Test Generation** - AI测试用例生成
5196. **Implementation** - 功能实现
5207. **Automated Test** - 自动化测试
5218. **Quality Assessment** - AI质量评估
5229. **Optimization** - 性能优化
52310. **Demo** - 智能演示"""
524 }
526 return descriptions.get(mode, descriptions["standard"])
528 @tool
529 def aceflow_stage(
530 self,
531 action: str,
532 stage: Optional[str] = None
533 ) -> Dict[str, Any]:
534 """Manage project stages and workflow.
536 Args:
537 action: Stage management action (status, next, list, reset)
538 stage: Optional target stage name
540 Returns:
541 Dict with success status and stage information
542 """
543 try:
544 # This is a placeholder - in real implementation, this would
545 # integrate with the existing aceflow-stage.py functionality
547 if action == "status":
548 return {
549 "success": True,
550 "action": action,
551 "result": {
552 "current_stage": "user_stories",
553 "progress": 25,
554 "completed_stages": [],
555 "next_stage": "task_breakdown"
556 }
557 }
558 elif action == "list":
559 return {
560 "success": True,
561 "action": action,
562 "result": {
563 "stages": [
564 "user_stories", "task_breakdown", "test_design",
565 "implementation", "unit_test", "integration_test",
566 "code_review", "demo"
567 ]
568 }
569 }
570 else:
571 return {
572 "success": False,
573 "error": f"Action '{action}' not yet implemented",
574 "message": "This tool is under development"
575 }
577 except Exception as e:
578 return {
579 "success": False,
580 "error": str(e),
581 "message": f"Failed to execute stage action: {action}"
582 }
584 @tool
585 def aceflow_validate(
586 self,
587 mode: str = "basic",
588 fix: bool = False,
589 report: bool = False
590 ) -> Dict[str, Any]:
591 """Validate project compliance and quality.
593 Args:
594 mode: Validation mode (basic, complete)
595 fix: Auto-fix issues if possible
596 report: Generate detailed report
598 Returns:
599 Dict with validation results
600 """
601 try:
602 # Placeholder implementation
603 return {
604 "success": True,
605 "validation_result": {
606 "status": "passed",
607 "checks_total": 10,
608 "checks_passed": 8,
609 "checks_failed": 2,
610 "warnings": 1,
611 "issues": [
612 "Missing documentation in aceflow_result/",
613 "Template validation incomplete"
614 ]
615 },
616 "message": "Validation completed with warnings"
617 }
619 except Exception as e:
620 return {
621 "success": False,
622 "error": str(e),
623 "message": "Validation failed"
624 }
626 @tool
627 def aceflow_template(
628 self,
629 action: str,
630 template: Optional[str] = None
631 ) -> Dict[str, Any]:
632 """Manage workflow templates.
634 Args:
635 action: Template action (list, apply, validate)
636 template: Optional template name
638 Returns:
639 Dict with template operation results
640 """
641 try:
642 if action == "list":
643 return {
644 "success": True,
645 "action": action,
646 "result": {
647 "available_templates": [
648 "minimal", "standard", "complete", "smart"
649 ],
650 "current_template": "standard"
651 }
652 }
653 else:
654 return {
655 "success": False,
656 "error": f"Action '{action}' not yet implemented",
657 "message": "This tool is under development"
658 }
660 except Exception as e:
661 return {
662 "success": False,
663 "error": str(e),
664 "message": f"Template action failed: {action}"
665 }