Pathfinder MCP Server

"""Research phase tools.""" import uuid from typing import Any from pathfinder_mcp.artifacts import ArtifactWriter from pathfinder_mcp.context import ContextMonitor from pathfinder_mcp.session import SessionManager, SessionSnapshot from pathfinder_mcp.state import Phase, PhaseState def start_research( task_description: str, session_id: str | None = None, *, session_manager: SessionManager, artifact_writer: ArtifactWriter, context_monitor: ContextMonitor, sessions: dict[str, PhaseState], ) -> dict[str, Any]: """Initialize a new research session. Args: task_description: Description of the task to research session_id: Optional session ID (auto-generated if not provided) session_manager: Session manager instance artifact_writer: Artifact writer instance context_monitor: Context monitor instance sessions: Active sessions dict Returns: Session info with session_id and artifact path """ # Generate session ID if not provided sid = session_id or f"session_{uuid.uuid4().hex[:8]}" # Create session directory session_path = session_manager.create_session(sid) # Initialize phase state state = PhaseState(session_id=sid, current_phase=Phase.RESEARCH) sessions[sid] = state # Create initial research artifact artifact_path = artifact_writer.write_research( sid, content="", task=task_description ) # Track tokens context_monitor.add_message(task_description) # Save initial snapshot snapshot = SessionSnapshot( session_id=sid, phase=Phase.RESEARCH, task_description=task_description, context_tokens=context_monitor.current_tokens, ) session_manager.save_snapshot(sid, snapshot) return { "session_id": sid, "phase": "research", "session_path": str(session_path), "artifact_path": str(artifact_path), "message": "Research session started. Document findings with save_research.", } def save_research( session_id: str, findings: str, *, session_manager: SessionManager, artifact_writer: ArtifactWriter, context_monitor: ContextMonitor, sessions: dict[str, PhaseState], ) -> dict[str, Any]: """Save research findings to the session. Args: session_id: Session ID findings: Research findings to save session_manager: Session manager instance artifact_writer: Artifact writer instance context_monitor: Context monitor instance sessions: Active sessions dict Returns: Status with context utilization info """ # Validate session exists and is in research phase state = sessions.get(session_id) if not state: return {"error": "Session not found", "code": "SESSION_NOT_FOUND"} if not state.is_research: return { "error": f"Cannot save research in {state.current_phase.value} phase", "code": "INVALID_PHASE", } # Append findings to research artifact artifact_path = artifact_writer.write_research(session_id, findings) # Track tokens context_monitor.add_message(findings) # Update snapshot snapshot = session_manager.load_snapshot(session_id) if snapshot: snapshot.research_summary += f"\n{findings[:200]}..." snapshot.context_tokens = context_monitor.current_tokens session_manager.save_snapshot(session_id, snapshot) # Build response response: dict[str, Any] = { "session_id": session_id, "artifact_path": str(artifact_path), "context": context_monitor.get_status(), } # Add compaction warning if needed if context_monitor.should_warn(): response["warning"] = ( "Context utilization >70%. Consider using compact_context." ) elif context_monitor.should_compact(): response["suggestion"] = ( "Context utilization >60%. Compaction recommended soon." ) return response