WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:670-834 (handler)

  Handler and registration for the 'mcp_job_status' tool. Retrieves status of in-memory asyncio jobs and detached WEAVE workers by scanning PIDs and persistent journals in nexus_runs directory. Supports querying all jobs or a specific job_id.

  @mcp.tool(
      name="mcp_job_status",
      description="Get status/result for long-running MCP jobs (ETCH, NEXUS, etc.).",
  )
  async def mcp_job_status(job_id: str | None = None) -> dict:
      """Get status for background jobs (in-memory + detached WEAVE workers).
  
      Scans:
      - In-memory asyncio jobs (_jobs registry)
      - Detached WEAVE workers via:
        • Active PIDs (.weave.pid under nexus_runs/)
        • Persistent journals (enchanted_terrain.weave_journal.json / *.weave_journal.json)
  
      This allows status/error inspection even after the worker PID has exited.
      """
      from .client.api_client import scan_active_weaves
      from pathlib import Path
      import json as json_module
  
      # Determine nexus_runs directory
      # TODO: make this configurable or derive from client
      nexus_runs_base = r"E:\__daniel347x\__Obsidian\__Inking into Mind\--TypingMind\Projects - All\Projects - Individual\TODO\temp\nexus_runs"
  
      def _scan_weave_journals(base_dir_str: str) -> list[dict[str, Any]]:
          """Scan nexus_runs/ for WEAVE journal files and summarize their status.
  
          Returns one entry per journal with:
          - job_id: weave-enchanted-<nexus_tag> or weave-direct-<stem>
          - mode:  'enchanted' | 'direct'
          - status: 'completed' | 'failed' | 'unknown'
          - detached: True
          - journal: path to *.weave_journal.json
          - log_file: associated .weave.log if present
          - last_run_* and phase fields from journal
          """
          base_dir = Path(base_dir_str)
          results: list[dict[str, Any]] = []
          if not base_dir.exists():
              return results
  
          for run_dir in base_dir.iterdir():
              if not run_dir.is_dir():
                  continue
              name = run_dir.name
              # Extract nexus_tag from directory name: either <tag> or <TIMESTAMP>__<tag>
              if "__" in name:
                  nexus_tag = name.split("__", 1)[1]
              else:
                  nexus_tag = name
  
              # Look for any *.weave_journal.json in this run directory
              for journal_path in run_dir.glob("*.weave_journal.json"):
                  try:
                      with open(journal_path, "r", encoding="utf-8") as jf:
                          journal = json_module.load(jf)
                  except Exception:
                      continue
  
                  json_file = journal.get("json_file")
                  if json_file and str(json_file).endswith("enchanted_terrain.json"):
                      mode = "enchanted"
                      job_id_val = f"weave-enchanted-{nexus_tag}"
                  else:
                      # Direct-mode weave (json_file is some other JSON path)
                      stem = Path(json_file).stem if json_file else name
                      mode = "direct"
                      job_id_val = f"weave-direct-{stem}"
  
                  last_completed = bool(journal.get("last_run_completed"))
                  last_error = journal.get("last_run_error")
                  if last_completed and not last_error:
                      status_val = "completed"
                  elif last_error:
                      status_val = "failed"
                  else:
                      status_val = "unknown"
  
                  log_file = run_dir / ".weave.log"
  
                  results.append(
                      {
                          "job_id": job_id_val,
                          "nexus_tag": nexus_tag,
                          "mode": mode,
                          "status": status_val,
                          "detached": True,
                          "journal": str(journal_path),
                          "log_file": str(log_file) if log_file.exists() else None,
                          "last_run_started_at": journal.get("last_run_started_at"),
                          "last_run_completed": journal.get("last_run_completed"),
                          "last_run_failed_at": journal.get("last_run_failed_at"),
                          "last_run_error": last_error,
                          "phase": journal.get("phase"),
                      }
                  )
  
          return results
  
      # Return status for one job (if job_id given) or all jobs
      if job_id is None:
          # List ALL jobs (in-memory + detached)
          in_memory_jobs: list[dict[str, Any]] = []
          for job in _jobs.values():
              in_memory_jobs.append(
                  {
                      "job_id": job.get("job_id"),
                      "kind": job.get("kind"),
                      "status": job.get("status"),
                      "started_at": job.get("started_at"),
                      "finished_at": job.get("finished_at"),
                      "detached": False,
                  }
              )
  
          # Active detached jobs (PIDs still running)
          active_weaves = {j["job_id"]: j for j in scan_active_weaves(nexus_runs_base)}
          # All WEAVE journals (completed/failed/unknown)
          journal_jobs = {j["job_id"]: j for j in _scan_weave_journals(nexus_runs_base)}
  
          # Merge active + journal info per job_id
          detached_jobs_dict: dict[str, dict[str, Any]] = {}
          for jid, jj in journal_jobs.items():
              merged = dict(jj)
              if jid in active_weaves:
                  # Active info (PID, running status) overrides journal's status,
                  # but we keep journal fields like last_run_error and phase.
                  merged.update(active_weaves[jid])
              detached_jobs_dict[jid] = merged
          # Any active jobs without journals (edge case)
          for jid, aj in active_weaves.items():
              if jid not in detached_jobs_dict:
                  detached_jobs_dict[jid] = aj
  
          detached_jobs = list(detached_jobs_dict.values())
  
          return {
              "success": True,
              "in_memory_jobs": in_memory_jobs,
              "detached_jobs": detached_jobs,
              "total": len(in_memory_jobs) + len(detached_jobs),
          }
  
      # Check in-memory first
      job = _jobs.get(job_id)
      if job:
          # Do not expose internal task handle
          view = {k: v for k, v in job.items() if k not in ("payload", "_task")}
          return {"success": True, **view}
  
      # Check detached WEAVE jobs (active or completed) using journals + PIDs
      active_weaves = {j["job_id"]: j for j in scan_active_weaves(nexus_runs_base)}
      journal_jobs = {j["job_id"]: j for j in _scan_weave_journals(nexus_runs_base)}
  
      if job_id in journal_jobs or job_id in active_weaves:
          merged: dict[str, Any] = {}
          if job_id in journal_jobs:
              merged.update(journal_jobs[job_id])
          if job_id in active_weaves:
              merged.update(active_weaves[job_id])
          merged["success"] = True
          return merged
  
      # Unknown job_id
      return {"success": False, "error": f"Unknown job_id: {job_id}"}