Coverage for session_buddy / tools / session_tools.py: 81.08%
354 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-04 00:43 -0800
1#!/usr/bin/env python3
2"""Session management MCP tools.
4This module provides tools for managing Claude session lifecycle including
5initialization, checkpoints, and cleanup.
7Refactored to use utility modules for reduced code duplication.
8"""
10from __future__ import annotations
12import shutil
13import subprocess # nosec B404
14from contextlib import suppress
15from dataclasses import dataclass, field
16from pathlib import Path
17from typing import TYPE_CHECKING, Any
19from session_buddy.di import get_sync_typed
20from session_buddy.di.container import depends
22if TYPE_CHECKING:
23 import psutil
24 from fastmcp import FastMCP
26from session_buddy.core import SessionLifecycleManager
27from session_buddy.utils.error_handlers import _get_logger
30@dataclass
31class SessionOutputBuilder:
32 """Centralized output formatting with consistent styling."""
34 sections: list[str] = field(default_factory=list)
36 def add_header(self, title: str, separator_char: str = "=") -> None:
37 """Add formatted header."""
38 separator = separator_char * len(title)
39 self.sections.extend([title, separator])
41 def add_section(self, title: str, items: list[str]) -> None:
42 """Add formatted section with items."""
43 if title: 43 ↛ 45line 43 didn't jump to line 45 because the condition on line 43 was always true
44 self.sections.append(f"\n{title}:")
45 self.sections.extend(items)
47 def add_status_item(self, name: str, status: bool, value: str = "") -> None:
48 """Add status indicator item."""
49 icon = "✅" if status else "❌"
50 display = f" • {name}: {icon}"
51 if value:
52 display += f" {value}"
53 self.sections.append(display)
55 def add_simple_item(self, item: str) -> None:
56 """Add simple item."""
57 self.sections.append(item)
59 def build(self) -> str:
60 """Build final output string."""
61 return "\n".join(self.sections)
64@dataclass
65class SessionSetupResults:
66 """Results from session setup operations."""
68 uv_setup: list[str] = field(default_factory=list)
69 shortcuts_result: dict[str, Any] = field(default_factory=dict)
70 recommendations: list[str] = field(default_factory=list)
73# ============================================================================
74# Service Resolution
75# ============================================================================
78def _get_session_manager() -> SessionLifecycleManager:
79 """Get or create SessionLifecycleManager instance.
81 Note:
82 Uses the Oneiric-backed service container for singleton resolution.
84 """
85 with suppress(Exception):
86 manager = get_sync_typed(SessionLifecycleManager) # type: ignore[no-any-return]
87 if isinstance(manager, SessionLifecycleManager): 87 ↛ 90line 87 didn't jump to line 90
88 return manager
90 manager = SessionLifecycleManager()
91 depends.set(SessionLifecycleManager, manager)
92 return manager
95# ============================================================================
96# Session Shortcuts Management
97# ============================================================================
100def _create_session_shortcuts() -> dict[str, Any]:
101 """Create Claude Code slash command shortcuts for session management.
103 Creates /start, /checkpoint, and /end shortcuts in ~/.claude/commands/
104 that map to session-mgmt MCP tools.
106 Returns:
107 Dict with 'created' bool, 'existed' bool, and 'shortcuts' list
109 """
110 claude_home = Path.home() / ".claude"
111 commands_dir = claude_home / "commands"
113 # Create commands directory if it doesn't exist
114 commands_dir.mkdir(parents=True, exist_ok=True)
116 shortcuts = {
117 "start": {
118 "file": "start.md",
119 "content": """---
120description: Start session management for current project
121---
123Please use the `mcp__session-buddy__start` tool to initialize session management for the current project.
125This will:
1261. Set up session tracking for the git repository
1272. Initialize conversation memory and context
1283. Prepare the project for enhanced Claude Code workflows
1294. Install UV dependencies and automation tools
1305. Create session management slash command shortcuts
131""",
132 },
133 "checkpoint": {
134 "file": "checkpoint.md",
135 "content": """---
136argument-hint: [checkpoint-name]
137description: Create a session checkpoint with progress summary
138---
140Please use the `mcp__session-buddy__checkpoint` tool to create a session checkpoint.
142This command will:
1431. Create a checkpoint of the current development session
1442. Analyze code quality and calculate quality scores
1453. Summarize progress made so far
1464. Document any pending tasks or context
1475. Prepare for seamless session resumption
149The tool will analyze the working directory and provide comprehensive quality metrics.
150""",
151 },
152 "end": {
153 "file": "end.md",
154 "content": """---
155description: End current session with cleanup and summary
156---
158Please use the `mcp__session-buddy__end` tool to gracefully end the current session.
160This will:
1611. Create a final checkpoint of all work completed
1622. Generate session summary and insights
1633. Clean up temporary resources
1644. Prepare handoff documentation for next session
1655. Store final quality metrics and learning data
166""",
167 },
168 }
170 created_shortcuts = []
171 existing_shortcuts = []
173 for shortcut_name, shortcut_data in shortcuts.items():
174 shortcut_path = commands_dir / shortcut_data["file"]
176 if shortcut_path.exists():
177 existing_shortcuts.append(shortcut_name)
178 else:
179 try:
180 shortcut_path.write_text(shortcut_data["content"])
181 created_shortcuts.append(shortcut_name)
182 _get_logger().info(f"Created slash command shortcut: /{shortcut_name}")
183 except Exception as e:
184 _get_logger().exception(
185 f"Failed to create shortcut /{shortcut_name}: {e}",
186 )
188 return {
189 "created": bool(created_shortcuts),
190 "existed": bool(existing_shortcuts) and not created_shortcuts,
191 "shortcuts": created_shortcuts or existing_shortcuts,
192 }
195# ============================================================================
196# Working Directory Detection
197# ============================================================================
200def _check_environment_variables() -> str | None:
201 """Check for Claude Code environment variables."""
202 import os
204 for env_var in ("CLAUDE_WORKING_DIR", "CLIENT_PWD", "CLAUDE_PROJECT_DIR"):
205 if env_var in os.environ:
206 client_dir = os.environ[env_var]
207 if client_dir and Path(client_dir).exists(): 207 ↛ 204line 207 didn't jump to line 204 because the condition on line 207 was always true
208 return client_dir
209 return None
212def _check_working_dir_file() -> str | None:
213 """Check for the temporary file used by Claude's auto-start scripts."""
214 import tempfile
216 working_dir_file = Path(tempfile.gettempdir()) / "claude-git-working-dir"
217 if working_dir_file.exists():
218 with suppress(OSError, PermissionError, ValueError, UnicodeDecodeError):
219 stored_dir = working_dir_file.read_text().strip()
220 # Only use if it's NOT the session-mgmt-mcp server directory
221 if ( 221 ↛ 226line 221 didn't jump to line 226 because the condition on line 221 was never true
222 stored_dir
223 and Path(stored_dir).exists()
224 and not stored_dir.endswith("session-mgmt-mcp")
225 ):
226 return stored_dir
227 return None
230def _check_parent_process_cwd() -> str | None:
231 """Check parent process working directory (advanced)."""
232 with suppress(ImportError, Exception):
233 import psutil
235 parent_process = psutil.Process().parent()
236 if parent_process: 236 ↛ 246line 236 didn't jump to line 246
237 parent_cwd = parent_process.cwd()
238 # Only use if it's a different directory and exists
239 if ( 239 ↛ 245line 239 didn't jump to line 245 because the condition on line 239 was never true
240 parent_cwd
241 and Path(parent_cwd).exists()
242 and parent_cwd != str(Path.cwd())
243 and not parent_cwd.endswith("session-mgmt-mcp")
244 ):
245 return parent_cwd
246 return None
249def _is_git_repository(repo_path: Path) -> bool:
250 """Check if a path is a git repository."""
251 return repo_path.is_dir() and (repo_path / ".git").exists()
254def _safe_get_mtime(repo_path: Path) -> float | None:
255 """Safely get modification time of a repository."""
256 try:
257 return repo_path.stat().st_mtime
258 except Exception:
259 return None
262def _collect_git_repos(projects_path: Path) -> list[tuple[float, str]]:
263 """Collect git repositories with modification times from a directory."""
264 git_repos: list[tuple[float, str]] = []
265 for repo_path in projects_path.iterdir():
266 if _is_git_repository(repo_path):
267 mtime = _safe_get_mtime(repo_path)
268 if mtime is not None: 268 ↛ 265line 268 didn't jump to line 265 because the condition on line 268 was always true
269 git_repos.append((mtime, str(repo_path)))
270 return git_repos
273def _get_most_recent_client_repo(git_repos: list[tuple[float, str]]) -> str | None:
274 """Get most recent repository excluding the server itself."""
275 git_repos.sort(reverse=True)
276 for _mtime, repo_path_str in git_repos: 276 ↛ 279line 276 didn't jump to line 279 because the loop on line 276 didn't complete
277 if not repo_path_str.endswith("session-mgmt-mcp"): 277 ↛ 276line 277 didn't jump to line 276 because the condition on line 277 was always true
278 return repo_path_str
279 return None
282def _find_recent_git_repository() -> str | None:
283 """Look for recent git repositories in common project directories."""
284 for projects_dir in ("/Users/les/Projects", str(Path.home() / "Projects")): 284 ↛ 294line 284 didn't jump to line 294 because the loop on line 284 didn't complete
285 projects_path = Path(projects_dir)
286 if not projects_path.exists(): 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true
287 continue
289 git_repos = _collect_git_repos(projects_path)
290 if git_repos: 290 ↛ 284line 290 didn't jump to line 284 because the condition on line 290 was always true
291 if repo := _get_most_recent_client_repo(git_repos): 291 ↛ 284line 291 didn't jump to line 284 because the condition on line 291 was always true
292 return repo
294 return None
297def _get_client_working_directory() -> str | None:
298 """Auto-detect the client's working directory using multiple detection methods."""
299 # Method 1: Check for Claude Code environment variables
300 if client_dir := _check_environment_variables(): 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 return client_dir
303 # Method 2: Check for the temporary file used by Claude's auto-start scripts
304 if client_dir := _check_working_dir_file(): 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true
305 return client_dir
307 # Method 3: Check parent process working directory (advanced)
308 if client_dir := _check_parent_process_cwd(): 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 return client_dir
311 # Method 4: Look for recent git repositories in common project directories
312 if client_dir := _find_recent_git_repository(): 312 ↛ 315line 312 didn't jump to line 315 because the condition on line 312 was always true
313 return client_dir
315 return None
318# ============================================================================
319# Environment Setup Operations
320# ============================================================================
323async def _perform_environment_setup(result: dict[str, Any]) -> SessionSetupResults:
324 """Perform all environment setup tasks. Target complexity: ≤5."""
325 working_dir = Path(result["working_directory"])
327 uv_setup = _setup_uv_dependencies(working_dir)
328 shortcuts_result = _create_session_shortcuts()
329 recommendations = result["quality_data"].get("recommendations", [])
331 return SessionSetupResults(
332 uv_setup=uv_setup,
333 shortcuts_result=shortcuts_result,
334 recommendations=recommendations,
335 )
338def _setup_uv_dependencies(current_dir: Path) -> list[str]:
339 """Set up UV dependencies and requirements.txt generation."""
340 output = []
341 output.extend(("\n" + "=" * 50, "📦 UV Package Management Setup", "=" * 50))
343 # Check if uv is available
344 uv_available = shutil.which("uv") is not None
345 if not uv_available:
346 output.extend(
347 (
348 "⚠️ UV not found in PATH",
349 "💡 Install UV: curl -LsSf https://astral.sh/uv/install.sh | sh",
350 )
351 )
352 return output
354 # Check for pyproject.toml
355 pyproject_path = current_dir / "pyproject.toml"
356 if pyproject_path.exists():
357 output.append("✅ Found pyproject.toml - UV project detected")
359 # Run uv sync if dependencies need updating
360 try:
361 sync_result = subprocess.run(
362 ["uv", "sync"],
363 check=False,
364 cwd=current_dir,
365 capture_output=True,
366 text=True,
367 timeout=60,
368 )
370 if sync_result.returncode == 0: 370 ↛ 373line 370 didn't jump to line 373 because the condition on line 370 was always true
371 output.append("✅ UV dependencies synchronized")
372 else:
373 output.append(f"⚠️ UV sync had issues: {sync_result.stderr}")
374 except subprocess.TimeoutExpired:
375 output.append(
376 "⚠️ UV sync timed out - dependencies may need manual attention",
377 )
378 except Exception as e:
379 output.append(f"⚠️ UV sync error: {e}")
380 else:
381 output.extend(
382 (
383 "ℹ️ No pyproject.toml found",
384 "💡 Consider running 'uv init' to create a new UV project",
385 )
386 )
388 return output
391# ============================================================================
392# Output Formatting Helpers
393# ============================================================================
396def _add_session_info_to_output(
397 output_builder: SessionOutputBuilder,
398 result: dict[str, Any],
399) -> None:
400 """Add session information to output. Target complexity: ≤5."""
401 output_builder.add_simple_item(f"📁 Current project: {result['project']}")
402 output_builder.add_simple_item(
403 f"📂 Working directory: {result['working_directory']}",
404 )
405 output_builder.add_simple_item(f"🏠 Claude directory: {result['claude_directory']}")
406 output_builder.add_simple_item(
407 f"📊 Initial quality score: {result['quality_score']}/100",
408 )
410 # Add project context info
411 context = result["project_context"]
412 context_items = sum(1 for detected in context.values() if detected)
413 output_builder.add_simple_item(
414 f"🎯 Project context: {context_items}/{len(context)} indicators detected",
415 )
418def _add_environment_info_to_output(
419 output_builder: SessionOutputBuilder,
420 setup_results: SessionSetupResults,
421) -> None:
422 """Add environment setup info to output. Target complexity: ≤5."""
423 # Add UV setup
424 output_builder.sections.extend(setup_results.uv_setup)
426 # Add recommendations
427 if setup_results.recommendations:
428 output_builder.add_section(
429 "💡 Setup recommendations",
430 [f" • {rec}" for rec in setup_results.recommendations[:3]],
431 )
433 # Add shortcuts
434 shortcuts = setup_results.shortcuts_result
435 if shortcuts.get("created"):
436 output_builder.add_section(
437 "🔧 Created session management shortcuts",
438 [f" • /{shortcut}" for shortcut in shortcuts["shortcuts"]],
439 )
440 elif shortcuts.get("existed"): 440 ↛ exitline 440 didn't return from function '_add_environment_info_to_output' because the condition on line 440 was always true
441 output_builder.add_simple_item("\n✅ Session shortcuts already exist")
444def _add_project_section_to_output(
445 output_builder: SessionOutputBuilder,
446 result: dict[str, Any],
447) -> None:
448 """Add project information to output. Target complexity: ≤3."""
449 output_builder.add_simple_item(f"📁 Project: {result['project']}")
450 output_builder.add_simple_item(
451 f"📂 Working directory: {result['working_directory']}",
452 )
453 output_builder.add_simple_item(f"📊 Quality score: {result['quality_score']}/100")
456def _add_quality_section_to_output(
457 output_builder: SessionOutputBuilder,
458 breakdown: dict[str, Any],
459) -> None:
460 """Add quality breakdown to output. Target complexity: ≤5."""
461 quality_items = [
462 f" • Code quality: {breakdown['code_quality']:.1f}/40",
463 f" • Project health: {breakdown['project_health']:.1f}/30",
464 f" • Dev velocity: {breakdown['dev_velocity']:.1f}/20",
465 f" • Security: {breakdown['security']:.1f}/10",
466 ]
467 output_builder.add_section("📈 Quality breakdown", quality_items)
470def _add_health_section_to_output(
471 output_builder: SessionOutputBuilder,
472 health: dict[str, Any],
473) -> None:
474 """Add system health to output. Target complexity: ≤5."""
475 output_builder.add_section("🏥 System health", [])
476 output_builder.add_status_item("UV package manager", health["uv_available"])
477 output_builder.add_status_item("Git repository", health["git_repository"])
478 output_builder.add_status_item("Claude directory", health["claude_directory"])
481def _add_project_context_to_output(
482 output_builder: SessionOutputBuilder,
483 context: dict[str, Any],
484) -> None:
485 """Add project context to output. Target complexity: ≤5."""
486 context_items = sum(1 for detected in context.values() if detected)
487 output_builder.add_simple_item(
488 f"\n🎯 Project context: {context_items}/{len(context)} indicators",
489 )
491 key_indicators = [
492 ("pyproject.toml", context.get("has_pyproject_toml", False)),
493 ("Git repository", context.get("has_git_repo", False)),
494 ("Test suite", context.get("has_tests", False)),
495 ("Documentation", context.get("has_docs", False)),
496 ]
498 for name, detected in key_indicators:
499 output_builder.add_status_item(name, detected)
502# ============================================================================
503# Checkpoint-Specific Helpers
504# ============================================================================
507async def _handle_auto_store_reflection(
508 result: dict[str, Any],
509 output: list[str],
510) -> None:
511 """Handle selective auto-store reflection logic."""
512 auto_store_decision = result.get("auto_store_decision")
513 if not auto_store_decision: 513 ↛ 516line 513 didn't jump to line 516 because the condition on line 513 was always true
514 return
516 if auto_store_decision.should_store:
517 from session_buddy.reflection_tools import get_reflection_database
518 from session_buddy.utils.reflection_utils import generate_auto_store_tags
520 try:
521 db = await get_reflection_database()
523 # Create meaningful checkpoint summary
524 checkpoint_content = f"Quality score: {result['quality_score']}/100. "
525 if auto_store_decision.metadata.get("delta"):
526 delta = auto_store_decision.metadata["delta"]
527 direction = (
528 "improved"
529 if auto_store_decision.reason.value == "quality_improvement"
530 else "changed"
531 )
532 checkpoint_content += f"Quality {direction} by {delta} points. "
534 checkpoint_content += (
535 f"Project: {_get_session_manager().current_project or 'unknown'}. "
536 )
537 checkpoint_content += f"Timestamp: {result['timestamp']}"
539 # Generate semantic tags
540 tags = generate_auto_store_tags(
541 reason=auto_store_decision.reason,
542 project=_get_session_manager().current_project,
543 quality_score=result["quality_score"],
544 )
546 # Store the reflection
547 await db.store_reflection(checkpoint_content, tags)
548 output.append(f"\n{result['auto_store_summary']}")
549 except Exception as e:
550 _get_logger().exception(f"Failed to store checkpoint reflection: {e}")
551 output.append(f"⚠️ Reflection storage failed: {e}")
552 else:
553 # Show why we skipped auto-store
554 output.append(f"\n{result.get('auto_store_summary', '')}")
557async def _handle_auto_compaction(output: list[str]) -> None:
558 """Handle automatic compaction analysis and execution."""
559 from session_buddy.server_optimized import (
560 _execute_auto_compact,
561 should_suggest_compact,
562 )
564 should_compact, reason = should_suggest_compact()
565 output.extend(("\n🔄 Automatic Compaction Analysis", f"📊 {reason}"))
567 if should_compact: 567 ↛ 580line 567 didn't jump to line 580 because the condition on line 567 was always true
568 output.append("\n🔄 Executing automatic compaction...")
569 try:
570 await _execute_auto_compact()
571 output.append("✅ Context automatically optimized")
572 except Exception as e:
573 output.extend(
574 (
575 f"⚠️ Auto-compact skipped: {e!s}",
576 "💡 Consider running /compact manually",
577 )
578 )
579 else:
580 output.append("✅ Context appears well-optimized for current session")
583# ============================================================================
584# End Session Formatting Helpers
585# ============================================================================
588def _format_successful_end(summary: dict[str, Any]) -> list[str]:
589 """Format successful session end output."""
590 output = [
591 f"📁 Project: {summary['project']}",
592 f"📊 Final quality score: {summary['final_quality_score']}/100",
593 f"⏰ Session ended: {summary['session_end_time']}",
594 ]
596 output.extend(_format_recommendations(summary.get("recommendations", [])))
597 output.extend(_format_session_summary(summary))
599 output.extend(
600 [
601 "\n✅ Session ended successfully!",
602 "💡 Use the session data to improve future development workflows.",
603 ],
604 )
606 return output
609def _format_recommendations(recommendations: list[str]) -> list[str]:
610 """Format recommendations section."""
611 if not recommendations: 611 ↛ 612line 611 didn't jump to line 612 because the condition on line 611 was never true
612 return []
614 output = ["\n🎯 Final recommendations for future sessions:"]
615 output.extend(f" • {rec}" for rec in recommendations[:5])
616 return output
619def _format_session_summary(summary: dict[str, Any]) -> list[str]:
620 """Format session summary section."""
621 output = [
622 "\n📝 Session Summary:",
623 f" • Working directory: {summary['working_directory']}",
624 " • Session data has been logged for future reference",
625 " • All temporary resources have been cleaned up",
626 ]
628 # Add handoff documentation info
629 handoff_doc = summary.get("handoff_documentation")
630 if handoff_doc:
631 output.append(f" • Handoff documentation: {handoff_doc}")
633 return output
636# ============================================================================
637# Tool Implementations
638# ============================================================================
641async def _start_impl(working_directory: str | None = None) -> str:
642 """Initialize session with comprehensive setup. Target complexity: ≤8."""
643 output_builder = SessionOutputBuilder()
644 output_builder.add_header("🚀 Claude Session Initialization via MCP Server")
646 try:
647 result = await _get_session_manager().initialize_session(working_directory)
649 if result["success"]:
650 _add_session_info_to_output(output_builder, result)
651 setup_results = await _perform_environment_setup(result)
652 _add_environment_info_to_output(output_builder, setup_results)
653 output_builder.add_simple_item(
654 "\n✅ Session initialization completed successfully!",
655 )
656 else:
657 output_builder.add_simple_item(
658 f"❌ Session initialization failed: {result['error']}",
659 )
661 except Exception as e:
662 _get_logger().exception("Session initialization error: %s", str(e))
663 output_builder.add_simple_item(
664 f"❌ Unexpected error during initialization: {e}",
665 )
667 return output_builder.build()
670async def _checkpoint_impl(working_directory: str | None = None) -> str:
671 """Implementation for checkpoint tool."""
672 # Auto-detect client working directory if not provided
673 if not working_directory:
674 working_directory = _get_client_working_directory()
676 output = []
677 output.extend(
678 (
679 f"🔍 Claude Session Checkpoint - {_get_session_manager().current_project or 'Current Project'}",
680 "=" * 50,
681 )
682 )
684 try:
685 # Determine if this is a manual checkpoint (always true for explicit tool calls)
686 result = await _get_session_manager().checkpoint_session(
687 working_directory,
688 is_manual=True,
689 )
691 if result["success"]:
692 # Add quality assessment output
693 output.extend(result["quality_output"])
695 # Add git checkpoint output
696 output.extend(result["git_output"])
698 # Handle selective auto-store reflection
699 try:
700 await _handle_auto_store_reflection(result, output)
701 except Exception as e:
702 _get_logger().warning(
703 f"Auto-store reflection error (non-critical): {e}"
704 )
705 # Continue - this is not critical for checkpoint success
707 # Auto-compact when needed
708 try:
709 await _handle_auto_compaction(output)
710 except Exception as e:
711 _get_logger().warning(f"Auto-compaction error (non-critical): {e}")
712 output.append(f"\n⚠️ Auto-compaction skipped: {e!s}")
714 output.extend(
715 (
716 f"\n⏰ Checkpoint completed at: {result['timestamp']}",
717 "\n💡 This checkpoint includes intelligent conversation management and optimization.",
718 )
719 )
720 else:
721 output.append(f"❌ Checkpoint failed: {result['error']}")
723 except Exception as e:
724 _get_logger().exception("Checkpoint error: %s", str(e))
725 output.append(f"❌ Unexpected checkpoint error: {e}")
727 return "\n".join(output)
730async def _end_impl(working_directory: str | None = None) -> str:
731 """Implementation for end tool."""
732 # Auto-detect client working directory if not provided
733 if not working_directory:
734 working_directory = _get_client_working_directory()
736 output = [
737 "🏁 Claude Session End - Cleanup and Handoff",
738 "=" * 50,
739 ]
741 try:
742 result = await _get_session_manager().end_session(working_directory)
744 if result["success"]:
745 output.extend(_format_successful_end(result["summary"]))
746 else:
747 output.append(f"❌ Session end failed: {result['error']}")
749 except Exception as e:
750 _get_logger().exception("Session end error: %s", str(e))
751 output.append(f"❌ Unexpected error during session end: {e}")
753 return "\n".join(output)
756async def _status_impl(working_directory: str | None = None) -> str:
757 """Get comprehensive session status. Target complexity: ≤8."""
758 output_builder = SessionOutputBuilder()
759 output_builder.add_header("📊 Claude Session Status Report")
761 try:
762 result = await _get_session_manager().get_session_status(working_directory)
764 if result["success"]:
765 _add_project_section_to_output(output_builder, result)
766 _add_quality_section_to_output(output_builder, result["quality_breakdown"])
767 _add_health_section_to_output(output_builder, result["system_health"])
768 _add_project_context_to_output(output_builder, result["project_context"])
770 # Recommendations
771 recommendations = result["recommendations"]
772 if recommendations: 772 ↛ 778line 772 didn't jump to line 778 because the condition on line 772 was always true
773 output_builder.add_section(
774 "💡 Recommendations",
775 [f" • {rec}" for rec in recommendations[:3]],
776 )
778 output_builder.add_simple_item(
779 f"\n⏰ Status generated: {result['timestamp']}",
780 )
782 else:
783 output_builder.add_simple_item(f"❌ Status check failed: {result['error']}")
785 except Exception as e:
786 _get_logger().exception("Status check error: %s", str(e))
787 output_builder.add_simple_item(f"❌ Unexpected error during status check: {e}")
789 return output_builder.build()
792# ============================================================================
793# MCP Tool Registration
794# ============================================================================
797def register_session_tools(mcp_server: FastMCP) -> None:
798 """Register all session management tools with the MCP server."""
800 @mcp_server.tool()
801 async def start(working_directory: str | None = None) -> str:
802 """Initialize Claude session with comprehensive setup including UV dependencies and automation tools.
804 Args:
805 working_directory: Optional working directory override (defaults to PWD environment variable or current directory)
807 """
808 return await _start_impl(working_directory)
810 @mcp_server.tool()
811 async def checkpoint(working_directory: str | None = None) -> str:
812 """Perform mid-session quality checkpoint with workflow analysis and optimization recommendations.
814 Args:
815 working_directory: Optional working directory override (defaults to PWD environment variable or current directory)
817 """
818 return await _checkpoint_impl(working_directory)
820 @mcp_server.tool()
821 async def end(working_directory: str | None = None) -> str:
822 """End Claude session with cleanup, learning capture, and handoff file creation.
824 Args:
825 working_directory: Optional working directory override (defaults to PWD environment variable or current directory)
827 """
828 return await _end_impl(working_directory)
830 @mcp_server.tool()
831 async def status(working_directory: str | None = None) -> str:
832 """Get current session status and project context information with health checks.
834 Args:
835 working_directory: Optional working directory override (defaults to PWD environment variable or current directory)
837 """
838 return await _status_impl(working_directory)
840 @mcp_server.tool()
841 async def health_check() -> str:
842 """Simple health check that doesn't require database or session context."""
843 import os
844 import platform
845 import time
847 try:
848 working_directory = str(Path.cwd())
849 except FileNotFoundError:
850 # Handle case where current working directory doesn't exist
851 working_directory = "[Current directory unavailable]"
853 health_info = {
854 "server_status": "✅ Active",
855 "timestamp": time.time(),
856 "platform": platform.system(),
857 "python_version": platform.python_version(),
858 "process_id": os.getpid(),
859 "working_directory": working_directory,
860 }
862 return f"""🏥 MCP Server Health Check
863================================
864Server Status: {health_info["server_status"]}
865Platform: {health_info["platform"]}
866Python: {health_info["python_version"]}
867Process ID: {health_info["process_id"]}
868Working Directory: {health_info["working_directory"]}
869Timestamp: {health_info["timestamp"]}
871✅ MCP server is operational and responding to requests."""
873 @mcp_server.tool()
874 async def server_info() -> str:
875 """Get basic server information without requiring session context."""
876 import time
878 try:
879 # Check if we can access basic file system info
880 home_dir = Path.home()
882 try:
883 current_dir = Path.cwd()
884 except FileNotFoundError:
885 # Handle case where current working directory doesn't exist
886 current_dir = Path("[Current directory unavailable]")
888 return f"""📊 Session-mgmt MCP Server Information
889===========================================
890🏠 Home Directory: {home_dir}
891📁 Current Directory: {current_dir}
892🕐 Server Time: {time.strftime("%Y-%m-%d %H:%M:%S")}
893🔧 FastMCP Framework: Active
894🌐 Transport: streamable-http
895📡 Endpoint: /mcp
897✅ Server is running and accessible."""
899 except Exception as e:
900 return f"⚠️ Server info error: {e!s}"
902 @mcp_server.tool()
903 async def ping() -> str:
904 """Simple ping endpoint to test MCP connectivity."""
905 return "🏓 Pong! MCP server is responding"