Coverage for session_buddy / cli.py: 50.00%

1#!/usr/bin/env python3 

2"""MCP Common CLI Factory for Session Management MCP Server. 

3 

4Replaces the custom Typer-based CLI with mcp-common's MCPServerCLIFactory 

5to provide standard lifecycle commands (start, stop, restart, status, health). 

6""" 

7 

8from __future__ import annotations 

9 

10import asyncio 

11import os 

12import warnings 

13 

14# Suppress transformers warnings about PyTorch/TensorFlow 

15os.environ["TRANSFORMERS_VERBOSITY"] = "error" 

16warnings.filterwarnings("ignore", message=".*PyTorch.*TensorFlow.*Flax.*") 

17 

18from mcp_common import MCPServerCLIFactory, MCPServerSettings, RuntimeHealthSnapshot 

19from session_buddy.tools.health_tools import get_health_status 

20from session_buddy.utils.runtime_snapshots import update_telemetry_counter 

21 

22 

23class SessionBuddySettings(MCPServerSettings): 

24 """Session Buddy specific MCP server settings extending MCPServerSettings.""" 

25 

26 # Session Buddy specific settings 

27 server_name: str = "session-buddy" 

28 

29 # HTTP server configuration 

30 http_port: int = 8678 

31 websocket_port: int = 8677 

32 

33 # Process management 

34 startup_timeout: int = 10 

35 shutdown_timeout: int = 10 

36 force_kill_timeout: int = 5 

37 

38 

39def start_server_handler() -> None: 

40 """Start handler that launches session_buddy.server.main() in HTTP streaming mode. 

41 

42 This function is called by the CLI factory when 'start' command is executed. 

43 """ 

44 from session_buddy.server import main 

45 

46 # Start server in HTTP mode with configured ports 

47 settings = SessionBuddySettings() 

48 

49 print("🚀 Starting Session Management MCP Server...") 

50 print(f"HTTP Port: {settings.http_port}") 

51 print(f"WebSocket Port: {settings.websocket_port}") 

52 

53 # Launch the server in HTTP streaming mode 

54 main(http_mode=True, http_port=settings.http_port) 

55 

56 

57def _read_running_pid(settings: MCPServerSettings) -> int | None: 

58 pid_path = settings.pid_path() 

59 if not pid_path.exists(): 

60 return None 

61 try: 

62 return int(pid_path.read_text().strip()) 

63 except (ValueError, OSError): 

64 return None 

65 

66 

67def _run_health_probe(settings: MCPServerSettings) -> RuntimeHealthSnapshot: 

68 pid = _read_running_pid(settings) 

69 health_state = asyncio.run(get_health_status(ready=False)) 

70 snapshot = RuntimeHealthSnapshot( 

71 orchestrator_pid=pid, 

72 watchers_running=pid is not None, 

73 activity_state={"health": health_state}, 

74 ) 

75 update_telemetry_counter(settings, name="health_probes", pid=pid) 

76 return snapshot 

77 

78 

79def create_session_buddy_cli() -> MCPServerCLIFactory: 

80 """Create the Session Buddy CLI using MCPServerCLIFactory. 

81 

82 Returns: 

83 Configured MCPServerCLIFactory instance ready for execution 

84 

85 """ 

86 # Initialize settings 

87 settings = SessionBuddySettings() 

88 

89 # Create the CLI factory with start handler 

90 return MCPServerCLIFactory( 

91 server_name=settings.server_name, 

92 settings=settings, 

93 start_handler=start_server_handler, 

94 health_probe_handler=lambda: _run_health_probe(settings), 

95 ) 

96 

97 

98def main() -> None: 

99 """Main entry point for the Session Buddy MCP CLI.""" 

100 # Create and configure the CLI 

101 cli_factory = create_session_buddy_cli() 

102 

103 # Create and run the CLI application 

104 app = cli_factory.create_app() 

105 

106 # Execute the CLI 

107 app() 

108 

109 

110if __name__ == "__main__": 

111 main()