Coverage for aceflow_mcp_server\server.py: 75%

1"""AceFlow MCP Server implementation using FastMCP framework.""" 

2 

3import click 

4from fastmcp import FastMCP 

5from typing import Dict, Any, Optional 

6from .tools import AceFlowTools 

7from .resources import AceFlowResources 

8from .prompts import AceFlowPrompts 

9 

10# Create global FastMCP instance 

11mcp = FastMCP("AceFlow") 

12 

13# Initialize components 

14tools = AceFlowTools() 

15resources = AceFlowResources() 

16prompts = AceFlowPrompts() 

17 

18# Register tools with decorators 

19@mcp.tool 

20def aceflow_init( 

21 mode: str, 

22 project_name: Optional[str] = None, 

23 directory: Optional[str] = None 

24) -> Dict[str, Any]: 

25 """Initialize AceFlow project with specified mode.""" 

26 return tools.aceflow_init(mode, project_name, directory) 

27 

28@mcp.tool 

29def aceflow_stage( 

30 action: str, 

31 stage: Optional[str] = None 

32) -> Dict[str, Any]: 

33 """Manage project stages and workflow.""" 

34 return tools.aceflow_stage(action, stage) 

35 

36@mcp.tool 

37def aceflow_validate( 

38 mode: str = "basic", 

39 fix: bool = False, 

40 report: bool = False 

41) -> Dict[str, Any]: 

42 """Validate project compliance and quality.""" 

43 return tools.aceflow_validate(mode, fix, report) 

44 

45@mcp.tool 

46def aceflow_template( 

47 action: str, 

48 template: Optional[str] = None 

49) -> Dict[str, Any]: 

50 """Manage workflow templates.""" 

51 return tools.aceflow_template(action, template) 

52 

53# Register resources with decorators 

54@mcp.resource("aceflow://project/state/{project_id}") 

55def project_state(project_id: str = "current") -> str: 

56 """Get current project state.""" 

57 return resources.project_state(project_id) 

58 

59@mcp.resource("aceflow://workflow/config/{config_id}") 

60def workflow_config(config_id: str = "default") -> str: 

61 """Get workflow configuration.""" 

62 return resources.workflow_config(config_id) 

63 

64@mcp.resource("aceflow://stage/guide/{stage}") 

65def stage_guide(stage: str) -> str: 

66 """Get stage-specific guidance.""" 

67 return resources.stage_guide(stage) 

68 

69# Register prompts with decorators 

70@mcp.prompt 

71def workflow_assistant( 

72 task: Optional[str] = None, 

73 context: Optional[str] = None 

74) -> str: 

75 """Generate workflow assistance prompt.""" 

76 return prompts.workflow_assistant(task, context) 

77 

78@mcp.prompt 

79def stage_guide_prompt(stage: str) -> str: 

80 """Generate stage-specific guidance prompt.""" 

81 return prompts.stage_guide(stage) 

82 

83 

84class AceFlowMCPServer: 

85 """Main AceFlow MCP Server class.""" 

86 

87 def __init__(self): 

88 """Initialize the MCP server.""" 

89 self.mcp = mcp 

90 

91 def run(self, host: str = "localhost", port: int = 8000, log_level: str = "INFO"): 

92 """Start the MCP server.""" 

93 self.mcp.run(host=host, port=port, log_level=log_level) 

94 

95 

96@click.command() 

97@click.option('--host', default='localhost', help='Host to bind to') 

98@click.option('--port', default=8000, type=int, help='Port to bind to') 

99@click.option('--log-level', default='INFO', help='Log level') 

100@click.version_option(version="1.0.1") 

101def main(host: str, port: int, log_level: str): 

102 """Start AceFlow MCP Server.""" 

103 server = AceFlowMCPServer() 

104 server.run(host=host, port=port, log_level=log_level) 

105 

106 

107if __name__ == "__main__": 

108 main()