#!/usr/bin/env python3 """Recoil Workspace MCP Server — stdio JSON-RPC 2.0. Provides 13 tools for Claude Code to interact with the workspace. Reads JSON-RPC from stdin, writes responses to stdout. Usage (in .claude.json mcpServers config): { "recoil-workspace": { "command": "python3", "args": ["/path/to/recoil/workspace/mcp_server.py"] } } (Note: "workspace" is a reserved MCP name — always use "recoil-workspace".) """ import json import logging import subprocess import sys import threading import traceback import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any, Optional # mcp_server.py runs in a separate process tree from FastAPI (Claude Code # spawns it via stdio JSON-RPC), so it can't import the in-memory BUS # directly — emits go through the loopback bridge route instead. _BUS_BRIDGE_URL = "http://127.0.0.1:8431/api/internal/bus" _BUS_EMIT_TIMEOUT_S = 2.0 class BusEmitFailedError(RuntimeError): """Raised when ``_emit_to_bus`` cannot deliver an event to the BUS bridge.""" def _emit_to_bus( scope: str, severity: str, summary: str, payload: Optional[dict[str, Any]] = None, ) -> None: """POST an event to the FastAPI BUS bridge over loopback.""" body = json.dumps( { "scope": scope, "severity": severity, "summary": summary, "payload": payload or {}, } ).encode("utf-8") req = urllib.request.Request( _BUS_BRIDGE_URL, data=body, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=_BUS_EMIT_TIMEOUT_S) as resp: if resp.status < 200 or resp.status >= 300: raise BusEmitFailedError(f"BUS bridge returned HTTP {resp.status}") except urllib.error.HTTPError as exc: raise BusEmitFailedError( f"BUS bridge HTTP error {exc.code}: {exc.reason}" ) from exc except urllib.error.URLError as exc: raise BusEmitFailedError(f"BUS bridge unreachable: {exc.reason}") from exc # ── Path setup ────────────────────────────────────────────────── # Both roots are needed: CLAUDE_PROJECTS so `from recoil.X` resolves # (Build D Phase 18.2 import flip), recoil/ so `from workspace.X` still # resolves for the not-yet-flipped imports below. _RECOIL_ROOT = Path(__file__).resolve().parent.parent _PROJECTS_ROOT = _RECOIL_ROOT.parent for _p in (_PROJECTS_ROOT, _RECOIL_ROOT): if str(_p) not in sys.path: sys.path.insert(0, str(_p)) from recoil.core.paths import ProjectPaths, projects_root from recoil.core.ref_resolver import ( get_all_project_refs, serialize_refs_for_workspace, validate_all_project_refs, ) from recoil.workspace import board from workspace import state as ws_state from workspace import session_log from workspace import sidecar as ws_sidecar from workspace import verdict # Phase 3 — verdict sidecar emission from recoil.core.exceptions import SidecarCorruptError # ── Logging (stderr only — stdout is the JSON-RPC channel) ───── logging.basicConfig( stream=sys.stderr, level=logging.INFO, format="[workspace-mcp] %(levelname)s %(message)s", ) log = logging.getLogger("workspace-mcp") # ── Tool Registry ─────────────────────────────────────────────── _TOOLS: dict[str, dict] = {} def _register_tool(name: str, description: str, input_schema: dict): """Decorator to register an MCP tool.""" def decorator(fn): _TOOLS[name] = { "name": name, "description": description, "inputSchema": input_schema, "handler": fn, } return fn return decorator # ── Shared helpers ────────────────────────────────────────────── from recoil.workspace.helpers import ( get_store as _get_store, get_ops_log_path as _get_ops_log_path, shot_status_color as _shot_status_color, ) # _shot_status_color imported from recoil.workspace.helpers # ── Phase 3: Verdict sidecar emission helper ──────────────────── def _emit_verdict_sidecar( *, verdict_value: str, project: str, shot: dict, shot_id: str, take: dict, takes: list, reason: str, provenance_hash: str | None, params: dict, ) -> None: """Build + write the verdict sidecar for approve/reject. Wrapped in try/except by callers — failures must not break the action. """ taxonomy = params.get("taxonomy", "taste-shaped") sub_tags = params.get("sub_tags") or [] jt_action = params.get("jt_action", "skip") claude_proposed_reason = params.get("claude_proposed_reason", "") reason_source = { "confirm": "jt_confirmed", "correct": "jt_corrected", "qualify": "jt_added", "skip": "claude_only", }.get(jt_action, "claude_only") episode_id = shot.get("episode_id") or "ep_001" try: take_number = takes.index(take) + 1 except ValueError: take_number = len(takes) auto_filled = verdict.build_auto_filled( project=project, shot_id=shot_id, take_number=take_number, chat_context_window=None, episode_id=episode_id, ) confirmation = { "claude_proposed_taxonomy": taxonomy, "claude_proposed_sub_tags": sub_tags, "claude_proposed_reason": claude_proposed_reason, "jt_action": jt_action, "jt_correction": None, } verdict.write_verdict( project=project, episode_id=episode_id, shot_id=shot_id, take_number=take_number, verdict=verdict_value, taxonomy=taxonomy, sub_tags=sub_tags, reason_text=reason or claude_proposed_reason or "", reason_source=reason_source, auto_filled=auto_filled, confirmation=confirmation, generation_id=provenance_hash, media_path=take.get("file_path"), ) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # CONTEXT TOOLS (Tools 1, 5, 6) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ── Tool 1: prime_project ─────────────────────────────────────── @_register_tool( name="prime_project", description=( "Load project context for the workspace session. Returns shot counts by status, " "pending review count, character and location names with hero image paths, " "active episodes, and last 10 ops log entries. Text only — no images loaded." ), input_schema={ "type": "object", "properties": { "name": { "type": "string", "description": "Project name (e.g. 'tartarus', 'the-afterimage')", }, }, "required": ["name"], }, ) def tool_prime_project(params: dict) -> dict: project = params["name"] # Validate project exists project_dir = projects_root() / project if not project_dir.is_dir(): return {"error": f"Project '{project}' not found at {project_dir}"} # Set as active project in workspace state ws_state.set_project(project) # Health checks shots_dir = ProjectPaths.for_project(project).shots_dir sessions_dir = projects_root() / project / "sessions" health = { "shots_dir_readable": shots_dir.is_dir(), "sessions_dir_writable": True, } try: sessions_dir.mkdir(parents=True, exist_ok=True) except OSError: health["sessions_dir_writable"] = False # Check for Dropbox conflicts conflicts = ws_state.detect_dropbox_conflicts(shots_dir) # Get shot summary store = _get_store(project) summary = store.summary() all_shots = store.get_all_shots() store.close() # Group by episode episodes = {} for shot in all_shots: ep = shot.get("episode_id", "unknown") if ep not in episodes: episodes[ep] = {"total": 0, "by_status": {}} episodes[ep]["total"] += 1 status = shot.get("status", "previs_pending") episodes[ep]["by_status"][status] = episodes[ep]["by_status"].get(status, 0) + 1 # Pending review count pending_statuses = { "previs_generated", "keyframe_generated", "video_complete", "needs_review", "icu_escalated", "pending_qc", } pending_review = sum( 1 for shot in all_shots if shot.get("status") in pending_statuses ) # Get refs (paths only, no images) + validation under v2 layout. # Resolve once, validate against the resolved tree (no double scan). paths = ProjectPaths.for_project(project) raw_refs = get_all_project_refs(paths) refs = serialize_refs_for_workspace(raw_refs, paths.project_root) ref_issues = validate_all_project_refs(paths, refs_tree=raw_refs) ref_warnings = [f"{i.entity_type}/{i.entity_id}: {i.message}" for i in ref_issues] # Get sidecar status counts sidecar_counts = { "candidate": 0, "pinned": 0, "canonical": 0, "archived": 0, "no_sidecar": 0, } # v2 layout — sidecars live under assets/, sequences/, renders/, state/. # _history/ is the archive root and is skipped (matches workspace/tree.py policy). from recoil.workspace.sidecar import MEDIA_EXTENSIONS as _SC_MEDIA_EXT _MEDIA_ROOTS = ("assets", "sequences", "renders", "state") for root_name in _MEDIA_ROOTS: root = project_dir / root_name if not root.is_dir(): continue for media_file in root.rglob("*"): if not media_file.is_file(): continue if media_file.suffix.lower() not in _SC_MEDIA_EXT: continue parts = media_file.relative_to(project_dir).parts if any(p.startswith(".") or p in ("_meta", "_history") for p in parts): continue try: sc = ws_sidecar.read_sidecar(media_file) except SidecarCorruptError as e: log.warning( "mcp_server: skipping corrupt sidecar at %s — %s", media_file, e, ) sidecar_counts["no_sidecar"] += 1 continue if sc is None: sidecar_counts["no_sidecar"] += 1 else: st = sc.get("status", "candidate") if st in sidecar_counts: sidecar_counts[st] += 1 # Get last 10 ops log entries ops_log_path = _get_ops_log_path(project) recent_ops = [] if ops_log_path.is_file(): lines = ops_log_path.read_text(encoding="utf-8").strip().split("\n") for line in reversed(lines[-20:]): # Read last 20, take 10 after dedup try: record = json.loads(line) recent_ops.append(record) if len(recent_ops) >= 10: break except (json.JSONDecodeError, TypeError): continue # Check for existing session log (context recovery) existing_session = session_log.read_entries(project, projects_root()) session_status = ( f"Resuming session ({len(existing_session)} entries today)" if existing_session else "New session" ) # Log the prime event session_log.append_entry( project, projects_root(), "prime", data={"total_shots": summary["total_shots"], "pending_review": pending_review}, ) result = { "project": project, "session_status": session_status, "health": health, "conflicts": conflicts, "summary": { "total_shots": summary["total_shots"], "total_cost": summary["total_cost"], "by_status": summary["by_status"], }, "episodes": episodes, "pending_review": pending_review, "characters": refs["characters"], "locations": refs["locations"], "props": refs.get("props", {}), "ref_warnings": ref_warnings if ref_warnings else [], "sidecar_counts": sidecar_counts, "recent_ops": recent_ops, } if conflicts: result["warning"] = ( f"Dropbox conflicts detected: {len(conflicts)} file(s). " "Do NOT auto-resolve — surface to JT for manual resolution." ) return result # ── Tool 5: get_shot_detail ───────────────────────────────────── @_register_tool( name="get_shot_detail", description=( "Get full shot state JSON including all takes with provenance, " "gate results, cost, status, and take file paths. Also checks for " "Dropbox conflict files for this shot." ), input_schema={ "type": "object", "properties": { "shot_id": { "type": "string", "description": "Shot identifier (e.g. 'EP001_SH03')", }, }, "required": ["shot_id"], }, ) def tool_get_shot_detail(params: dict) -> dict: shot_id = params["shot_id"] project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} store = _get_store(project) shot = store.get_shot(shot_id) store.close() if shot is None: return {"error": f"Shot '{shot_id}' not found in project '{project}'"} # Resolve take file paths to absolute paths for viewer takes = shot.get("takes", []) for take in takes: fp = take.get("file_path", "") if fp and not fp.startswith("/"): take["absolute_path"] = str(projects_root() / project / fp) # Check for Dropbox conflicts on this specific shot shots_dir = ProjectPaths.for_project(project).shots_dir shot_conflicts = ( [ str(p) for p in shots_dir.iterdir() if shot_id in p.name and "conflicted copy" in p.name.lower() ] if shots_dir.is_dir() else [] ) # Attach sidecar data to takes that have file paths for take in takes: fp = take.get("file_path", "") if fp: abs_fp = ( projects_root() / project / fp if not fp.startswith("/") else Path(fp) ) try: sc = ws_sidecar.read_sidecar(abs_fp) except SidecarCorruptError as e: log.warning( "mcp_server: corrupt sidecar at %s — %s; take metadata omitted", abs_fp, e, ) sc = None if sc: take["sidecar_status"] = sc.get("status") take["sidecar_notes"] = sc.get("notes", "") take["sidecar_source"] = sc.get("source", "") result = { "shot_id": shot_id, "project": project, "episode_id": shot.get("episode_id", ""), "status": shot.get("status", "previs_pending"), "status_color": _shot_status_color(shot.get("status", "previs_pending")), "pipeline": shot.get("pipeline"), "model": shot.get("model"), "cost_incurred": shot.get("cost_incurred", 0), "retry_waste_cost": shot.get("retry_waste_cost", 0), "attempts": shot.get("attempts", 0), "max_attempts": shot.get("max_attempts", 3), "gate_results": shot.get("gate_results", {}), "output_path": shot.get("output_path"), "error_message": shot.get("error_message"), "takes": takes, "take_count": len(takes), "is_coverage": shot.get("is_coverage", False), "coverage_of": shot.get("coverage_of"), "updated_at": shot.get("updated_at"), } if shot_conflicts: result["dropbox_conflicts"] = shot_conflicts result["warning"] = ( f"Dropbox conflict detected for {shot_id}. Manual resolution required." ) return result # ── Tool 6: get_shot_neighbors ────────────────────────────────── @_register_tool( name="get_shot_neighbors", description=( "Get previous and next shot IDs in episode sequence with status and " "latest take path. For continuity reasoning — helps Claude understand " "what comes before and after this shot." ), input_schema={ "type": "object", "properties": { "shot_id": { "type": "string", "description": "Shot identifier (e.g. 'EP001_SH03')", }, }, "required": ["shot_id"], }, ) def tool_get_shot_neighbors(params: dict) -> dict: shot_id = params["shot_id"] project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} store = _get_store(project) shot = store.get_shot(shot_id) if shot is None: store.close() return {"error": f"Shot '{shot_id}' not found"} episode_id = shot.get("episode_id", "") episode_shots = store.get_shots_by_episode(episode_id) store.close() # Sort by shot number (numeric, not string — fixes 10+ shot ordering) import re as _re def _shot_sort_key(s): m = _re.search(r"(\d+)$", s.get("shot_id", "0")) return int(m.group(1)) if m else 0 episode_shots.sort(key=_shot_sort_key) # Find the current shot's index idx = None for i, s in enumerate(episode_shots): if s.get("shot_id") == shot_id: idx = i break if idx is None: return {"error": f"Shot '{shot_id}' not found in episode '{episode_id}' list"} def _shot_summary(s: dict) -> dict: takes = s.get("takes", []) latest_take_path = None if takes: last = takes[-1] fp = last.get("file_path", "") if fp and not fp.startswith("/"): latest_take_path = str(projects_root() / project / fp) else: latest_take_path = fp return { "shot_id": s.get("shot_id"), "status": s.get("status", "previs_pending"), "status_color": _shot_status_color(s.get("status", "previs_pending")), "latest_take_path": latest_take_path, "take_count": len(takes), } result = { "shot_id": shot_id, "episode_id": episode_id, "position": f"{idx + 1} of {len(episode_shots)}", "previous": _shot_summary(episode_shots[idx - 1]) if idx > 0 else None, "next": _shot_summary(episode_shots[idx + 1]) if idx < len(episode_shots) - 1 else None, } return result # ── Tool: get_episode_board ───────────────────────────────────── def get_episode_board(project: str, episode_id: str) -> dict: return board.build_episode_board(project, episode_id) @_register_tool( name="get_episode_board", description=( "Returns the episode board wall payload for an episode, including " "storyboard artifacts, photoreal artifacts, batch panels, and coverage summary." ), input_schema={ "type": "object", "properties": { "project": { "type": "string", "description": "Project name (e.g. 'tartarus', 'the-afterimage').", }, "episode_id": { "type": "string", "description": "Episode id, accepting forms like '1', 'ep_001', or 'EP001'.", }, }, "required": ["project", "episode_id"], }, ) def tool_get_episode_board(params: dict) -> dict: return get_episode_board(params["project"], params["episode_id"]) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # SELECTION + VIEWER TOOLS (Tools 2, 3, 4) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ── Tool 2: get_selection ─────────────────────────────────────── @_register_tool( name="get_selection", description=( "Returns the list of currently selected shot IDs in the workspace file browser. " "Supports multi-select — may return 0 or more IDs." ), input_schema={ "type": "object", "properties": {}, }, ) def tool_get_selection(params: dict) -> dict: project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} selection = ws_state.get_selection() return { "project": project, "selection": selection, "count": len(selection), } # ── Tool 3: get_viewer_state ─────────────────────────────────── @_register_tool( name="get_viewer_state", description=( "Returns the current viewer state: what shot/take is being displayed, " "the file path, media type (image/video), and any context annotation." ), input_schema={ "type": "object", "properties": {}, }, ) def tool_get_viewer_state(params: dict) -> dict: project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} viewer = ws_state.get_viewer_state() return { "project": project, "viewer": viewer, } # ── Tool 4: show_in_viewer ───────────────────────────────────── @_register_tool( name="show_in_viewer", description=( "Push a specific file (image or video) to the workspace viewer pane. " "Can display any media file from the project. Optional context annotation " "describes why this is being shown." ), input_schema={ "type": "object", "properties": { "path": { "type": "string", "description": ( "File path to display. Can be absolute or relative to project root " "(e.g. 'renders/ep_001/shot_003_take2.png')." ), }, "context": { "type": "string", "description": "Optional context annotation (e.g. 'Comparing framing with previous take').", }, }, "required": ["path"], }, ) def tool_show_in_viewer(params: dict) -> dict: project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} path = params["path"] context = params.get("context") # Resolve relative paths against project root if not path.startswith("/"): abs_path = str(projects_root() / project / path) else: abs_path = path # Validate file exists if not Path(abs_path).is_file(): return {"error": f"File not found: {abs_path}"} # Compute the path relative to projects_root() for the media endpoint # Let set_viewer_state auto-detect media_type from extension try: rel_path = str(Path(abs_path).relative_to(projects_root())) except ValueError: rel_path = path ws_state.set_viewer_state( file_path=rel_path, context=context, ) # Sync sidebar selection to the shown file (project-relative path) try: project_rel = str(Path(abs_path).relative_to(projects_root() / project)) ws_state.set_selection([project_rel]) except (ValueError, Exception): pass # Log to session session_log.append_entry( project, projects_root(), "action", data={ "action": "show_in_viewer", "path": rel_path, "context": context, }, ) # Read back the auto-detected media_type viewer = ws_state.get_viewer_state() return { "displayed": rel_path, "media_type": viewer.get("media_type"), "context": context, } # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ACTION TOOLS (Tools 7, 8, 9) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ── Tool 7: approve_shot ──────────────────────────────────────── @_register_tool( name="approve_shot", description=( "Mark a shot take as approved. Updates the shot status in ExecutionStore " "and logs the approval with provenance hash to the session log." ), input_schema={ "type": "object", "properties": { "shot_id": { "type": "string", "description": "Shot identifier (e.g. 'EP001_SH03')", }, "take_id": { "type": "string", "description": "Take identifier (e.g. 'EP001_SH03_T11195'). If not provided, approves the latest take.", }, "reason": { "type": "string", "description": "Optional reason for approval.", }, "taxonomy": { "type": "string", "enum": ["validator-escape", "taste-shaped", "strategic"], "description": "Three-way failure/success taxonomy. Claude proposes; JT confirms in chat.", }, "sub_tags": { "type": "array", "items": {"type": "string"}, "description": "Sub-tags drawn from editorial_tags.json (17 vocabulary). Claude proposes; JT confirms or corrects.", }, "claude_proposed_reason": { "type": "string", "description": "Claude's guess at the reason from chat context. Logged separately from the confirmed reason.", }, "jt_action": { "type": "string", "enum": ["confirm", "correct", "qualify", "skip"], "description": "How JT responded to Claude's proposed taxonomy/reason. Drives the confirm-vs-correct ratio metric.", }, }, "required": ["shot_id"], }, ) def tool_approve_shot(params: dict) -> dict: shot_id = params["shot_id"] take_id = params.get("take_id") reason = params.get("reason", "") project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} store = _get_store(project) shot = store.get_shot(shot_id) if shot is None: store.close() return {"error": f"Shot '{shot_id}' not found"} current_status = shot.get("status", "previs_pending") # Determine the target approved status based on current status if "previs" in current_status: target_status = "previs_approved" elif "keyframe" in current_status: target_status = "keyframe_approved" elif "video" in current_status: target_status = "approved" else: target_status = "approved" # Find the take takes = shot.get("takes", []) if not takes: store.close() return {"error": f"Shot '{shot_id}' has no takes"} if take_id: matching = [t for t in takes if t.get("take_id") == take_id] if not matching: store.close() return {"error": f"Take '{take_id}' not found on shot '{shot_id}'"} take = matching[0] else: take = takes[-1] take_id = take.get("take_id", f"take_{len(takes)}") # Extract provenance hash if available provenance_hash = take.get("provenance_hash") or take.get("inputs_snapshot_hash") # FIX (Gemini #3, Opus H3): Single atomic update — status + gate_results # in one call, not two separate store opens. try: store.update_shot( shot_id, status=target_status, gate_results={"hero_frame": take.get("file_path")}, ) except Exception as e: store.close() return {"error": f"Failed to update shot status: {e}"} store.close() # Log approval to session log with provenance session_log.append_entry( project, projects_root(), "action", shot_id=shot_id, take_id=take_id, provenance_hash=provenance_hash, data={ "action": "approve_shot", "from_status": current_status, "to_status": target_status, "reason": reason, "take_file": take.get("file_path"), }, ) # ── Phase 3: Verdict sidecar (approve) ──────────────────────── try: _emit_verdict_sidecar( verdict_value="approve", project=project, shot=shot, shot_id=shot_id, take=take, takes=takes, reason=reason, provenance_hash=provenance_hash, params=params, ) except Exception as e: log.warning("Verdict sidecar write failed for %s: %s", shot_id, e) return { "shot_id": shot_id, "take_id": take_id, "previous_status": current_status, "new_status": target_status, "provenance_hash": provenance_hash, "reason": reason, } # ── Tool 8: reject_shot ───────────────────────────────────────── @_register_tool( name="reject_shot", description=( "Mark a shot take as rejected. Updates the shot status and logs " "the rejection with reason to the session log. Optionally classifies " "the failure mode for the retry-strategy learning engine." ), input_schema={ "type": "object", "properties": { "shot_id": { "type": "string", "description": "Shot identifier", }, "take_id": { "type": "string", "description": "Take identifier. If not provided, rejects the latest take.", }, "reason": { "type": "string", "description": "Reason for rejection (e.g. 'flat lighting', 'wrong framing').", }, "failure_mode": { "type": "string", "description": ( "Failure classification for retry strategy. One of: identity_drift, " "composition_wrong, motion_failure, style_drift, cuts_too_soft, " "ref_bleed, content_filter_hard_block, unknown. Defaults to 'unknown'." ), }, "taxonomy": { "type": "string", "enum": ["validator-escape", "taste-shaped", "strategic"], "description": "Three-way failure/success taxonomy. Claude proposes; JT confirms in chat.", }, "sub_tags": { "type": "array", "items": {"type": "string"}, "description": "Sub-tags drawn from editorial_tags.json (17 vocabulary). Claude proposes; JT confirms or corrects.", }, "claude_proposed_reason": { "type": "string", "description": "Claude's guess at the reason from chat context. Logged separately from the confirmed reason.", }, "jt_action": { "type": "string", "enum": ["confirm", "correct", "qualify", "skip"], "description": "How JT responded to Claude's proposed taxonomy/reason. Drives the confirm-vs-correct ratio metric.", }, }, "required": ["shot_id"], }, ) def tool_reject_shot(params: dict) -> dict: shot_id = params["shot_id"] take_id = params.get("take_id") reason = params.get("reason", "") failure_mode_raw = params.get("failure_mode", "unknown") # Validate against FailureMode enum — fall back to "unknown" for garbage values valid_modes = { "identity_drift", "composition_wrong", "motion_failure", "style_drift", "cuts_too_soft", "content_filter_hard_block", "ref_bleed", "prompt_duration_mismatch", "cost_overrun", "transient", "unknown", "gate_mechanical", } failure_mode = failure_mode_raw if failure_mode_raw in valid_modes else "unknown" project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} store = _get_store(project) shot = store.get_shot(shot_id) if shot is None: store.close() return {"error": f"Shot '{shot_id}' not found"} current_status = shot.get("status", "previs_pending") # Determine the target rejected status if "previs" in current_status: target_status = "previs_rejected" elif "keyframe" in current_status: target_status = "keyframe_rejected" elif "video" in current_status: target_status = "video_rejected" else: target_status = "rejected" # Find the take takes = shot.get("takes", []) if not takes: store.close() return {"error": f"Shot '{shot_id}' has no takes"} if take_id: matching = [t for t in takes if t.get("take_id") == take_id] if not matching: store.close() return {"error": f"Take '{take_id}' not found on shot '{shot_id}'"} take = matching[0] else: take = takes[-1] take_id = take.get("take_id", f"take_{len(takes)}") provenance_hash = take.get("provenance_hash") or take.get("inputs_snapshot_hash") # Mark the take as rejected (take already resolved above) take["rejected"] = True take["rejection_reason"] = reason take["failure_mode"] = failure_mode try: store.update_shot(shot_id, status=target_status, takes=takes) except Exception as e: store.close() return {"error": f"Failed to update shot: {e}"} store.close() # Ingest into LearningEngine for retry strategy learning try: from recoil.pipeline.orchestrator.learning_engine import LearningEngine le = LearningEngine(project=project) le.ingest_retry( shot_id=shot_id, failure_mode=failure_mode, source="human_reject", notes=reason, ) le.flush() except Exception as e: log.warning("Could not log reject to LearningEngine: %s", e) # Log rejection session_log.append_entry( project, projects_root(), "action", shot_id=shot_id, take_id=take_id, provenance_hash=provenance_hash, data={ "action": "reject_shot", "from_status": current_status, "to_status": target_status, "reason": reason, "failure_mode": failure_mode, }, ) # ── Phase 3: Verdict sidecar (reject) ───────────────────────── try: _emit_verdict_sidecar( verdict_value="reject", project=project, shot=shot, shot_id=shot_id, take=take, takes=takes, reason=reason, provenance_hash=provenance_hash, params=params, ) except Exception as e: log.warning("Verdict sidecar write failed for %s: %s", shot_id, e) return { "shot_id": shot_id, "take_id": take_id, "previous_status": current_status, "new_status": target_status, "failure_mode": failure_mode, "reason": reason, } # ── submit_generation: dispatch helpers ───────────────────────── _DISPATCH_SCRIPT = _RECOIL_ROOT / "pipeline" / "tools" / "dispatch_cli.py" def _build_dispatch_argv( *, project: str, shot_id: str, model: str, override_prompt: str | None, ) -> list[str]: """Build the dispatch_cli.py argv for an LLM-driven workspace dispatch. Pure function — testable without invoking subprocess. The override_prompt fully REPLACES the shot's plan-derived prompt; it does not merge. """ argv: list[str] = [ sys.executable, str(_DISPATCH_SCRIPT), "--project", project, "--shot", shot_id, "--model", model, ] if override_prompt: argv += ["--prompt", override_prompt] return argv def _run_dispatch_async( *, op_id: str, ops_log_path: Path, argv: list[str], cwd: Path, ) -> None: """Run the dispatch subprocess and log its outcome to ops_log. Designed to run in a daemon thread. Never raises; surfaces all errors as log_op_failed / log_op_crashed records so the MCP caller can poll. """ from recoil.pipeline._lib.ops_log import ( log_op_completed, log_op_failed, log_op_crashed, ) try: proc = subprocess.run( argv, cwd=str(cwd), capture_output=True, text=True, timeout=1800, # 30 min hard cap ) if proc.returncode == 0: log_op_completed( ops_log_path, op_id=op_id, outputs={ "returncode": 0, "stdout_tail": proc.stdout[-2000:] if proc.stdout else "", }, ) else: log_op_failed( ops_log_path, op_id=op_id, error=f"dispatch exited {proc.returncode}", issues=[ (proc.stderr or "")[-2000:], (proc.stdout or "")[-1000:], ], ) except subprocess.TimeoutExpired: log_op_failed( ops_log_path, op_id=op_id, error="dispatch timed out after 1800s", ) except Exception as e: log_op_crashed( ops_log_path, op_id=op_id, error=str(e), traceback_text=traceback.format_exc(), ) # ── Tool 9: submit_generation ─────────────────────────────────── @_register_tool( name="submit_generation", description=( "Submit a new generation take for a shot. Returns an op_id immediately; " "the generation runs asynchronously via dispatch_cli.py in a " "daemon thread. Poll status with get_activity(). " "Adjustments (framing, lighting, mood, action, continuity, or freeform " "keys) are concatenated as a prompt override — this REPLACES the shot's " "plan-derived prompt. Refs/locations resolve automatically from the " "project plan." ), input_schema={ "type": "object", "properties": { "shot_id": { "type": "string", "description": "Shot identifier", }, "adjustments": { "type": "object", "description": ( "Structured adjustments dict. Keys: framing, lighting, mood, " "action, continuity, or any freeform key. Values are natural " "language strings describing the desired change." ), }, "model": { "type": "string", "description": "Override model (e.g. 'kling-v3', 'seeddance-2.0'). Uses current shot model if not specified.", }, }, "required": ["shot_id"], }, ) def tool_submit_generation(params: dict) -> dict: shot_id = params["shot_id"] adjustments = params.get("adjustments", {}) model_override = params.get("model") project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} store = _get_store(project) shot = store.get_shot(shot_id) if shot is None: store.close() return {"error": f"Shot '{shot_id}' not found"} # Mark intent on the shot (kept for UI status reflection). try: store.update_shot(shot_id, generation_requested=True) except Exception: pass # Non-fatal — field may not be recognized by all store versions store.close() from recoil.pipeline._lib.ops_log import make_op_id, log_op_started op_id = make_op_id() override_parts = [] for key, value in adjustments.items(): override_parts.append(f"[{key.upper()}] {value}") override_text = " | ".join(override_parts) if override_parts else "" model = model_override or shot.get("model") or "kling-v3" argv = _build_dispatch_argv( project=project, shot_id=shot_id, model=model, override_prompt=override_text or None, ) ops_log_path = _get_ops_log_path(project) log_op_started( ops_log_path, op_id=op_id, name="workspace_generation", args={ "shot_id": shot_id, "adjustments": adjustments, "model": model, "override_text": override_text, "argv": argv, }, context={"operator": "workspace_mcp"}, ) session_log.append_entry( project, projects_root(), "generation_submitted", shot_id=shot_id, data={ "op_id": op_id, "adjustments": adjustments, "model": model, "override_text": override_text, }, ) threading.Thread( target=_run_dispatch_async, kwargs={ "op_id": op_id, "ops_log_path": ops_log_path, "argv": argv, "cwd": _RECOIL_ROOT, }, daemon=True, name=f"workspace-dispatch-{op_id}", ).start() return { "op_id": op_id, "shot_id": shot_id, "status": "dispatched", "model": model, "adjustments": adjustments, "override_text": override_text, "note": ( "Generation dispatched via dispatch_cli.py in a daemon thread. " "Poll get_activity() for completion status. Adjustments fully override " "the shot's plan prompt; refs auto-resolve from the project plan." ), } # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # FEEDBACK + ACTIVITY TOOLS (Tools 10, 11, 12) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ── Tool 10: log_feedback ─────────────────────────────────────── @_register_tool( name="log_feedback", description=( "Log structured feedback linked to a shot's provenance hash. " "This is the primary learning loop data: JT's comments tied to " "the exact combination of prompt, model, refs, and settings that " "produced the output. Categories: pattern, correction, observation, " "preference, trivial." ), input_schema={ "type": "object", "properties": { "shot_id": { "type": "string", "description": "Shot identifier", }, "take_id": { "type": "string", "description": "Take identifier", }, "text": { "type": "string", "description": "The feedback text (JT's comment or Claude's observation)", }, "action": { "type": "string", "description": "Action taken in response (e.g. 'approve', 'reject', 'reroll', 'adjust_prompt')", }, "category": { "type": "string", "enum": [ "pattern", "correction", "observation", "preference", "trivial", ], "description": ( "Feedback category. pattern=recurring issue, correction=specific fix, " "observation=notable finding, preference=taste signal, trivial=low-value" ), }, }, "required": ["shot_id", "take_id", "text", "action", "category"], }, ) def tool_log_feedback(params: dict) -> dict: shot_id = params["shot_id"] take_id = params["take_id"] text = params["text"] action = params["action"] category = params["category"] project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} # Look up the take's provenance hash store = _get_store(project) shot = store.get_shot(shot_id) store.close() provenance_hash = None if shot: for t in shot.get("takes", []): if t.get("take_id") == take_id: provenance_hash = t.get("provenance_hash") or t.get( "inputs_snapshot_hash" ) break entry = session_log.append_entry( project, projects_root(), "feedback", shot_id=shot_id, take_id=take_id, provenance_hash=provenance_hash, data={ "text": text, "action": action, "category": category, "model": shot.get("model") if shot else None, "status": shot.get("status") if shot else None, }, ) return { "logged": True, "shot_id": shot_id, "take_id": take_id, "category": category, "provenance_hash": provenance_hash, "entry": entry, } # ── Tool 11: get_session_log ─────────────────────────────────── @_register_tool( name="get_session_log", description=( "Get session history for context recovery after compaction. " "Returns recent session log entries. If 'since' is provided, " "only returns entries after that timestamp." ), input_schema={ "type": "object", "properties": { "since": { "type": "string", "description": "ISO 8601 timestamp. Only return entries after this time.", }, }, }, ) def tool_get_session_log(params: dict) -> dict: since = params.get("since") project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} entries = session_log.read_entries(project, projects_root(), since=since) # Summarize by type by_type = {} for e in entries: t = e.get("type", "unknown") by_type[t] = by_type.get(t, 0) + 1 return { "project": project, "total_entries": len(entries), "by_type": by_type, "entries": entries[-50:], # Last 50 to avoid context bloat "truncated": len(entries) > 50, } # ── Tool 12: get_activity ────────────────────────────────────── @_register_tool( name="get_activity", description=( "Get in-flight generations and recent completions/failures. " "Reads the ops log to find pending operations (no completion record) " "and recent completed/failed operations." ), input_schema={ "type": "object", "properties": {}, }, ) def tool_get_activity(params: dict) -> dict: project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} ops_log_path = _get_ops_log_path(project) # Get dangling (in-flight) operations from recoil.pipeline._lib.ops_log import scan_for_dangling_ops in_flight = scan_for_dangling_ops(ops_log_path) # Get recent completed/failed (last 20 lines of ops log) recent = [] if ops_log_path.is_file(): lines = ops_log_path.read_text(encoding="utf-8").strip().split("\n") for line in reversed(lines[-40:]): try: record = json.loads(line) if record.get("status") in ("completed", "failed", "crashed"): recent.append(record) if len(recent) >= 10: break except (json.JSONDecodeError, TypeError): continue return { "project": project, "in_flight": in_flight, "in_flight_count": len(in_flight), "recent_completed": [r for r in recent if r.get("status") == "completed"][:5], "recent_failed": [ r for r in recent if r.get("status") in ("failed", "crashed") ][:5], } # ── Tool 13: get_file_provenance ────────────────────────────── @_register_tool( name="get_file_provenance", description=( "Read the universal sidecar for any media file. Returns full provenance " "(model, prompt, refs, cost, gates), status, lineage, and notes. " "Works for both pipeline-generated and manually-dropped files." ), input_schema={ "type": "object", "properties": { "path": { "type": "string", "description": ( "File path. Can be absolute or relative to project root " "(e.g. 'assets/char/sadie/base/sadie_identity_hero_v04.jpeg')." ), }, }, "required": ["path"], }, ) def tool_get_file_provenance(params: dict) -> dict: project = ws_state.get_project() if not project: return {"error": "No project active. Call prime_project first."} path = params["path"] # Resolve relative paths against project root if not path.startswith("/"): abs_path = projects_root() / project / path else: abs_path = Path(path) # Security check — ensure path is within project directory try: abs_path.resolve().relative_to((projects_root() / project).resolve()) except ValueError: return {"error": "Access denied: path is outside of the project directory."} if not abs_path.is_file(): return {"error": f"File not found: {abs_path}"} # Read the universal sidecar — corruption surfaces explicitly so the # MCP client sees the error rather than silent fallthrough to _meta/. try: data = ws_sidecar.read_sidecar(abs_path) except SidecarCorruptError as e: return { "error": "corrupt sidecar", "path": str(abs_path), "detail": str(e), } if data is None: # Try the _meta/ sidecar (for canonical refs) meta_sidecar = abs_path.parent / "_meta" / f"{abs_path.name}.json" if meta_sidecar.is_file(): try: import json as _json data = _json.loads(meta_sidecar.read_text(encoding="utf-8")) data["_sidecar_format"] = "canonical_meta" except FileNotFoundError: # TOCTOU race; treat as no _meta sidecar. pass except (json.JSONDecodeError, OSError) as e: # Tenet 6: surface corrupt _meta to the MCP client rather # than silent fallthrough. return { "error": "corrupt _meta sidecar", "path": str(meta_sidecar), "detail": str(e), } # Phase 8 — also check for a synthetic reclaim meta.yaml sidecar. # Reclaim sidecars are written by pipeline/tools/reclaim_orphans.py # at {parent}/{shot_id}_meta.yaml and carry reclaim.synthetic: true. reclaim_badge = None try: # Derive shot_id by iteratively stripping trailing TAKE / VERSION / # MODEL suffixes — must match # pipeline.tools.reclaim_orphans.parse_orphan_filename so the lookup # finds reclaimed metas for filenames like # `shot_123_take1_v2_kling.mp4` (which become shot_id=`123`). We # replicate the loop locally rather than importing to avoid pulling # the full reclaim module (and its yaml/ffprobe deps) into the MCP # server boot path. import re as _re import yaml as _yaml _TAKE_RE = _re.compile(r"_take(\d+)$", _re.IGNORECASE) _VERSION_RE = _re.compile(r"_v(\d+)$", _re.IGNORECASE) _MODEL_HINTS_RE = _re.compile( r"_(seedance|kling|veo|nbp|gemini)$", _re.IGNORECASE ) stem = abs_path.stem if stem.startswith("shot_"): stem = stem[len("shot_") :] take_number = 1 while True: m = _TAKE_RE.search(stem) if m: # Match reclaim_orphans.parse_orphan_filename: each `_takeN$` # match overwrites take_number, so the innermost (leftmost) # take suffix wins after the loop strips outer tokens. try: take_number = int(m.group(1)) except (TypeError, ValueError): pass stem = _TAKE_RE.sub("", stem) continue if _VERSION_RE.search(stem): stem = _VERSION_RE.sub("", stem) continue if _MODEL_HINTS_RE.search(stem): stem = _MODEL_HINTS_RE.sub("", stem) continue break shot_id = stem.strip("_") # reclaim_orphans writes `{shot_id}_take{N}_meta.yaml` when take > 1 # to disambiguate multi-take collisions (take==1 keeps the canonical # `{shot_id}_meta.yaml` name). Look for the take-suffixed name first # when take_number > 1, then fall back to the canonical name. meta_yaml = None if take_number > 1: take_meta = abs_path.parent / f"{shot_id}_take{take_number}_meta.yaml" if take_meta.is_file(): meta_yaml = take_meta if meta_yaml is None: meta_yaml = abs_path.parent / f"{shot_id}_meta.yaml" if meta_yaml.is_file(): ymeta = _yaml.safe_load(meta_yaml.read_text(encoding="utf-8")) or {} reclaim = ymeta.get("reclaim") or {} if reclaim.get("synthetic"): reclaim_badge = { "synthetic": True, "confidence": reclaim.get("confidence"), "inference_sources": reclaim.get("inference_sources", []), "unknown_fields": reclaim.get("unknown_fields", []), "reclaimed_at": reclaim.get("reclaimed_at"), "meta_yaml": str(meta_yaml), } except Exception: reclaim_badge = None if data is None: resp = { "path": path, "sidecar": None, "note": "No sidecar found. File is untracked.", } if reclaim_badge: resp["reclaim_badge"] = reclaim_badge return resp # Sidecar may itself carry a reclaim flag; promote it if present and we # didn't already find a YAML meta. if reclaim_badge is None: sc_reclaim = data.get("reclaim") or {} if sc_reclaim.get("synthetic"): reclaim_badge = { "synthetic": True, "confidence": sc_reclaim.get("confidence"), "inference_sources": sc_reclaim.get("inference_sources", []), "unknown_fields": sc_reclaim.get("unknown_fields", []), "reclaimed_at": sc_reclaim.get("reclaimed_at"), } resp = { "path": path, "sidecar": data, "status": data.get("status"), "source": data.get("source"), "provenance": data.get("provenance", {}), "lineage": data.get("lineage", {}), "notes": data.get("notes", ""), "tags": data.get("tags", []), } if reclaim_badge: resp["reclaim_badge"] = reclaim_badge return resp # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # PROPOSAL TOOL (propose_action) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # The MCP server runs in a separate process tree (Claude Code spawns it # via stdio), so it can't reach the in-memory ChatSessionsStore directly. # To resolve the active project we lean on the same heuristic the Phase 3 # capture path uses: scan ~/.claude/projects// for the latest-mtime # JSONL and look that session up in chat-sessions.json. # # If unresolvable we fall back to a "default" project bucket. The spec's # propose_action input schema is pinned (verbatim from §5) and may NOT # be modified — but the route accepts an optional ``project`` field on # top of the schema, so callers that already know the project can avoid # the heuristic. _PROPOSALS_BRIDGE_URL = "http://127.0.0.1:8431/api/chat/proposals" _PROPOSALS_BRIDGE_TIMEOUT_S = 4.0 _CLAUDE_PROJECTS_ROOT = Path.home() / ".claude" / "projects" def _infer_active_session_id() -> Optional[str]: """Best-effort: find the latest-mtime *.jsonl across ~/.claude/projects/*. The MCP server's launch args include the session log path; reading that deterministically is the right answer but Claude Code doesn't surface it as an env var or command-line flag the server can introspect today. Until that lands we fall back to "newest jsonl wins" — same heuristic Phase 3's capture path uses, so behaviour is consistent. """ if not _CLAUDE_PROJECTS_ROOT.is_dir(): return None latest_path: Optional[Path] = None latest_mtime = 0.0 for proj_dir in _CLAUDE_PROJECTS_ROOT.iterdir(): if not proj_dir.is_dir(): continue for entry in proj_dir.glob("*.jsonl"): try: m = entry.stat().st_mtime except OSError: continue if m > latest_mtime: latest_mtime = m latest_path = entry if latest_path is None: return None return latest_path.stem @_register_tool( name="propose_action", description=( "Propose a typed action for JT to approve/reject in the chat ProposalTray. " "The proposal is persisted to ~/.recoil/proposals//.json and " "emitted on the BUS under scope='chat/proposals'. JT approves (dispatches) " "or rejects (drops) via the tray. Use this for any action that costs money " "or generation time — re-rolls, ref swaps, prompt rewrites, parameter changes." ), input_schema={ "type": "object", "required": ["target", "diff", "est_cost_usd", "est_time", "title"], "properties": { "target": { "type": "string", "description": "Dispatch target — `shot:` | `take:` | `pass:`", }, "title": {"type": "string"}, "est_cost_usd": {"type": "number"}, "est_time": {"type": "string", "description": "Free-form, e.g. '2m 30s'"}, "diff": { "type": "array", "items": { "type": "object", "required": ["kind"], "properties": { "kind": {"type": "string"}, "before": {"type": "string"}, "after": {"type": "string"}, "text": {"type": "string"}, "key": {"type": "string"}, }, }, }, }, }, ) def tool_propose_action(params: dict) -> dict: # Resolve project: explicit caller-supplied param > active session lookup # > fallback to "default". The schema is pinned, so callers that supply # ``project`` are off-spec — we honour it anyway (the route accepts it) # because the bridge's behaviour is "use what you have". project = params.get("project") session_id = _infer_active_session_id() body: dict[str, Any] = { "target": params["target"], "title": params["title"], "est_cost_usd": params["est_cost_usd"], "est_time": params["est_time"], "diff": params.get("diff", []), } if project: body["project"] = project if session_id: body["session_id"] = session_id data = json.dumps(body).encode("utf-8") req = urllib.request.Request( _PROPOSALS_BRIDGE_URL, data=data, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=_PROPOSALS_BRIDGE_TIMEOUT_S) as resp: response_body = resp.read().decode("utf-8") if resp.status < 200 or resp.status >= 300: return { "error": f"proposals API HTTP {resp.status}", "body": response_body, } try: parsed = json.loads(response_body) except json.JSONDecodeError: return { "error": "proposals API returned non-JSON", "body": response_body, } except urllib.error.HTTPError as exc: return {"error": f"proposals API HTTP {exc.code}", "detail": exc.reason} except urllib.error.URLError as exc: return {"error": "proposals API unreachable", "detail": str(exc.reason)} return { "ok": parsed.get("ok", False), "id": parsed.get("id"), "target": params["target"], "title": params["title"], "note": ( "Proposal queued in the chat ProposalTray. JT approves to dispatch " "or rejects to drop. Status updates emit BUS events on scope=chat/proposals." ), } # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # RECENT CLICKS TOOL (get_recent_clicks) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ _CLICKS_BRIDGE_URL = "http://127.0.0.1:8431/api/clicks" _CLICKS_BRIDGE_TIMEOUT_S = 4.0 @_register_tool( name="get_recent_clicks", description=( "Return the most recent tree-leaf clicks (shot/take/pass/file) for the " "active project, newest-first. Reads from the click-history ring backing " "~/.recoil/click-history.jsonl. Project is inferred from the newest " "Claude session jsonl under ~/.claude/projects/." ), input_schema={ "type": "object", "properties": { "n": { "type": "integer", "description": "Max clicks to return (1–50). Defaults to 10.", "minimum": 1, "maximum": 50, }, }, }, ) def tool_get_recent_clicks(params: dict) -> dict: n = int(params.get("n", 10) or 10) if n < 1: n = 1 if n > 50: n = 50 # Project inference mirrors propose_action's heuristic so both tools # resolve to the same active project even though the MCP server runs # in a separate process tree from FastAPI. session_id = _infer_active_session_id() project = ws_state.get_project() if not project and session_id: try: from recoil.api.chat_sessions import ChatSessionsStore project = ChatSessionsStore().lookup_project_by_session_id(session_id) except Exception: project = None if not project: project = "default" url = f"{_CLICKS_BRIDGE_URL}?project_id={urllib.parse.quote(str(project))}&n={n}" req = urllib.request.Request(url, method="GET") try: with urllib.request.urlopen(req, timeout=_CLICKS_BRIDGE_TIMEOUT_S) as resp: response_body = resp.read().decode("utf-8") if resp.status < 200 or resp.status >= 300: return { "error": f"clicks API HTTP {resp.status}", "body": response_body, } try: parsed = json.loads(response_body) except json.JSONDecodeError: return {"error": "clicks API returned non-JSON", "body": response_body} except urllib.error.HTTPError as exc: return {"error": f"clicks API HTTP {exc.code}", "detail": exc.reason} except urllib.error.URLError as exc: return {"error": "clicks API unreachable", "detail": str(exc.reason)} if not isinstance(parsed, list): return {"error": "clicks API returned non-list", "body": parsed} # Project the wire shape onto the spec's output schema: # [{id, kind, ts, project}] out: list[dict[str, Any]] = [] for rec in parsed: if not isinstance(rec, dict): continue out.append( { "id": rec.get("id"), "kind": rec.get("kind"), "ts": rec.get("ts"), "project": rec.get("project_id"), } ) return {"project": project, "clicks": out, "count": len(out)} # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # JSON-RPC 2.0 PROTOCOL HANDLER # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def _build_tools_list() -> list[dict]: """Build the MCP tools/list response.""" return [ { "name": t["name"], "description": t["description"], "inputSchema": t["inputSchema"], } for t in _TOOLS.values() ] def _handle_request(request: dict) -> dict: """Handle a single JSON-RPC request and return the response.""" method = request.get("method", "") req_id = request.get("id") params = request.get("params", {}) if method == "initialize": return { "jsonrpc": "2.0", "id": req_id, "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": { "name": "recoil-workspace", "version": "0.1.0", }, }, } if method == "notifications/initialized": # Notification — no response needed return None if method == "tools/list": return { "jsonrpc": "2.0", "id": req_id, "result": {"tools": _build_tools_list()}, } if method == "tools/call": tool_name = params.get("name", "") tool_args = params.get("arguments", {}) tool_def = _TOOLS.get(tool_name) if tool_def is None: return { "jsonrpc": "2.0", "id": req_id, "result": { "content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}], "isError": True, }, } try: result = tool_def["handler"](tool_args) text = json.dumps(result, indent=2, default=str) return { "jsonrpc": "2.0", "id": req_id, "result": { "content": [{"type": "text", "text": text}], "isError": False, }, } except Exception as e: log.exception("Tool %s failed", tool_name) return { "jsonrpc": "2.0", "id": req_id, "result": { "content": [{"type": "text", "text": f"Error: {e}"}], "isError": True, }, } # Unknown method return { "jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Method not found: {method}"}, } def _main_loop(): """Read JSON-RPC messages from stdin, write responses to stdout.""" log.info("Recoil Workspace MCP server starting...") # Health check: verify projects root is accessible if not projects_root().is_dir(): log.error("projects_root() not found: %s", projects_root()) sys.exit(1) for line in sys.stdin: line = line.strip() if not line: continue try: request = json.loads(line) except json.JSONDecodeError as e: log.warning("Invalid JSON: %s", e) response = { "jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": f"Parse error: {e}"}, } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() continue response = _handle_request(request) if response is not None: sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() if __name__ == "__main__": _main_loop()