Coverage for aceflow_mcp_server\tools.py: 84%

1"""AceFlow MCP Tools implementation."""

3from typing import Dict, Any, Optional, List

4import json

5import os

6import sys

7from pathlib import Path

8import shutil

9import datetime

11# Import core functionality

12from .core import ProjectManager, WorkflowEngine, TemplateManager

14# Import existing AceFlow functionality

15current_dir = Path(__file__).parent

16aceflow_scripts_dir = current_dir.parent.parent / "aceflow" / "scripts"

17sys.path.insert(0, str(aceflow_scripts_dir))

19try:

20 from utils.platform_compatibility import PlatformUtils, SafeFileOperations, EnhancedErrorHandler

21except ImportError:

22 # Fallback implementations if utils are not available

23 class PlatformUtils:

24 @staticmethod

25 def get_os_type(): return "unknown"

27 class SafeFileOperations:

28 @staticmethod

29 def write_text_file(path, content, encoding="utf-8"):

30 with open(path, 'w', encoding=encoding) as f:

31 f.write(content)

33 class EnhancedErrorHandler:

34 @staticmethod

35 def handle_file_error(error, context=""): return str(error)

38class AceFlowTools:

39 """AceFlow MCP Tools collection."""

41 def __init__(self):

42 """Initialize tools with necessary dependencies."""

43 self.platform_utils = PlatformUtils()

44 self.file_ops = SafeFileOperations()

45 self.error_handler = EnhancedErrorHandler()

46 self.project_manager = ProjectManager()

47 self.workflow_engine = WorkflowEngine()

48 self.template_manager = TemplateManager()

50 def aceflow_init(

51 self,

52 mode: str,

53 project_name: Optional[str] = None,

54 directory: Optional[str] = None

55 ) -> Dict[str, Any]:

56 """Initialize AceFlow project with specified mode.

58 Args:

59 mode: Workflow mode (minimal, standard, complete, smart)

60 project_name: Optional project name

61 directory: Optional target directory (defaults to current directory)

63 Returns:

64 Dict with success status, message, and project info

65 """

66 try:

67 # Validate mode

68 valid_modes = ["minimal", "standard", "complete", "smart"]

69 if mode not in valid_modes:

70 return {

71 "success": False,

72 "error": f"Invalid mode '{mode}'. Valid modes: {', '.join(valid_modes)}",

73 "message": "Mode validation failed"

74 }

76 # Determine target directory

77 if directory:

78 target_dir = Path(directory).resolve()

79 else:

80 target_dir = Path.cwd()

82 # Create directory if it doesn't exist

83 target_dir.mkdir(parents=True, exist_ok=True)

85 # Set project name

86 if not project_name:

87 project_name = target_dir.name

89 # Check if already initialized (unless forced)

90 aceflow_dir = target_dir / ".aceflow"

91 clinerules_file = target_dir / ".clinerules"

93 if aceflow_dir.exists() or clinerules_file.exists():

94 return {

95 "success": False,

96 "error": "Directory already contains AceFlow configuration",

97 "message": "Use --force flag to overwrite existing configuration"

98 }

100 # Initialize project structure

101 result = self._initialize_project_structure(target_dir, project_name, mode)

102

103 if result["success"]:

104 return {

105 "success": True,

106 "message": f"Project '{project_name}' initialized successfully in {mode} mode",

107 "project_info": {

108 "name": project_name,

109 "mode": mode,

110 "directory": str(target_dir),

111 "created_files": result.get("created_files", [])

112 }

113 }

114 else:

115 return result

116

117 except Exception as e:

118 return {

119 "success": False,

120 "error": str(e),

121 "message": "Failed to initialize project"

122 }

123

124 def _initialize_project_structure(self, target_dir: Path, project_name: str, mode: str) -> Dict[str, Any]:

125 """Initialize the complete project structure."""

126 created_files = []

127

128 try:

129 # Create .aceflow directory

130 aceflow_dir = target_dir / ".aceflow"

131 aceflow_dir.mkdir(exist_ok=True)

132 created_files.append(".aceflow/")

133

134 # Create aceflow_result directory

135 result_dir = target_dir / "aceflow_result"

136 result_dir.mkdir(exist_ok=True)

137 created_files.append("aceflow_result/")

138

139 # Create project state file

140 state_data = {

141 "project": {

142 "name": project_name,

143 "mode": mode.upper(),

144 "created_at": datetime.datetime.now().isoformat(),

145 "version": "3.0"

146 },

147 "flow": {

148 "current_stage": "user_stories" if mode != "minimal" else "implementation",

149 "completed_stages": [],

150 "progress_percentage": 0

151 },

152 "metadata": {

153 "total_stages": self._get_stage_count(mode),

154 "last_updated": datetime.datetime.now().isoformat()

155 }

156 }

157

158 state_file = aceflow_dir / "current_state.json"

159 with open(state_file, 'w', encoding='utf-8') as f:

160 json.dump(state_data, f, indent=2, ensure_ascii=False)

161 created_files.append(".aceflow/current_state.json")

162

163 # Create .clinerules file

164 clinerules_content = self._generate_clinerules(project_name, mode)

165 clinerules_file = target_dir / ".clinerules"

166 with open(clinerules_file, 'w', encoding='utf-8') as f:

167 f.write(clinerules_content)

168 created_files.append(".clinerules")

169

170 # Create template.yaml

171 template_content = self._generate_template_yaml(mode)

172 template_file = aceflow_dir / "template.yaml"

173 with open(template_file, 'w', encoding='utf-8') as f:

174 f.write(template_content)

175 created_files.append(".aceflow/template.yaml")

176

177 # Copy management scripts

178 script_files = ["aceflow-stage.py", "aceflow-validate.py", "aceflow-templates.py"]

179 for script in script_files:

180 source_path = aceflow_scripts_dir / script

181 if source_path.exists():

182 dest_path = target_dir / script

183 shutil.copy2(source_path, dest_path)

184 created_files.append(script)

185

186 # Create README

187 readme_content = self._generate_readme(project_name, mode)

188 readme_file = target_dir / "README_ACEFLOW.md"

189 with open(readme_file, 'w', encoding='utf-8') as f:

190 f.write(readme_content)

191 created_files.append("README_ACEFLOW.md")

192

193 return {

194 "success": True,

195 "created_files": created_files

196 }

197

198 except Exception as e:

199 return {

200 "success": False,

201 "error": str(e),

202 "message": "Failed to create project structure"

203 }

204

205 def _get_stage_count(self, mode: str) -> int:

206 """Get the number of stages for the given mode."""

207 stage_counts = {

208 "minimal": 3,

209 "standard": 8,

210 "complete": 12,

211 "smart": 10

212 }

213 return stage_counts.get(mode, 8)

214

215 def _generate_clinerules(self, project_name: str, mode: str) -> str:

216 """Generate .clinerules content."""

217 return f"""# AceFlow v3.0 - AI Agent 集成配置

218# 项目: {project_name}

219# 模式: {mode}

220# 初始化时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

221

222## 工作模式配置

223AceFlow模式: {mode}

224输出目录: aceflow_result/

225配置目录: .aceflow/

226项目名称: {project_name}

227

228## 核心工作原则

2291. 所有项目文档和代码必须输出到 aceflow_result/ 目录

2302. 严格按照 .aceflow/template.yaml 中定义的流程执行

2313. 每个阶段完成后更新项目状态文件

2324. 保持跨对话的工作记忆和上下文连续性

2335. 遵循AceFlow v3.0规范进行标准化输出

234

235## 质量标准

236- 代码质量: 遵循项目编码规范，注释完整

237- 文档质量: 结构清晰，内容完整，格式统一

238- 测试覆盖: 根据模式要求执行相应测试策略

239- 交付标准: 符合 aceflow-spec_v3.0.md 规范

240

241## 工具集成命令

242- python aceflow-validate.py: 验证项目状态和合规性

243- python aceflow-stage.py: 管理项目阶段和进度

244- python aceflow-templates.py: 管理模板配置

245

246记住: AceFlow是AI Agent的增强层，通过规范化输出和状态管理，实现跨对话的工作连续性。

247"""

248

249 def _generate_template_yaml(self, mode: str) -> str:

250 """Generate template.yaml content based on mode."""

251 templates = {

252 "minimal": """# AceFlow Minimal模式配置

253name: "Minimal Workflow"

254version: "3.0"

255description: "快速原型和概念验证工作流"

256

257stages:

258 - name: "implementation"

259 description: "快速实现核心功能"

260 required: true

261 - name: "test"

262 description: "基础功能测试"

263 required: true

264 - name: "demo"

265 description: "功能演示"

266 required: true

267

268quality_gates:

269 - stage: "implementation"

270 criteria: ["核心功能完成", "基本可运行"]

271 - stage: "test"

272 criteria: ["主要功能测试通过"]""",

273

274 "standard": """# AceFlow Standard模式配置

275name: "Standard Workflow"

276version: "3.0"

277description: "标准软件开发工作流"

278

279stages:

280 - name: "user_stories"

281 description: "用户故事分析"

282 required: true

283 - name: "task_breakdown"

284 description: "任务分解"

285 required: true

286 - name: "test_design"

287 description: "测试用例设计"

288 required: true

289 - name: "implementation"

290 description: "功能实现"

291 required: true

292 - name: "unit_test"

293 description: "单元测试"

294 required: true

295 - name: "integration_test"

296 description: "集成测试"

297 required: true

298 - name: "code_review"

299 description: "代码审查"

300 required: true

301 - name: "demo"

302 description: "功能演示"

303 required: true

304

305quality_gates:

306 - stage: "user_stories"

307 criteria: ["用户故事完整", "验收标准明确"]

308 - stage: "implementation"

309 criteria: ["代码质量合格", "功能完整"]

310 - stage: "unit_test"

311 criteria: ["测试覆盖率 > 80%", "所有测试通过"]""",

312

313 "complete": """# AceFlow Complete模式配置

314name: "Complete Workflow"

315version: "3.0"

316description: "完整企业级开发工作流"

317

318stages:

319 - name: "requirement_analysis"

320 description: "需求分析"

321 required: true

322 - name: "architecture_design"

323 description: "架构设计"

324 required: true

325 - name: "user_stories"

326 description: "用户故事分析"

327 required: true

328 - name: "task_breakdown"

329 description: "任务分解"

330 required: true

331 - name: "test_design"

332 description: "测试用例设计"

333 required: true

334 - name: "implementation"

335 description: "功能实现"

336 required: true

337 - name: "unit_test"

338 description: "单元测试"

339 required: true

340 - name: "integration_test"

341 description: "集成测试"

342 required: true

343 - name: "performance_test"

344 description: "性能测试"

345 required: true

346 - name: "security_review"

347 description: "安全审查"

348 required: true

349 - name: "code_review"

350 description: "代码审查"

351 required: true

352 - name: "demo"

353 description: "功能演示"

354 required: true

355

356quality_gates:

357 - stage: "architecture_design"

358 criteria: ["架构设计完整", "技术选型合理"]

359 - stage: "implementation"

360 criteria: ["代码质量优秀", "性能满足要求"]

361 - stage: "security_review"

362 criteria: ["安全检查通过", "无重大漏洞"]""",

363

364 "smart": """# AceFlow Smart模式配置

365name: "Smart Adaptive Workflow"

366version: "3.0"

367description: "AI增强的自适应工作流"

368

369stages:

370 - name: "project_analysis"

371 description: "AI项目复杂度分析"

372 required: true

373 - name: "adaptive_planning"

374 description: "自适应规划"

375 required: true

376 - name: "user_stories"

377 description: "用户故事分析"

378 required: true

379 - name: "smart_breakdown"

380 description: "智能任务分解"

381 required: true

382 - name: "test_generation"

383 description: "AI测试用例生成"

384 required: true

385 - name: "implementation"

386 description: "功能实现"

387 required: true

388 - name: "automated_test"

389 description: "自动化测试"

390 required: true

391 - name: "quality_assessment"

392 description: "AI质量评估"

393 required: true

394 - name: "optimization"

395 description: "性能优化"

396 required: true

397 - name: "demo"

398 description: "智能演示"

399 required: true

400

401ai_features:

402 - "复杂度智能评估"

403 - "动态流程调整"

404 - "自动化测试生成"

405 - "质量智能分析"

406

407quality_gates:

408 - stage: "project_analysis"

409 criteria: ["复杂度评估完成", "技术栈确定"]

410 - stage: "implementation"

411 criteria: ["AI代码质量检查通过", "性能指标达标"]"""

412 }

413

414 return templates.get(mode, templates["standard"])

415

416 def _generate_readme(self, project_name: str, mode: str) -> str:

417 """Generate README content."""

418 return f"""# {project_name}

419

420## AceFlow项目说明

421

422本项目使用AceFlow v3.0工作流管理系统，采用 **{mode.upper()}** 模式。

423

424### 项目信息

425- **项目名称**: {project_name}

426- **工作流模式**: {mode.upper()}

427- **初始化时间**: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

428- **AceFlow版本**: 3.0

429

430### 目录结构

431```

432{project_name}/

433├── .aceflow/ # AceFlow配置目录

434│ ├── current_state.json # 项目状态文件

435│ └── template.yaml # 工作流模板

436├── aceflow_result/ # 项目输出目录

437├── .clinerules # AI Agent工作配置

438├── aceflow-stage.py # 阶段管理脚本

439├── aceflow-validate.py # 项目验证脚本

440├── aceflow-templates.py # 模板管理脚本

441└── README_ACEFLOW.md # 本文件

442```

443

444### 快速开始

445

4461. **查看当前状态**

447 ```bash

448 python aceflow-stage.py --action status

449 ```

450

4512. **验证项目配置**

452 ```bash

453 python aceflow-validate.py

454 ```

455

4563. **推进到下一阶段**

457 ```bash

458 python aceflow-stage.py --action next

459 ```

460

461### 工作流程

462

463根据{mode}模式，项目将按以下阶段进行：

464

465{self._get_stage_description(mode)}

466

467### 注意事项

468

469- 所有项目文档和代码请输出到 `aceflow_result/` 目录

470- 使用AI助手时，确保.clinerules配置已加载

471- 每个阶段完成后，使用 `aceflow-stage.py` 更新状态

472- 定期使用 `aceflow-validate.py` 检查项目合规性

473

474### 帮助和支持

475

476如需帮助，请参考：

477- AceFlow官方文档

478- 项目状态文件: `.aceflow/current_state.json`

479- 工作流配置: `.aceflow/template.yaml`

480

481---

482*Generated by AceFlow v3.0 MCP Server*"""

483

484 def _get_stage_description(self, mode: str) -> str:

485 """Get stage descriptions for the mode."""

486 descriptions = {

487 "minimal": """1. **Implementation** - 快速实现核心功能

4882. **Test** - 基础功能测试

4893. **Demo** - 功能演示""",

490

491 "standard": """1. **User Stories** - 用户故事分析

4922. **Task Breakdown** - 任务分解

4933. **Test Design** - 测试用例设计

4944. **Implementation** - 功能实现

4955. **Unit Test** - 单元测试

4966. **Integration Test** - 集成测试

4977. **Code Review** - 代码审查

4988. **Demo** - 功能演示""",

499

500 "complete": """1. **Requirement Analysis** - 需求分析

5012. **Architecture Design** - 架构设计

5023. **User Stories** - 用户故事分析

5034. **Task Breakdown** - 任务分解

5045. **Test Design** - 测试用例设计

5056. **Implementation** - 功能实现

5067. **Unit Test** - 单元测试

5078. **Integration Test** - 集成测试

5089. **Performance Test** - 性能测试

50910. **Security Review** - 安全审查

51011. **Code Review** - 代码审查

51112. **Demo** - 功能演示""",

512

513 "smart": """1. **Project Analysis** - AI项目复杂度分析

5142. **Adaptive Planning** - 自适应规划

5153. **User Stories** - 用户故事分析

5164. **Smart Breakdown** - 智能任务分解

5175. **Test Generation** - AI测试用例生成

5186. **Implementation** - 功能实现

5197. **Automated Test** - 自动化测试

5208. **Quality Assessment** - AI质量评估

5219. **Optimization** - 性能优化

52210. **Demo** - 智能演示"""

523 }

524

525 return descriptions.get(mode, descriptions["standard"])

526

527 def aceflow_stage(

528 self,

529 action: str,

530 stage: Optional[str] = None

531 ) -> Dict[str, Any]:

532 """Manage project stages and workflow.

533

534 Args:

535 action: Stage management action (status, next, list, reset)

536 stage: Optional target stage name

537

538 Returns:

539 Dict with success status and stage information

540 """

541 try:

542 if action == "status":

543 result = self.workflow_engine.get_current_status()

544 return {

545 "success": True,

546 "action": action,

547 "result": result

548 }

549 elif action == "next":

550 result = self.workflow_engine.advance_to_next_stage()

551 return {

552 "success": True,

553 "action": action,

554 "result": result

555 }

556 elif action == "list":

557 stages = self.workflow_engine.list_all_stages()

558 return {

559 "success": True,

560 "action": action,

561 "result": {

562 "stages": stages

563 }

564 }

565 elif action == "reset":

566 result = self.workflow_engine.reset_project()

567 return {

568 "success": True,

569 "action": action,

570 "result": result

571 }

572 else:

573 return {

574 "success": False,

575 "error": f"Invalid action '{action}'. Valid actions: status, next, list, reset",

576 "message": "Action not supported"

577 }

578

579 except Exception as e:

580 return {

581 "success": False,

582 "error": str(e),

583 "message": f"Failed to execute stage action: {action}"

584 }

585

586 def aceflow_validate(

587 self,

588 mode: str = "basic",

589 fix: bool = False,

590 report: bool = False

591 ) -> Dict[str, Any]:

592 """Validate project compliance and quality.

593

594 Args:

595 mode: Validation mode (basic, complete)

596 fix: Auto-fix issues if possible

597 report: Generate detailed report

598

599 Returns:

600 Dict with validation results

601 """

602 try:

603 validator = self.project_manager.get_validator()

604 validation_result = validator.validate(mode=mode, auto_fix=fix, generate_report=report)

605

606 return {

607 "success": True,

608 "validation_result": {

609 "status": validation_result["status"],

610 "checks_total": validation_result["checks"]["total"],

611 "checks_passed": validation_result["checks"]["passed"],

612 "checks_failed": validation_result["checks"]["failed"],

613 "mode": mode,

614 "auto_fix_enabled": fix,

615 "report_generated": report

616 },

617 "message": f"Validation completed in {mode} mode"

618 }

619

620 except Exception as e:

621 return {

622 "success": False,

623 "error": str(e),

624 "message": "Validation failed"

625 }

626

627 def aceflow_template(

628 self,

629 action: str,

630 template: Optional[str] = None

631 ) -> Dict[str, Any]:

632 """Manage workflow templates.

633

634 Args:

635 action: Template action (list, apply, validate)

636 template: Optional template name

637

638 Returns:

639 Dict with template operation results

640 """

641 try:

642 if action == "list":

643 result = self.template_manager.list_templates()

644 return {

645 "success": True,

646 "action": action,

647 "result": {

648 "available_templates": result["available"],

649 "current_template": result["current"]

650 }

651 }

652 elif action == "apply":

653 if not template:

654 return {

655 "success": False,

656 "error": "Template name is required for apply action",

657 "message": "Please specify a template name"

658 }

659 result = self.template_manager.apply_template(template)

660 return {

661 "success": True,

662 "action": action,

663 "result": result

664 }

665 elif action == "validate":

666 result = self.template_manager.validate_current_template()

667 return {

668 "success": True,

669 "action": action,

670 "result": result

671 }

672 else:

673 return {

674 "success": False,

675 "error": f"Invalid action '{action}'. Valid actions: list, apply, validate",

676 "message": "Action not supported"

677 }

678

679 except Exception as e:

680 return {

681 "success": False,

682 "error": str(e),

683 "message": f"Template action failed: {action}"

684 }