Pathfinder MCP Server

"""Research phase handler.""" import uuid from typing import Any from pathfinder_mcp.handlers.base import BaseHandler from pathfinder_mcp.session import SessionSnapshot from pathfinder_mcp.state import Phase, PhaseState class ResearchHandler(BaseHandler): """Handler for research phase operations.""" phase = Phase.RESEARCH async def execute( self, session_id: str | None = None, task_description: str = "", findings: str | None = None, **kwargs: Any, ) -> dict[str, Any]: """Execute research operation. If findings is None, starts new research session. If findings is provided, saves to existing session. Args: session_id: Session ID (generated if None and starting) task_description: Task description (for new sessions) findings: Research findings to save (for existing sessions) Returns: Operation result """ if findings is not None: return self._save_findings(session_id or "", findings) return self._start_research(task_description, session_id) def _start_research( self, task_description: str, session_id: str | None ) -> dict[str, Any]: """Start a new research session.""" sid = session_id or f"session_{uuid.uuid4().hex[:8]}" # Create session session_path = self.session_manager.create_session(sid) # Initialize state state = PhaseState(session_id=sid, current_phase=Phase.RESEARCH) self.set_session(sid, state) # Create artifact artifact_path = self.artifact_writer.write_research( sid, content="", task=task_description ) # Track context self.context_monitor.add_message(task_description) # Save snapshot snapshot = SessionSnapshot( session_id=sid, phase=Phase.RESEARCH, task_description=task_description, context_tokens=self.context_monitor.current_tokens, ) self.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_findings(self, session_id: str, findings: str) -> dict[str, Any]: """Save research findings.""" # Validate error = self.validate_phase(session_id, [Phase.RESEARCH]) if error: return error # Save findings artifact_path = self.artifact_writer.write_research(session_id, findings) # Track context self.context_monitor.add_message(findings) # Update snapshot snapshot = self.session_manager.load_snapshot(session_id) if snapshot: snapshot.research_summary += f"\n{findings[:200]}..." snapshot.context_tokens = self.context_monitor.current_tokens self.session_manager.save_snapshot(session_id, snapshot) response: dict[str, Any] = { "session_id": session_id, "artifact_path": str(artifact_path), "context": self.get_context_status(), } if self.context_monitor.should_warn(): response["warning"] = "Context >70%. Consider compact_context." elif self.context_monitor.should_compact(): response["suggestion"] = "Context >60%. Compaction recommended." return response