Pathfinder MCP Server

"""Pathfinder MCP Server - Main FastMCP server instance.""" from fastmcp import FastMCP from fastmcp.prompts import Message from fastmcp.server.context import Context from pathfinder_mcp.artifacts import ( ArtifactWriter, PLAN_TEMPLATE, PROGRESS_TEMPLATE, RESEARCH_TEMPLATE, ) from pathfinder_mcp.config import ServerConfig from pathfinder_mcp.context import ContextMonitor from pathfinder_mcp.handlers import ImplementHandler, PlanHandler, ResearchHandler from pathfinder_mcp.session import SessionManager from pathfinder_mcp.state import PhaseState from pathfinder_mcp.tools import compact, implement, plan, research # Load configuration config = ServerConfig.from_env() # Initialize FastMCP server mcp = FastMCP("PathfinderMCP") # Global instances session_manager = SessionManager() artifact_writer = ArtifactWriter(session_manager) context_monitor = ContextMonitor(max_tokens=config.max_context_tokens) # Track active sessions _active_sessions: dict[str, PhaseState] = {} # Initialize handlers (for future subagent delegation) _research_handler = ResearchHandler( session_manager, artifact_writer, context_monitor, _active_sessions ) _plan_handler = PlanHandler( session_manager, artifact_writer, context_monitor, _active_sessions ) _implement_handler = ImplementHandler( session_manager, artifact_writer, context_monitor, _active_sessions ) # ============================================================================= # Resources (Templates) # ============================================================================= @mcp.resource("pathfinder://templates/research") def research_template() -> str: """Research phase markdown template.""" return RESEARCH_TEMPLATE.format(task="[Task Description]") @mcp.resource("pathfinder://templates/plan") def plan_template() -> str: """Plan phase markdown template.""" return PLAN_TEMPLATE.format(name="[Plan Name]", overview="[Overview]") @mcp.resource("pathfinder://templates/checklist") def implementation_checklist() -> str: """Implementation verification checklist template.""" return """# Implementation Checklist ## Pre-Implementation - [ ] Research documented in research.md - [ ] Plan created with all required sections - [ ] Dependencies identified and documented - [ ] Rollback strategy defined ## During Implementation - [ ] Following plan phases sequentially - [ ] Updating progress.md after each task - [ ] Context utilization < 70% - [ ] Committing changes incrementally ## Post-Implementation - [ ] All plan phases completed - [ ] Tests passing (if applicable) - [ ] Documentation updated - [ ] Session artifacts archived """ @mcp.resource("pathfinder://templates/progress") def progress_template() -> str: """Progress tracking markdown template.""" return PROGRESS_TEMPLATE # ============================================================================= # Prompts # ============================================================================= @mcp.prompt() def generate_plan_from_research(session_id: str) -> list[Message]: """Generate an implementation plan from research findings. Args: session_id: Session ID with completed research """ research_content = artifact_writer.read_artifact(session_id, "research.md") if not research_content: return [ Message( role="user", content=f"No research found for session {session_id}.", ) ] return [ Message( role="user", content=f"""Generate a structured implementation plan from this research. ## Research Findings {research_content} ## Required Format Create plan.md with: 1. YAML frontmatter (name, overview, todos with id/content/status/dependencies) 2. Sections: Architecture Overview, Core Components, Phase 1-N, Implementation Details, Success Criteria """, ) ] @mcp.prompt() def compact_session(session_id: str) -> list[Message]: """Generate a compressed session summary.""" research_content = artifact_writer.read_artifact(session_id, "research.md") or "" plan_content = artifact_writer.read_artifact(session_id, "plan.md") or "" progress_content = artifact_writer.read_artifact(session_id, "progress.md") or "" state = _active_sessions.get(session_id) phase = state.current_phase.value if state else "unknown" return [ Message( role="user", content=f"""Compress session {session_id} (phase: {phase}) into a summary. Research: {research_content[:1500]} Plan: {plan_content[:1500]} Progress: {progress_content[:800]} Include: key decisions, progress, next steps, blockers. Under 500 words. """, ) ] @mcp.prompt() def resume_session(session_id: str) -> list[Message]: """Generate context to resume a session.""" summary = artifact_writer.read_artifact(session_id, "session_summary.md") if summary: return [ Message( role="user", content=f"Resume session {session_id}:\n\n{summary}\n\nWhat's next?", ) ] snapshot = session_manager.load_snapshot(session_id) if not snapshot: return [ Message( role="user", content=f"Session {session_id} not found. Use start_research to begin.", ) ] return [ Message( role="user", content=f"""Resume {session_id}: - Phase: {snapshot.phase.value} - Task: {snapshot.task_description} - Research: {snapshot.research_summary[:400] if snapshot.research_summary else "N/A"} - Plan: {snapshot.plan_summary[:400] if snapshot.plan_summary else "N/A"} What's next? """, ) ] # ============================================================================= # Tools # ============================================================================= @mcp.tool def health_check() -> dict: """Check server health and return status.""" return { "status": "healthy", "server": "PathfinderMCP", "version": "0.1.0", "transport": config.transport, "active_sessions": len(_active_sessions), "context": context_monitor.get_status(), } @mcp.tool def start_research(task_description: str, session_id: str | None = None) -> dict: """Start a new research session. Args: task_description: Description of the task to research session_id: Optional session ID (auto-generated if not provided) """ return research.start_research( task_description=task_description, session_id=session_id, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=_active_sessions, ) @mcp.tool def save_research(session_id: str, findings: str) -> dict: """Save research findings to the session. Args: session_id: Session ID findings: Research findings to save """ return research.save_research( session_id=session_id, findings=findings, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=_active_sessions, ) @mcp.tool async def start_plan(session_id: str, ctx: Context) -> dict: """Transition from research to plan phase. Creates plan.md template. Requires research to be complete. Args: session_id: Session ID """ return await plan.start_plan( session_id=session_id, ctx=ctx, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=_active_sessions, ) @mcp.tool def save_plan(session_id: str, plan_content: str) -> dict: """Save implementation plan. Validates plan follows Cursor plan format with YAML frontmatter. Args: session_id: Session ID plan_content: Plan content in Cursor plan format """ return plan.save_plan( session_id=session_id, plan_content=plan_content, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=_active_sessions, ) @mcp.tool async def implement_phase( session_id: str, phase_number: int | None, ctx: Context ) -> dict: """Execute an implementation phase from the plan. Args: session_id: Session ID phase_number: Specific phase to execute (defaults to next uncompleted) """ return await implement.implement_phase( session_id=session_id, phase_number=phase_number, ctx=ctx, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=_active_sessions, ) @mcp.tool def compact_context(session_id: str) -> dict: """Compress session context into summary artifacts. Use when context utilization exceeds 60%. Args: session_id: Session ID """ return compact.compact_context( session_id=session_id, session_manager=session_manager, artifact_writer=artifact_writer, context_monitor=context_monitor, sessions=_active_sessions, ) def main() -> None: """Run the MCP server with configured transport.""" if config.transport == "sse": mcp.run(transport="sse", host=config.host, port=config.port) else: mcp.run(transport="stdio") if __name__ == "__main__": main()