Coverage for session_mgmt_mcp/tools/session_tools.py: 0.00%

1#!/usr/bin/env python3 

2"""Session management MCP tools. 

3 

4This module provides tools for managing Claude session lifecycle including 

5initialization, checkpoints, and cleanup. 

6""" 

7 

8import shutil 

9import subprocess 

10from pathlib import Path 

11 

12from session_mgmt_mcp.core import SessionLifecycleManager 

13from session_mgmt_mcp.utils.logging import get_session_logger 

14 

15# Global session manager 

16session_manager = SessionLifecycleManager() 

17logger = get_session_logger() 

18 

19 

20# Tool implementations 

21async def _init_impl(working_directory: str | None = None) -> str: 

22 """Implementation for init tool.""" 

23 output = [] 

24 output.append("🚀 Claude Session Initialization via MCP Server") 

25 output.append("=" * 60) 

26 

27 try: 

28 # Use the session manager for initialization 

29 result = await session_manager.initialize_session(working_directory) 

30 

31 if result["success"]: 

32 output.append(f"📁 Current project: {result['project']}") 

33 output.append(f"📂 Working directory: {result['working_directory']}") 

34 output.append(f"🏠 Claude directory: {result['claude_directory']}") 

35 output.append(f"📊 Initial quality score: {result['quality_score']}/100") 

36 

37 # Add project context info 

38 context = result["project_context"] 

39 context_items = sum(1 for detected in context.values() if detected) 

40 output.append( 

41 f"🎯 Project context: {context_items}/{len(context)} indicators detected", 

42 ) 

43 

44 # Add UV setup 

45 output.extend(_setup_uv_dependencies(Path(result["working_directory"]))) 

46 

47 # Add recommendations 

48 recommendations = result["quality_data"].get("recommendations", []) 

49 if recommendations: 

50 output.append("\n💡 Setup recommendations:") 

51 for rec in recommendations[:3]: 

52 output.append(f" • {rec}") 

53 

54 output.append("\n✅ Session initialization completed successfully!") 

55 

56 else: 

57 output.append(f"❌ Session initialization failed: {result['error']}") 

58 

59 except Exception as e: 

60 logger.exception("Session initialization error", error=str(e)) 

61 output.append(f"❌ Unexpected error during initialization: {e}") 

62 

63 return "\n".join(output) 

64 

65 

66async def _checkpoint_impl() -> str: 

67 """Implementation for checkpoint tool.""" 

68 output = [] 

69 output.append( 

70 f"🔍 Claude Session Checkpoint - {session_manager.current_project or 'Current Project'}", 

71 ) 

72 output.append("=" * 50) 

73 

74 try: 

75 result = await session_manager.checkpoint_session() 

76 

77 if result["success"]: 

78 # Add quality assessment output 

79 output.extend(result["quality_output"]) 

80 

81 # Add git checkpoint output 

82 output.extend(result["git_output"]) 

83 

84 output.append(f"\n⏰ Checkpoint completed at: {result['timestamp']}") 

85 output.append( 

86 "\n💡 Use this checkpoint data to track session progress and identify optimization opportunities.", 

87 ) 

88 

89 else: 

90 output.append(f"❌ Checkpoint failed: {result['error']}") 

91 

92 except Exception as e: 

93 logger.exception("Checkpoint error", error=str(e)) 

94 output.append(f"❌ Unexpected checkpoint error: {e}") 

95 

96 return "\n".join(output) 

97 

98 

99async def _end_impl() -> str: 

100 """Implementation for end tool.""" 

101 output = [] 

102 output.append("🏁 Claude Session End - Cleanup and Handoff") 

103 output.append("=" * 50) 

104 

105 try: 

106 result = await session_manager.end_session() 

107 

108 if result["success"]: 

109 summary = result["summary"] 

110 output.append(f"📁 Project: {summary['project']}") 

111 output.append( 

112 f"📊 Final quality score: {summary['final_quality_score']}/100", 

113 ) 

114 output.append(f"⏰ Session ended: {summary['session_end_time']}") 

115 

116 # Add final recommendations 

117 recommendations = summary.get("recommendations", []) 

118 if recommendations: 

119 output.append("\n🎯 Final recommendations for future sessions:") 

120 for rec in recommendations[:5]: 

121 output.append(f" • {rec}") 

122 

123 output.append("\n📝 Session Summary:") 

124 output.append(f" • Working directory: {summary['working_directory']}") 

125 output.append(" • Session data has been logged for future reference") 

126 output.append(" • All temporary resources have been cleaned up") 

127 

128 output.append("\n✅ Session ended successfully!") 

129 output.append( 

130 "💡 Use the session data to improve future development workflows.", 

131 ) 

132 

133 else: 

134 output.append(f"❌ Session end failed: {result['error']}") 

135 

136 except Exception as e: 

137 logger.exception("Session end error", error=str(e)) 

138 output.append(f"❌ Unexpected error during session end: {e}") 

139 

140 return "\n".join(output) 

141 

142 

143async def _status_impl(working_directory: str | None = None) -> str: 

144 """Implementation for status tool.""" 

145 output = [] 

146 output.append("📊 Claude Session Status Report") 

147 output.append("=" * 40) 

148 

149 try: 

150 result = await session_manager.get_session_status(working_directory) 

151 

152 if result["success"]: 

153 output.append(f"📁 Project: {result['project']}") 

154 output.append(f"📂 Working directory: {result['working_directory']}") 

155 output.append(f"📊 Quality score: {result['quality_score']}/100") 

156 

157 # Quality breakdown 

158 breakdown = result["quality_breakdown"] 

159 output.append("\n📈 Quality breakdown:") 

160 output.append(f" • Project health: {breakdown['project_health']:.1f}/40") 

161 output.append(f" • Permissions: {breakdown['permissions']:.1f}/20") 

162 output.append( 

163 f" • Session tools: {breakdown['session_management']:.1f}/20", 

164 ) 

165 output.append(f" • Tool availability: {breakdown['tools']:.1f}/20") 

166 

167 # System health 

168 health = result["system_health"] 

169 output.append("\n🏥 System health:") 

170 output.append( 

171 f" • UV package manager: {'✅' if health['uv_available'] else '❌'}", 

172 ) 

173 output.append( 

174 f" • Git repository: {'✅' if health['git_repository'] else '❌'}", 

175 ) 

176 output.append( 

177 f" • Claude directory: {'✅' if health['claude_directory'] else '❌'}", 

178 ) 

179 

180 # Project context 

181 context = result["project_context"] 

182 context_items = sum(1 for detected in context.values() if detected) 

183 output.append( 

184 f"\n🎯 Project context: {context_items}/{len(context)} indicators", 

185 ) 

186 

187 # Key indicators 

188 key_indicators = [ 

189 ("pyproject.toml", context.get("has_pyproject_toml", False)), 

190 ("Git repository", context.get("has_git_repo", False)), 

191 ("Test suite", context.get("has_tests", False)), 

192 ("Documentation", context.get("has_docs", False)), 

193 ] 

194 

195 for name, detected in key_indicators: 

196 status_icon = "✅" if detected else "❌" 

197 output.append(f" • {name}: {status_icon}") 

198 

199 # Recommendations 

200 recommendations = result["recommendations"] 

201 if recommendations: 

202 output.append("\n💡 Recommendations:") 

203 for rec in recommendations[:3]: 

204 output.append(f" • {rec}") 

205 

206 output.append(f"\n⏰ Status generated: {result['timestamp']}") 

207 

208 else: 

209 output.append(f"❌ Status check failed: {result['error']}") 

210 

211 except Exception as e: 

212 logger.exception("Status check error", error=str(e)) 

213 output.append(f"❌ Unexpected error during status check: {e}") 

214 

215 return "\n".join(output) 

216 

217 

218def _setup_uv_dependencies(current_dir: Path) -> list[str]: 

219 """Set up UV dependencies and requirements.txt generation.""" 

220 output = [] 

221 output.append("\n" + "=" * 50) 

222 output.append("📦 UV Package Management Setup") 

223 output.append("=" * 50) 

224 

225 # Check if uv is available 

226 uv_available = shutil.which("uv") is not None 

227 if not uv_available: 

228 output.append("⚠️ UV not found in PATH") 

229 output.append("💡 Install UV: curl -LsSf https://astral.sh/uv/install.sh | sh") 

230 return output 

231 

232 # Check for pyproject.toml 

233 pyproject_path = current_dir / "pyproject.toml" 

234 if pyproject_path.exists(): 

235 output.append("✅ Found pyproject.toml - UV project detected") 

236 

237 # Run uv sync if dependencies need updating 

238 try: 

239 sync_result = subprocess.run( 

240 ["uv", "sync"], 

241 check=False, 

242 cwd=current_dir, 

243 capture_output=True, 

244 text=True, 

245 timeout=60, 

246 ) 

247 

248 if sync_result.returncode == 0: 

249 output.append("✅ UV dependencies synchronized") 

250 else: 

251 output.append(f"⚠️ UV sync had issues: {sync_result.stderr}") 

252 except subprocess.TimeoutExpired: 

253 output.append( 

254 "⚠️ UV sync timed out - dependencies may need manual attention", 

255 ) 

256 except Exception as e: 

257 output.append(f"⚠️ UV sync error: {e}") 

258 else: 

259 output.append("ℹ️ No pyproject.toml found") 

260 output.append("💡 Consider running 'uv init' to create a new UV project") 

261 

262 return output 

263 

264 

265def register_session_tools(mcp_server) -> None: 

266 """Register all session management tools with the MCP server.""" 

267 

268 @mcp_server.tool() 

269 async def init(working_directory: str | None = None) -> str: 

270 """Initialize Claude session with comprehensive setup including UV dependencies and automation tools. 

271 

272 Args: 

273 working_directory: Optional working directory override (defaults to PWD environment variable or current directory) 

274 

275 """ 

276 return await _init_impl(working_directory) 

277 

278 @mcp_server.tool() 

279 async def checkpoint() -> str: 

280 """Perform mid-session quality checkpoint with workflow analysis and optimization recommendations.""" 

281 return await _checkpoint_impl() 

282 

283 @mcp_server.tool() 

284 async def end() -> str: 

285 """End Claude session with cleanup, learning capture, and handoff file creation.""" 

286 return await _end_impl() 

287 

288 @mcp_server.tool() 

289 async def status(working_directory: str | None = None) -> str: 

290 """Get current session status and project context information with health checks. 

291 

292 Args: 

293 working_directory: Optional working directory override (defaults to PWD environment variable or current directory) 

294 

295 """ 

296 return await _status_impl(working_directory)