Pathfinder MCP Server

"""Implementation phase handler.""" import re from typing import Any from fastmcp.server.context import Context from pathfinder_mcp.handlers.base import BaseHandler from pathfinder_mcp.state import Phase class ImplementHandler(BaseHandler): """Handler for implementation phase operations.""" phase = Phase.IMPLEMENT async def execute( # noqa: C901 self, session_id: str, phase_number: int | None = None, ctx: Context | None = None, **kwargs: Any, ) -> dict[str, Any]: """Execute implementation phase. Args: session_id: Session ID phase_number: Specific phase to execute (or next uncompleted) ctx: FastMCP context for elicitation Returns: Phase details and tasks """ # Validate phase (must be PLAN or IMPLEMENT) error = self.validate_phase(session_id, [Phase.PLAN, Phase.IMPLEMENT]) if error: return error # Load plan plan_content = self.artifact_writer.read_artifact(session_id, "plan.md") if not plan_content: return {"error": "Plan not found", "code": "MISSING_PLAN"} # Extract phases phases = self._extract_phases(plan_content) if not phases: return {"error": "No phases in plan", "code": "NO_PHASES"} # Find target phase if phase_number is not None: target = next((p for p in phases if p["number"] == phase_number), None) if not target: return { "error": f"Phase {phase_number} not found", "code": "PHASE_NOT_FOUND", "available_phases": [p["number"] for p in phases], } else: target = next((p for p in phases if not p["completed"]), None) if not target: return { "session_id": session_id, "status": "all_phases_completed", "message": "All phases complete!", } # Elicit confirmation if transitioning state = self.get_session(session_id) if state and state.is_plan and ctx is not None: try: result = await ctx.elicit( f"Begin Phase {target['number']}: {target['title']}?", response_type=["yes", "no"], ) if result.action != "accept" or result.data == "no": return { "session_id": session_id, "phase": "plan", "message": "Implementation cancelled.", "cancelled": True, } except Exception: pass # Transition if needed if state and state.is_plan: new_state = state.transition_to(Phase.IMPLEMENT) self.set_session(session_id, new_state) # Update progress progress = f"\n## Phase {target['number']}: {target['title']}\n\n" progress += "**Status**: In Progress\n\n### Tasks\n" for task in target["tasks"]: status = "x" if task["done"] else " " progress += f"- [{status}] {task['text']}\n" self.artifact_writer.write_progress(session_id, progress) self.context_monitor.add_message(progress) return { "session_id": session_id, "phase": "implement", "executing_phase": { "number": target["number"], "title": target["title"], "tasks": target["tasks"], }, "total_phases": len(phases), "completed_phases": sum(1 for p in phases if p["completed"]), "context": self.get_context_status(), "message": f"Executing Phase {target['number']}.", } def _extract_phases(self, plan_content: str) -> list[dict[str, Any]]: """Extract phase sections from plan.""" phases: list[dict[str, Any]] = [] pattern = ( r"## Phase (\d+):?\s*([^\n]*)\n(.*?)(?=## Phase \d+|## Implementation|$)" ) for match in re.findall(pattern, plan_content, re.DOTALL): num = int(match[0]) title = match[1].strip() content = match[2].strip() tasks = re.findall(r"- \[([ x])\] (.+)", content) phases.append( { "number": num, "title": title, "tasks": [{"done": t[0] == "x", "text": t[1]} for t in tasks], "completed": all(t[0] == "x" for t in tasks) if tasks else False, } ) return phases