WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1567-1607 (registration)

  MCP tool registration for 'nexus_transform_jewel' with parameters and description defining the schema and usage.
  name="nexus_transform_jewel", description=( "Apply JEWELSTORM semantic operations to a NEXUS working_gem JSON file " "(PHANTOM GEM working copy). This is the semantic analogue of edit_file " "for PHANTOM GEM JSON: MOVE_NODE, DELETE_NODE, RENAME_NODE, SET_NOTE, " "SET_ATTRS, CREATE_NODE, all referencing nodes by jewel_id, plus text-level " "SEARCH_REPLACE / SEARCH_AND_TAG over name/note fields (substring/whole-word, " "optional regex, tagging in name and/or note based on matches)." ), ) def nexus_transform_jewel( jewel_file: str, operations: list[dict[str, Any]], dry_run: bool = False, stop_on_error: bool = True, ) -> dict: """JEWELSTORM transform on a NEXUS working_gem JSON file. This tool wraps nexus_json_tools.transform_jewel via the WorkFlowyClient, providing an MCP-friendly interface for JEWELSTORM operations. Args: jewel_file: Absolute path to working_gem JSON (typically a QUILLSTRIKE working stone derived from phantom_gem.json). operations: List of operation dicts. Each must include an "op" key and operation-specific fields (e.g., jewel_id, parent_jewel_id, position, etc.). dry_run: If True, simulate only (no file write). stop_on_error: If True, abort on first error (no write). Returns: Result dict from transform_jewel with success flag, counts, and errors. """ client = get_client() return client.nexus_transform_jewel( jewel_file=jewel_file, operations=operations, dry_run=dry_run, stop_on_error=stop_on_error, )
• src/workflowy_mcp/nexus_json_tools.py:79-837 (handler)

  Core handler function implementing all JEWELSTORM operations (MOVE_NODE, DELETE_NODE, RENAME_NODE, SET_NOTE, CREATE_NODE, SEARCH_REPLACE, etc.) on working_gem JSON using jewel_id handles.
  def transform_jewel( jewel_file: str, operations: List[Dict[str, Any]], dry_run: bool = False, stop_on_error: bool = True, ) -> Dict[str, Any]: """Apply JEWELSTORM semantic operations to a NEXUS working_gem JSON file. This is the semantic analogue of pattern-based edit_file() for PHANTOM GEM / working_gem JSON: - Operates purely offline (no Workflowy API calls) - Works on a local JSON file produced by JEWELSTRIKE (typically phantom_gem.json copied to a working stone) - Uses jewel_id as the stable identity handle inside JEWELSTORM - Never invents real Workflowy IDs: new nodes are written without an "id" field so NEXUS/WEAVE can treat them as CREATE operations Args: jewel_file: Path to working_gem JSON file operations: List of operation dictionaries, e.g.: { "op": "MOVE_NODE", "jewel_id": "J-001", "new_parent_jewel_id": "J-010", "position": "LAST" | "FIRST" | "BEFORE" | "AFTER", "relative_to_jewel_id": "J-999" # for BEFORE/AFTER } dry_run: If True, simulate only (no file write) stop_on_error: If True, abort on first error (no write) Returns: Dict with success flag, counts, and error details. Example: { "success": True, "applied_count": 3, "dry_run": False, "nodes_created": 1, "nodes_deleted": 0, "nodes_moved": 1, "nodes_renamed": 1, "notes_updated": 0, "attrs_updated": 0, "errors": [] } """ import uuid # ------- Load JSON safely (convert die/SystemExit into structured error) ------- try: data = load_json(jewel_file) except SystemExit as e: # die() inside load_json uses sys.exit return { "success": False, "applied_count": 0, "dry_run": dry_run, "nodes_created": 0, "nodes_deleted": 0, "nodes_moved": 0, "nodes_renamed": 0, "notes_updated": 0, "attrs_updated": 0, "errors": [ { "index": -1, "op": None, "code": "LOAD_ERROR", "message": f"Failed to load JSON from {jewel_file}: {e}", } ], } # Determine editable roots list (supports both export-package dict and bare list) if isinstance(data, dict) and isinstance(data.get("nodes"), list): original_roots = data["nodes"] elif isinstance(data, list): original_roots = data else: return { "success": False, "applied_count": 0, "dry_run": dry_run, "nodes_created": 0, "nodes_deleted": 0, "nodes_moved": 0, "nodes_renamed": 0, "notes_updated": 0, "attrs_updated": 0, "errors": [ { "index": -1, "op": None, "code": "FORMAT_ERROR", "message": "JSON must be a dict with 'nodes' or a bare list of nodes for transform_jewel.", } ], } # Work on a deep copy so we never mutate the original unless we succeed roots: List[JsonDict] = copy.deepcopy(original_roots) # ------- Indexing: jewel_id-based identity ------- by_jewel_id: Dict[str, JsonDict] = {} parent_by_jewel_id: Dict[str, Optional[str]] = {} existing_ids: Set[str] = set() def _ensure_children_list(node: JsonDict) -> List[JsonDict]: children = node.get("children") if not isinstance(children, list): children = [] node["children"] = children return children def _register_node(node: JsonDict, parent_jid: Optional[str]) -> None: """Register node and its subtree in jewel_id index. Identity: - Prefer "jewel_id" if present - Fallback to "id" for older files (pre-JEWELSTRIKE) """ jewel_id = node.get("jewel_id") or node.get("id") if jewel_id: if jewel_id in by_jewel_id: raise ValueError(f"Duplicate jewel_id/id {jewel_id!r} in JSON tree") by_jewel_id[jewel_id] = node parent_by_jewel_id[jewel_id] = parent_jid existing_ids.add(jewel_id) children = node.get("children") or [] if not isinstance(children, list): children = [] node["children"] = children for child in children: if isinstance(child, dict): _register_node(child, jewel_id) for root in roots: if isinstance(root, dict): _register_node(root, None) def _new_jewel_id() -> str: """Generate a fresh, human-friendly jewel_id that cannot collide.""" while True: candidate = "J-" + uuid.uuid4().hex[:8] if candidate not in existing_ids: existing_ids.add(candidate) return candidate def _resolve_to_jewel_id(identifier: str) -> str: """Auto-resolve Workflowy UUID to jewel_id if needed. If identifier looks like a Workflowy UUID (standard UUID format, not J-NNN), search the tree for a node with that id and return its jewel_id. If identifier is already a jewel_id (J-NNN format), return as-is. Args: identifier: Either a jewel_id (J-NNN) or Workflowy UUID Returns: jewel_id to use for the operation Raises: ValueError: If UUID not found in tree """ import re # Check if it's a standard UUID format (8-4-4-4-12 hex pattern) uuid_pattern = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE) if uuid_pattern.match(identifier): # It's a Workflowy UUID - search for node with this id for jewel_id, node in by_jewel_id.items(): if node.get("id") == identifier: return jewel_id # Not found raise ValueError( f"Node with Workflowy UUID '{identifier}' not found in GEM. " "(Auto-resolved from jewel_id parameter - agent can use Workflowy UUIDs directly)" ) else: # Already a jewel_id (J-NNN format) or other format - return as-is return identifier def _resolve_to_jewel_id(identifier: str) -> str: """Auto-resolve Workflowy UUID to jewel_id if needed. If identifier looks like a Workflowy UUID (standard UUID format, not J-NNN), search the tree for a node with that id and return its jewel_id. If identifier is already a jewel_id (J-NNN format), return as-is. Args: identifier: Either a jewel_id (J-NNN) or Workflowy UUID Returns: jewel_id to use for the operation Raises: ValueError: If UUID not found in tree """ import re # Check if it's a standard UUID format (8-4-4-4-12 hex pattern) uuid_pattern = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE) if uuid_pattern.match(identifier): # It's a Workflowy UUID - search for node with this id for jewel_id, node in by_jewel_id.items(): if node.get("id") == identifier: return jewel_id # Not found raise ValueError( f"Node with Workflowy UUID '{identifier}' not found in GEM. " "(Auto-resolved from jewel_id parameter - agent can use Workflowy UUIDs directly)" ) else: # Already a jewel_id (J-NNN format) or other format - return as-is return identifier def _get_node_and_parent_list(jewel_id: str) -> Tuple[JsonDict, Optional[str], List[JsonDict]]: # Auto-resolve Workflowy UUID to jewel_id if needed jewel_id = _resolve_to_jewel_id(jewel_id) node = by_jewel_id.get(jewel_id) if node is None: raise ValueError(f"Node with jewel_id/id {jewel_id!r} not found") parent_jid = parent_by_jewel_id.get(jewel_id) if parent_jid is None: siblings = roots else: parent_node = by_jewel_id.get(parent_jid) if parent_node is None: raise ValueError(f"Parent with jewel_id/id {parent_jid!r} not found for node {jewel_id!r}") siblings = _ensure_children_list(parent_node) return node, parent_jid, siblings def _assert_no_cycle(node_jewel_id: str, new_parent_jewel_id: Optional[str]) -> None: cur = new_parent_jewel_id while cur is not None: if cur == node_jewel_id: raise ValueError( f"Cannot move node {node_jewel_id!r} under its own descendant {new_parent_jewel_id!r}" ) cur = parent_by_jewel_id.get(cur) def _remove_subtree_from_index(node: JsonDict) -> None: jid = node.get("jewel_id") or node.get("id") if jid: by_jewel_id.pop(jid, None) parent_by_jewel_id.pop(jid, None) existing_ids.discard(jid) for child in node.get("children") or []: if isinstance(child, dict): _remove_subtree_from_index(child) def _build_subtree_from_spec(spec: Dict[str, Any]) -> JsonDict: """Build a new subtree from a CREATE_NODE spec (Pattern A). Spec keys used: - name (required) - note (optional) - attrs or data (optional, mapped to node['data']) - jewel_id (optional; if omitted, auto-generated) - children (optional list of nested specs) - parent_id (optional, ignored - will be set during insertion) """ if "name" not in spec: raise ValueError("CREATE_NODE spec requires 'name' field in node object") node: JsonDict = {"name": spec["name"]} if "note" in spec: node["note"] = spec["note"] attrs = spec.get("attrs") or spec.get("data") if isinstance(attrs, dict): node["data"] = copy.deepcopy(attrs) # Assign jewel_id (never assign real Workflowy id here) jewel_id = spec.get("jewel_id") if jewel_id is None: jewel_id = _new_jewel_id() else: if jewel_id in existing_ids: raise ValueError(f"CREATE_NODE requested duplicate jewel_id {jewel_id!r}") existing_ids.add(jewel_id) node["jewel_id"] = jewel_id # Recursively build children children_specs = spec.get("children") or [] children_nodes: List[JsonDict] = [] for child_spec in children_specs: if isinstance(child_spec, dict): child_node = _build_subtree_from_spec(child_spec) children_nodes.append(child_node) node["children"] = children_nodes return node def _classify_children_for_destructive_ops(node: JsonDict) -> Dict[str, Any]: """Classify immediate children for destructive JEWELSTORM ops. Returns a dict with: children: list of child dict nodes children_status: str truncated: bool has_loaded_ether_children: bool # child has a real Workflowy id has_new_children: bool # child has no id yet (JEWEL-only) category: "EMPTY" | "JEWEL_ONLY" | "ETHER_ONLY" | "MIXED" """ children_raw = node.get("children") or [] children = [c for c in children_raw if isinstance(c, dict)] children_status = node.get("children_status") or "complete" truncated = children_status in {"truncated_by_depth", "truncated_by_count"} has_loaded_ether_children = any(c.get("id") is not None for c in children) has_new_children = any(c.get("id") is None for c in children) if not children: category = "EMPTY" elif (not has_loaded_ether_children) and has_new_children and not truncated: category = "JEWEL_ONLY" elif has_loaded_ether_children and not has_new_children: category = "ETHER_ONLY" else: category = "MIXED" return { "children": children, "children_status": children_status, "truncated": truncated, "has_loaded_ether_children": has_loaded_ether_children, "has_new_children": has_new_children, "category": category, } # ------- Apply operations ------- applied_count = 0 nodes_created = 0 nodes_deleted = 0 nodes_moved = 0 nodes_renamed = 0 notes_updated = 0 attrs_updated = 0 errors: List[Dict[str, Any]] = [] for idx, op in enumerate(operations or []): op_type_raw = op.get("op") or op.get("operation") if not op_type_raw: err = { "index": idx, "op": op, "code": "MISSING_OP", "message": "Operation missing 'op' field", } errors.append(err) if stop_on_error: break continue op_type = str(op_type_raw).upper() try: # MOVE_NODE if op_type == "MOVE_NODE": src_jid = op.get("jewel_id") if not src_jid: raise ValueError("MOVE_NODE requires 'jewel_id'") position = str(op.get("position", "LAST")).upper() # Normalize synonyms if position == "BOTTOM": position = "LAST" if position == "TOP": position = "FIRST" rel_jid = op.get("relative_to_jewel_id") if rel_jid: rel_jid = _resolve_to_jewel_id(rel_jid) new_parent_jid = op.get("new_parent_jewel_id") if position in {"BEFORE", "AFTER"}: if not rel_jid: raise ValueError( "MOVE_NODE with position BEFORE/AFTER requires 'relative_to_jewel_id'" ) # Target parent/siblings come from relative node _, rel_parent_jid, rel_siblings = _get_node_and_parent_list(rel_jid) target_parent_jid = rel_parent_jid target_list = rel_siblings else: # FIRST/LAST under explicit parent (or root when parent None) target_parent_jid = new_parent_jid if new_parent_jid is None: target_list = roots else: parent_node = by_jewel_id.get(new_parent_jid) if parent_node is None: raise ValueError(f"New parent jewel_id/id {new_parent_jid!r} not found") target_list = _ensure_children_list(parent_node) # Cycle check _assert_no_cycle(src_jid, target_parent_jid) node, old_parent_jid, old_siblings = _get_node_and_parent_list(src_jid) # Remove from old siblings if node in old_siblings: old_siblings.remove(node) # Insert into new location if position == "FIRST": target_list.insert(0, node) elif position == "LAST": target_list.append(node) elif position in {"BEFORE", "AFTER"}: rel_node, _, _ = _get_node_and_parent_list(rel_jid) # type: ignore[arg-type] try: rel_index = target_list.index(rel_node) except ValueError as e: raise ValueError( f"Relative node {rel_jid!r} not found under chosen parent for MOVE_NODE" ) from e insert_index = rel_index if position == "BEFORE" else rel_index + 1 target_list.insert(insert_index, node) else: raise ValueError(f"Unsupported MOVE_NODE position {position!r}") # Update parent mapping parent_by_jewel_id[src_jid] = target_parent_jid nodes_moved += 1 # DELETE_NODE elif op_type == "DELETE_NODE": jid = op.get("jewel_id") if not jid: raise ValueError("DELETE_NODE requires 'jewel_id'") delete_from_ether = bool(op.get("delete_from_ether")) mode_raw = op.get("mode") mode = str(mode_raw).upper() if mode_raw is not None else "SMART" node, parent_jid, siblings = _get_node_and_parent_list(jid) info = _classify_children_for_destructive_ops(node) children = info["children"] category = info["category"] # Legacy strict mode: preserve original FAIL_IF_HAS_CHILDREN semantics if mode == "FAIL_IF_HAS_CHILDREN": if children: raise ValueError( f"DELETE_NODE {jid!r} refused: node has children and mode=FAIL_IF_HAS_CHILDREN" ) else: # SMART semantics (default when mode not provided) if category == "MIXED": raise ValueError( f"DELETE_NODE {jid!r} refused: mixed new/ETHER/truncated children not supported; " "delete or move children individually first" ) if category == "ETHER_ONLY" and not delete_from_ether: raise ValueError( f"DELETE_NODE {jid!r} refused: node has ETHER-backed children; " "set delete_from_ether=True to delete subtree in Workflowy" ) # EMPTY and JEWEL_ONLY are always allowed. # Remove from siblings and index if node in siblings: siblings.remove(node) _remove_subtree_from_index(node) nodes_deleted += 1 # DELETE_ALL_CHILDREN elif op_type == "DELETE_ALL_CHILDREN": jid = op.get("jewel_id") if not jid: raise ValueError("DELETE_ALL_CHILDREN requires 'jewel_id'") delete_from_ether = bool(op.get("delete_from_ether")) mode_raw = op.get("mode") mode = str(mode_raw).upper() if mode_raw is not None else "SMART" node = by_jewel_id.get(jid) if node is None: raise ValueError(f"Node with jewel_id/id {jid!r} not found") info = _classify_children_for_destructive_ops(node) children = info["children"] category = info["category"] truncated = info["truncated"] # Nothing to do if not children and not truncated: pass else: # For DELETE_ALL_CHILDREN we do not support truncated children sets in v1 if truncated: raise ValueError( f"DELETE_ALL_CHILDREN {jid!r} refused: children set is truncated; " "re-GLIMPSE with full children or delete the node instead" ) if mode == "FAIL_IF_HAS_CHILDREN": if children: raise ValueError( f"DELETE_ALL_CHILDREN {jid!r} refused: node has children and " "mode=FAIL_IF_HAS_CHILDREN" ) else: if category == "MIXED": raise ValueError( f"DELETE_ALL_CHILDREN {jid!r} refused: mixed new/ETHER children not supported; " "delete or move children individually first" ) if category == "ETHER_ONLY" and not delete_from_ether: raise ValueError( f"DELETE_ALL_CHILDREN {jid!r} refused: children are ETHER-backed; " "set delete_from_ether=True to delete them in Workflowy" ) # EMPTY and JEWEL_ONLY are always allowed here. # Allowed path: remove all current children from index and node for child in list(children): _remove_subtree_from_index(child) node["children"] = [] nodes_deleted += len(children) # RENAME_NODE elif op_type == "RENAME_NODE": jid = op.get("jewel_id") if not jid: raise ValueError("RENAME_NODE requires 'jewel_id'") new_name = op.get("new_name") if new_name is None: raise ValueError("RENAME_NODE requires 'new_name'") node = by_jewel_id.get(jid) if node is None: raise ValueError(f"Node with jewel_id/id {jid!r} not found") node["name"] = new_name nodes_renamed += 1 # SET_NOTE elif op_type == "SET_NOTE": jid = op.get("jewel_id") if not jid: raise ValueError("SET_NOTE requires 'jewel_id'") new_note = op.get("new_note") # Allow empty string / None to clear node = by_jewel_id.get(jid) if node is None: raise ValueError(f"Node with jewel_id/id {jid!r} not found") node["note"] = new_note notes_updated += 1 # SET_ATTRS elif op_type == "SET_ATTRS": jid = op.get("jewel_id") if not jid: raise ValueError("SET_ATTRS requires 'jewel_id'") attrs = op.get("attrs") or {} if not isinstance(attrs, dict): raise ValueError("SET_ATTRS 'attrs' must be a dict") node = by_jewel_id.get(jid) if node is None: raise ValueError(f"Node with jewel_id/id {jid!r} not found") data = node.get("data") if not isinstance(data, dict): data = {} allowed_keys = {"completed", "layoutMode", "priority", "tags"} for key, value in attrs.items(): if key not in allowed_keys: raise ValueError(f"Unsupported attr key {key!r} in SET_ATTRS") if value is None: data.pop(key, None) else: data[key] = value if data: node["data"] = data elif "data" in node: del node["data"] attrs_updated += 1 # CREATE_NODE (Pattern A with optional jewel_id) elif op_type == "CREATE_NODE": parent_jid = op.get("parent_jewel_id") position = str(op.get("position", "LAST")).upper() # Normalize synonyms if position == "BOTTOM": position = "LAST" if position == "TOP": position = "FIRST" rel_jid = op.get("relative_to_jewel_id") if rel_jid: rel_jid = _resolve_to_jewel_id(rel_jid) if parent_jid is None: parent_children = roots target_parent_jid = None else: parent_node = by_jewel_id.get(parent_jid) if parent_node is None: raise ValueError(f"CREATE_NODE parent_jewel_id {parent_jid!r} not found") parent_children = _ensure_children_list(parent_node) target_parent_jid = parent_jid # Build node spec for subtree # Two formats supported: # 1. Compact: node fields at operation level (name, note, children, etc.) # 2. Wrapped: node fields inside "node" key if "node" in op: spec = op["node"] else: # Remove op-specific keys to get node spec spec = {k: v for k, v in op.items() if k not in { "op", "operation", "parent_jewel_id", "position", "relative_to_jewel_id", }} new_node = _build_subtree_from_spec(spec) # Register subtree in indexes with correct parent mapping _register_node(new_node, target_parent_jid) # Insert relative to siblings if position == "FIRST": parent_children.insert(0, new_node) elif position == "LAST": parent_children.append(new_node) elif position in {"BEFORE", "AFTER"}: if not rel_jid: raise ValueError( "CREATE_NODE with position BEFORE/AFTER requires 'relative_to_jewel_id'" ) rel_node, _, rel_siblings = _get_node_and_parent_list(rel_jid) # type: ignore[arg-type] # Force siblings to be the same list as parent_children if rel_siblings is not parent_children: raise ValueError( "CREATE_NODE BEFORE/AFTER relative_to_jewel_id must share the same parent" ) try: rel_index = parent_children.index(rel_node) except ValueError as e: raise ValueError( f"Relative node {rel_jid!r} not found under chosen parent for CREATE_NODE" ) from e insert_index = rel_index if position == "BEFORE" else rel_index + 1 parent_children.insert(insert_index, new_node) else: raise ValueError(f"Unsupported CREATE_NODE position {position!r}") nodes_created += 1 # SET_ATTRS_BY_PATH (path-based attribute update, used for JEWEL UUID injection) elif op_type == "SET_ATTRS_BY_PATH": path = op.get("path") attrs = op.get("attrs") or {} if not isinstance(path, list) or not path: raise ValueError("SET_ATTRS_BY_PATH requires non-empty 'path' list") if not isinstance(attrs, dict): raise ValueError("SET_ATTRS_BY_PATH 'attrs' must be a dict") # Navigate by index path from roots current_list: List[JsonDict] = roots target_node: Optional[JsonDict] = None for level, idx in enumerate(path): if not isinstance(idx, int): raise ValueError( f"SET_ATTRS_BY_PATH path index at position {level} must be int, got {type(idx).__name__}" ) if idx < 0 or idx >= len(current_list): raise ValueError( f"SET_ATTRS_BY_PATH path index {idx} out of range at position {level}" ) target_node = current_list[idx] if level < len(path) - 1: children = target_node.get("children") if not isinstance(children, list): raise ValueError( f"SET_ATTRS_BY_PATH path descends into non-list children at position {level}" ) current_list = children if target_node is None: raise ValueError("SET_ATTRS_BY_PATH could not resolve target node from path") for key, value in attrs.items(): if key == "id": if value is None: target_node.pop("id", None) else: if not isinstance(value, str): raise ValueError( "SET_ATTRS_BY_PATH 'id' value must be a string when not None" ) target_node["id"] = value else: raise ValueError( f"Unsupported attr key {key!r} in SET_ATTRS_BY_PATH (only 'id' is currently allowed)" ) attrs_updated += 1 else: raise ValueError(f"Unknown operation type {op_type!r}") applied_count += 1 except Exception as e: # noqa: BLE001 errors.append( { "index": idx, "op": op, "code": "OP_ERROR", "message": str(e), } ) if stop_on_error: break # ------- Persist changes (if not dry-run and no stop-on-error failure) ------- if not dry_run and (not stop_on_error or not errors): # Attach modified roots back to original structure if isinstance(data, dict) and isinstance(data.get("nodes"), list): data["nodes"] = roots # Recalculate counts for readability in the JEWEL working_gem, # but DO NOT touch children_status here. Truncation semantics belong # to the NEXUS/TERRAIN layer (shimmering/enchanted terrain). wrapper = {"nodes": roots} recalc_all_counts_gem(wrapper) elif isinstance(data, list): data = roots # type: ignore[assignment] wrapper = {"nodes": roots} recalc_all_counts_gem(wrapper) save_json(jewel_file, data) # type: ignore[arg-type] return { "success": len(errors) == 0, "applied_count": applied_count, "dry_run": dry_run, "nodes_created": nodes_created, "nodes_deleted": nodes_deleted, "nodes_moved": nodes_moved, "nodes_renamed": nodes_renamed, "notes_updated": notes_updated, "attrs_updated": attrs_updated, "errors": errors, }