Coverage for session_mgmt_mcp/server_optimized.py: 0.00%

1#!/usr/bin/env python3 

2"""Optimized Session Management MCP Server. 

3 

4This is the refactored, modular version of the session management server. 

5It's organized into focused modules for better maintainability and performance. 

6""" 

7 

8import sys 

9from pathlib import Path 

10 

11# Add project root to Python path 

12project_root = Path(__file__).parent.parent 

13if str(project_root) not in sys.path: 

14 sys.path.insert(0, str(project_root)) 

15 

16# Lazy loading for FastMCP 

17try: 

18 from fastmcp import FastMCP 

19 

20 MCP_AVAILABLE = True 

21except ImportError: 

22 # Check if we're in a test environment 

23 if "pytest" in sys.modules or "test" in sys.argv[0].lower(): 

24 print( 

25 "Warning: FastMCP not available in test environment, using mock", 

26 file=sys.stderr, 

27 ) 

28 

29 # Create a minimal mock FastMCP for testing 

30 class MockFastMCP: 

31 def __init__(self, name) -> None: 

32 self.name = name 

33 self.tools = {} 

34 self.prompts = {} 

35 

36 def tool(self, *args, **kwargs): 

37 def decorator(func): 

38 return func 

39 

40 return decorator 

41 

42 def prompt(self, *args, **kwargs): 

43 def decorator(func): 

44 return func 

45 

46 return decorator 

47 

48 def run(self, *args, **kwargs) -> None: 

49 pass 

50 

51 FastMCP = MockFastMCP 

52 MCP_AVAILABLE = False 

53 else: 

54 print("FastMCP not available. Install with: uv add fastmcp", file=sys.stderr) 

55 sys.exit(1) 

56 

57# Initialize MCP server 

58mcp = FastMCP("session-mgmt-mcp") 

59 

60# Initialize logging 

61from session_mgmt_mcp.utils.logging import get_session_logger 

62 

63logger = get_session_logger() 

64 

65# Register modularized tools 

66from session_mgmt_mcp.tools import register_memory_tools, register_session_tools 

67 

68# Core session management tools 

69register_session_tools(mcp) 

70 

71# Memory and reflection tools 

72register_memory_tools(mcp) 

73 

74 

75# Permission management (simplified for now) 

76class SessionPermissionsManager: 

77 """Simplified session permissions manager.""" 

78 

79 def __init__(self) -> None: 

80 self.trusted_operations = set() 

81 self.auto_checkpoint = False 

82 self.checkpoint_frequency = 300 

83 

84 def is_operation_trusted(self, operation: str) -> bool: 

85 return operation in self.trusted_operations 

86 

87 def add_trusted_operation(self, operation: str) -> None: 

88 self.trusted_operations.add(operation) 

89 

90 

91# Global permissions manager 

92permissions_manager = SessionPermissionsManager() 

93 

94 

95@mcp.tool() 

96async def permissions(action: str = "status", operation: str | None = None) -> str: 

97 """Manage session permissions for trusted operations to avoid repeated prompts. 

98 

99 Args: 

100 action: Action to perform: status (show current), trust (add operation), revoke_all (reset) 

101 operation: Operation to trust (required for 'trust' action) 

102 

103 """ 

104 output = [] 

105 output.append("🔐 Session Permissions Management") 

106 output.append("=" * 40) 

107 

108 if action == "status": 

109 if permissions_manager.trusted_operations: 

110 output.append( 

111 f"✅ {len(permissions_manager.trusted_operations)} trusted operations:", 

112 ) 

113 for op in sorted(permissions_manager.trusted_operations): 

114 output.append(f" • {op}") 

115 output.append( 

116 "\n💡 These operations will not prompt for permission in future sessions", 

117 ) 

118 else: 

119 output.append("⚠️ No operations are currently trusted") 

120 output.append( 

121 "💡 Operations will be automatically trusted on first successful use", 

122 ) 

123 

124 output.append("\n🛠️ Common Operations That Can Be Trusted:") 

125 output.append(" • UV Package Management - uv sync, pip operations") 

126 output.append(" • Git Repository Access - git status, commit, push") 

127 output.append(" • Project File Access - reading/writing project files") 

128 output.append(" • Subprocess Execution - running build tools, tests") 

129 output.append(" • Claude Directory Access - accessing ~/.claude/") 

130 

131 elif action == "trust": 

132 if not operation: 

133 output.append("❌ Error: 'operation' parameter required for 'trust' action") 

134 output.append( 

135 "💡 Example: permissions with action='trust' and operation='uv_package_management'", 

136 ) 

137 else: 

138 permissions_manager.add_trusted_operation(operation) 

139 output.append( 

140 f"✅ Operation '{operation}' has been added to trusted operations", 

141 ) 

142 output.append("💡 This operation will no longer require permission prompts") 

143 

144 elif action == "revoke_all": 

145 count = len(permissions_manager.trusted_operations) 

146 permissions_manager.trusted_operations.clear() 

147 output.append(f"🗑️ Revoked {count} trusted operations") 

148 output.append("💡 All operations will now require permission prompts") 

149 

150 else: 

151 output.append(f"❌ Unknown action: {action}") 

152 output.append("💡 Valid actions: status, trust, revoke_all") 

153 

154 return "\n".join(output) 

155 

156 

157# Additional simplified tools for compatibility 

158@mcp.tool() 

159async def auto_compact() -> str: 

160 """Automatically trigger conversation compaction with intelligent summary.""" 

161 output = [] 

162 output.append("🗜️ Auto-Compaction Feature") 

163 output.append("=" * 30) 

164 output.append("✅ This feature is now handled by the modular memory system") 

165 output.append("💡 Memory optimization occurs automatically in the background") 

166 return "\n".join(output) 

167 

168 

169@mcp.tool() 

170async def quality_monitor() -> str: 

171 """Phase 3: Proactive quality monitoring with early warning system.""" 

172 output = [] 

173 output.append("📊 Quality Monitoring") 

174 output.append("=" * 25) 

175 output.append( 

176 "✅ Quality monitoring is integrated into the session management system", 

177 ) 

178 output.append("💡 Use the 'status' tool to get current quality metrics") 

179 output.append("💡 Use the 'checkpoint' tool for comprehensive quality assessment") 

180 return "\n".join(output) 

181 

182 

183# Server startup 

184def run_server() -> None: 

185 """Run the optimized MCP server.""" 

186 try: 

187 logger.info("Starting optimized session-mgmt-mcp server") 

188 

189 # Log the modular structure 

190 logger.info( 

191 "Modular components loaded", 

192 session_tools=True, 

193 memory_tools=True, 

194 git_operations=True, 

195 logging_utils=True, 

196 ) 

197 

198 if MCP_AVAILABLE: 

199 mcp.run() 

200 else: 

201 logger.warning("Running in mock mode - FastMCP not available") 

202 

203 except Exception as e: 

204 logger.exception("Server startup failed", error=str(e)) 

205 sys.exit(1) 

206 

207 

208if __name__ == "__main__": 

209 run_server()