Coverage for aceflow_mcp_server/server.py: 73%

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

2 

3import click 

4try: 

5 from fastmcp import FastMCP 

6except ImportError: 

7 # Mock implementation for testing 

8 class FastMCP: 

9 def __init__(self, name): 

10 self.name = name 

11 def add_tool(self, tool): pass 

12 def add_resource(self, resource): pass 

13 def add_prompt(self, prompt): pass 

14 def run(self, host="localhost", port=8000, log_level="INFO"): 

15 print(f"Mock MCP Server {self.name} running on {host}:{port}") 

16 

17from .tools import AceFlowTools 

18from .resources import AceFlowResources 

19from .prompts import AceFlowPrompts 

20 

21 

22class AceFlowMCPServer: 

23 """Main AceFlow MCP Server class.""" 

24 

25 def __init__(self): 

26 """Initialize the MCP server with all tools, resources, and prompts.""" 

27 self.mcp = FastMCP("AceFlow") 

28 self.tools = AceFlowTools() 

29 self.resources = AceFlowResources() 

30 self.prompts = AceFlowPrompts() 

31 

32 self._register_tools() 

33 self._register_resources() 

34 self._register_prompts() 

35 

36 def _register_tools(self): 

37 """Register all AceFlow tools.""" 

38 self.mcp.add_tool(self.tools.aceflow_init) 

39 self.mcp.add_tool(self.tools.aceflow_stage) 

40 self.mcp.add_tool(self.tools.aceflow_validate) 

41 self.mcp.add_tool(self.tools.aceflow_template) 

42 

43 def _register_resources(self): 

44 """Register all AceFlow resources.""" 

45 self.mcp.add_resource(self.resources.project_state) 

46 self.mcp.add_resource(self.resources.workflow_config) 

47 self.mcp.add_resource(self.resources.stage_guide) 

48 

49 def _register_prompts(self): 

50 """Register all AceFlow prompts.""" 

51 self.mcp.add_prompt(self.prompts.workflow_assistant) 

52 self.mcp.add_prompt(self.prompts.stage_guide) 

53 

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

55 """Start the MCP server.""" 

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

57 

58 

59@click.command() 

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

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

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

63@click.version_option(version="1.0.0") 

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

65 """Start AceFlow MCP Server.""" 

66 server = AceFlowMCPServer() 

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

68 

69 

70if __name__ == "__main__": 

71 main()