WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/client/api_client_etch.py:466-817 (handler)

  The core handler function that implements the workflowy_etch tool logic: parses nodes from JSON or file, validates/escapes content, handles replace_all or additive modes, skips existing nodes by name, recursively creates hierarchical node trees via API calls with retries and rate limiting.

  async def workflowy_etch(
      self,
      parent_id: str,
      nodes: list[dict[str, Any]] | str,
      replace_all: bool = False,
      nodes_file: str | None = None,
  ) -> dict[str, Any]:
      """Create multiple nodes from JSON structure (ETCH command).
  
      Args:
          parent_id: Target Workflowy parent id.
          nodes: Either a parsed list[dict] structure or a JSON string
              representing that list. Ignored when nodes_file is provided.
          replace_all: If True, delete all existing children under
              parent_id first; otherwise additive.
          nodes_file: Optional path to a file containing ETCH payload.
              - If the file contains a JSON object with a top-level
                "nodes" key, that value is used.
              - Else, the entire file content is treated as a JSON
                string to be parsed the same way as the "nodes" string
                parameter (including autofix strategies).
      """
      import asyncio
  
      logger = _ClientLogger()
  
      # Optional file-based payload: if nodes_file is provided, it wins
      # over the inline "nodes" argument.
      if nodes_file:
          try:
              with open(nodes_file, "r", encoding="utf-8") as f:
                  file_text = f.read()
              # Try JSON object with top-level "nodes" first
              try:
                  maybe_json = json.loads(file_text)
                  if isinstance(maybe_json, dict) and "nodes" in maybe_json:
                      nodes = maybe_json["nodes"]
                  else:
                      # Not a dict-with-nodes; treat entire file_text as
                      # the stringified nodes payload
                      nodes = file_text
                  logger.info(
                      f"📄 workflowy_etch: loaded nodes from file '{nodes_file}'"
                  )
              except json.JSONDecodeError:
                  # Not valid JSON as a whole; assume raw JSON string
                  # payload for nodes and let the normal string handler
                  # below deal with it (including autofix).
                  nodes = file_text
                  logger.info(
                      f"📄 workflowy_etch: using raw file text from '{nodes_file}'"
                  )
          except Exception as e:
              error_msg = (
                  f"Failed to read nodes_file '{nodes_file}': {e}"
              )
              self._log_to_file(error_msg, "etch")
              return {
                  "success": False,
                  "nodes_created": 0,
                  "root_node_ids": [],
                  "errors": [error_msg],
              }
      
      # Auto-fix stringified JSON
      stringify_strategy_used = None
      if isinstance(nodes, str):
          logger.warning("⚠️ Received stringified JSON - attempting parse strategies")
  
          # Strategy 1: direct json.loads() on the full string
          try:
              parsed = json.loads(nodes)
              nodes = parsed
              stringify_strategy_used = "Strategy 1: Direct json.loads()"
              logger.info(f"✅ {stringify_strategy_used}")
          except json.JSONDecodeError as e:
              # Dan debug note:
              # In practice we've seen cases where the string is a valid JSON
              # array followed by a few stray characters (e.g. the outer '}'
              # from the request body). json.loads() raises
              # JSONDecodeError('Extra data', ...).
              #
              # Rather than naively stripping N characters, we do a
              # _structure-aware_ trim:
              #  - Find the last closing bracket ']' in the string.
              #  - If it exists, take everything up to and including that
              #    bracket and try json.loads() again.
              #  - This preserves the full array and only discards trailing
              #    junk after a syntactically complete list literal.
              #
              # We only apply this once; if it still fails, we surface a
              # clear error and log to etch_debug.log.
              last_bracket = nodes.rfind("]")
              if last_bracket != -1:
                  candidate = nodes[: last_bracket + 1]
                  try:
                      parsed2 = json.loads(candidate)
                      nodes = parsed2
                      stringify_strategy_used = "Strategy 2: Trim after last ']' (recover from trailing junk)"
                      self._log_to_file(
                          "STRINGIFY AUTOFIX: Strategy 2 used. "
                          f"Original len={len(nodes)}, trimmed_len={len(candidate)}. "
                          f"Original JSONDecodeError={repr(e)}",
                          "etch",
                      )
                      logger.info(f"✅ {stringify_strategy_used}")
                  except json.JSONDecodeError as e2:
                      # Still not valid JSON even after trimming at the
                      # last ']'. At this point we give up and return a
                      # structured error so the caller can see it in the
                      # MCP tool response.
                      error_msg = (
                          "Failed to parse stringified JSON after Strategy 1 "
                          "and Strategy 2. "
                          f"Strategy 1 error={repr(e)}, Strategy 2 error={repr(e2)}"
                      )
                      self._log_to_file(error_msg, "etch")
                      return {
                          "success": False,
                          "nodes_created": 0,
                          "root_node_ids": [],
                          "errors": [error_msg],
                      }
              else:
                  # No closing bracket at all; there's nothing safe to trim
                  # against. Log and fail clearly.
                  error_msg = (
                      "Failed to parse stringified JSON: no closing ']' "
                      f"found. Original error={repr(e)}"
                  )
                  self._log_to_file(error_msg, "etch")
                  return {
                      "success": False,
                      "nodes_created": 0,
                      "root_node_ids": [],
                      "errors": [error_msg],
                  }
      
      # Validate nodes is a list
      if not isinstance(nodes, list):
          return {
              "success": False,
              "nodes_created": 0,
              "root_node_ids": [],
              "errors": ["Parameter 'nodes' must be a list"]
          }
      
      # Validation checkpoint - NOTE fields only
      def validate_and_escape_nodes_recursive(
          nodes_list: list[dict[str, Any]], path: str = "root"
      ) -> tuple[bool, str | None, list[str]]:
          """Recursively validate and auto-escape NAME and NOTE fields."""
          warnings = []
          
          for idx, node in enumerate(nodes_list):
              node_path = f"{path}[{idx}].{node.get('name', 'unnamed')}"
              
              # Validate NAME
              name = node.get('name')
              if not isinstance(name, str) or not name.strip():
                  return (
                      False,
                      f"Node: {node_path}\n\nName must be non-empty string.",
                      warnings,
                  )
              
              processed_name, name_warning = self._validate_name_field(name)
              if processed_name is not None:
                  node['name'] = processed_name
              if name_warning:
                  warnings.append(f"{node_path} - Name escaped")
              
              # Validate NOTE
              note = node.get('note')
              if note:
                  processed_note, note_warning = self._validate_note_field(note, skip_newline_check=False)
                  
                  if processed_note is None and note_warning:
                      return (False, f"Node: {node_path}\n\n{note_warning}", warnings)
                  
                  node['note'] = processed_note
                  
                  if note_warning and "AUTO-ESCAPED" in note_warning:
                      warnings.append(f"{node_path} - Note escaped")
              
              # Recurse
              children = node.get('children', [])
              if children:
                  success, error_msg, child_warnings = validate_and_escape_nodes_recursive(children, node_path)
                  if not success:
                      return (False, error_msg, warnings)
                  warnings.extend(child_warnings)
          
          return (True, None, warnings)
      
      success, error_msg, warnings = validate_and_escape_nodes_recursive(nodes)
      
      if not success:
          return {
              "success": False,
              "nodes_created": 0,
              "root_node_ids": [],
              "errors": [error_msg or "Validation failed"]
          }
      
      if warnings:
          logger.info(f"✅ Auto-escaped {len(warnings)} node(s)")
      
      # Stats tracking
      stats = {
          "api_calls": 0,
          "retries": 0,
          "rate_limit_hits": 0,
          "nodes_created": 0,
          "skipped": 0,
          "errors": []
      }
      
      # REPLACE_ALL or DEFAULT mode
      if replace_all:
          logger.info("🗑️ replace_all=True - Deleting all existing children")
          try:
              request = NodeListRequest(parentId=parent_id)
              existing_children, _ = await self.list_nodes(request)
              stats["api_calls"] += 1
              
              for child in existing_children:
                  try:
                      await self.delete_node(child.id)
                      logger.info(f"  Deleted: {child.nm}")
                      stats["api_calls"] += 1
                  except Exception as e:
                      logger.warning(f"  Failed to delete {child.nm}: {e}")
          except Exception as e:
              logger.warning(f"Could not list/delete existing: {e}")
          
          nodes_to_create = nodes
          existing_names = set()
      else:
          # Additive mode - skip existing by name
          try:
              request = NodeListRequest(parentId=parent_id)
              existing_children, _ = await self.list_nodes(request)
              stats["api_calls"] += 1
              
              existing_names = {child.nm.strip() for child in existing_children if child.nm}
              nodes_to_create = [
                  node for node in nodes 
                  if node.get('name', '').strip() not in existing_names
              ]
              
              stats["skipped"] = len(nodes) - len(nodes_to_create)
              
              if stats["skipped"] > 0:
                  logger.info(f"📝 Skipped {stats['skipped']} existing node(s)")
              
              if not nodes_to_create:
                  return {
                      "success": True,
                      "nodes_created": 0,
                      "root_node_ids": [],
                      "skipped": stats["skipped"],
                      "api_calls": stats["api_calls"],
                      "message": "All nodes already exist - nothing to create"
                  }
          except Exception as e:
              logger.warning(f"Could not check existing: {e}")
              nodes_to_create = nodes
              existing_names = set()
      
      # Create tree recursively
      async def create_tree(parent_id: str, nodes: list[dict[str, Any]]) -> list[str]:
          """Recursively create node tree."""
          created_ids = []
          
          for node_data in nodes:
              try:
                  node_name = node_data['name']
                  
                  if not replace_all and node_name in existing_names:
                      stats["skipped"] += 1
                      continue
                  
                  request = NodeCreateRequest(
                      name=node_name,
                      parent_id=parent_id,
                      note=node_data.get('note'),
                      layoutMode=node_data.get('layout_mode'),
                      position=node_data.get('position', 'bottom')
                  )
                  
                  # Create with retry (internal call)
                  node = await self.create_node(request, _internal_call=True)
                  
                  if node:
                      created_ids.append(node.id)
                      stats["nodes_created"] += 1
                      self._log_to_file(f"  Created: {node_name} ({node.id})", "etch")
                      
                      # Recursively create children
                      if 'children' in node_data and node_data['children']:
                          await create_tree(node.id, node_data['children'])
              
              except Exception as e:
                  error_msg = f"Failed to create '{node_data.get('name', 'unknown')}': {str(e)}"
                  logger.error(error_msg)
                  stats["errors"].append(error_msg)
                  continue
          
          return created_ids
      
      try:
          self._log_to_file(f"ETCH start (replace_all={replace_all}) parent={parent_id}", "etch")
          root_ids = await create_tree(parent_id, nodes_to_create)
          
          self._log_to_file(f"ETCH complete: {stats['nodes_created']} created", "etch")
          
          result = {
              "success": len(stats["errors"]) == 0,
              "nodes_created": stats["nodes_created"],
              "root_node_ids": root_ids,
              "api_calls": stats["api_calls"],
              "retries": stats["retries"],
              "errors": stats["errors"]
          }
          
          if not replace_all:
              result["skipped"] = stats["skipped"]
          
          if stringify_strategy_used:
              result["_stringify_autofix"] = stringify_strategy_used
  
          # Mark parent dirty
          if result.get("success", False):
              try:
                  self._mark_nodes_export_dirty([parent_id])
              except Exception:
                  pass
          
          return result
          
      except Exception as e:
          error_msg = f"ETCH failed: {str(e)}"
          log_event(error_msg, "ETCH")
          stats["errors"].append(error_msg)
          
          return {
              "success": False,
              "nodes_created": stats["nodes_created"],
              "errors": stats["errors"]
          }

• src/workflowy_mcp/server.py:2097-2201 (registration)

  FastMCP tool registration decorator and wrapper function that exposes the workflowy_etch tool to MCP clients, handling parameter resolution (nodes as list/string/file), forwarding to client.workflowy_etch, and returning results.

  @mcp.tool(
      name="workflowy_etch",
      description="Create multiple nodes from JSON structure (no file intermediary). ETCH command for direct node creation."
  )
  async def etch(
      parent_id: str,
      nodes: list[dict] | str | None = None,
      replace_all: bool = False,
      nodes_file: str | None = None,
  ) -> dict:
      """Create multiple nodes from JSON structure.
  
      ETCH command - simple additive node creation (no UUIDs, no updates/moves).
      Fallback: If this fails, use INSCRIBE scroll (write_file → bulk_import).
  
      DEFAULT: Additive (skip existing by name, add new children only)
      REPLACE: Wipe all children, create fresh
  
      For complex operations (moves/updates with UUIDs): Use NEXUS scroll instead.
  
      Args:
          parent_id: Parent UUID where nodes should be created
          nodes: List of node objects (NO UUIDs - just name/note/children)
          replace_all: If True, delete ALL existing children first. Default False.
  
      Returns:
          Dictionary with success status, nodes created, skipped (if append_only), API call stats, and errors
      """
      client = get_client()
  
      # Resolve nodes source: direct value or file-based
      payload_nodes: list[dict] | str | None = nodes
      used_nodes_file = False
  
      if nodes_file:
          try:
              with open(nodes_file, "r", encoding="utf-8") as f:
                  payload_nodes = f.read()
              used_nodes_file = True
          except Exception as e:
              return {
                  "success": False,
                  "nodes_created": 0,
                  "root_node_ids": [],
                  "api_calls": 0,
                  "retries": 0,
                  "rate_limit_hits": 0,
                  "errors": [
                      f"Failed to read nodes_file '{nodes_file}': {str(e)}",
                      "Hint: Ensure the path is correct and accessible from the MCP server",
                  ],
              }
      elif isinstance(payload_nodes, str):
          # Agent convenience: if 'nodes' looks like a real JSON file path, try it as such.
          candidate = payload_nodes.strip()
          if os.path.exists(candidate) and candidate.lower().endswith(".json"):
              try:
                  with open(candidate, "r", encoding="utf-8") as f:
                      payload_nodes = f.read()
                  used_nodes_file = True
                  nodes_file = candidate
                  log_event(
                      f"workflowy_etch: treating 'nodes' string as nodes_file path -> {candidate}",
                      "ETCH",
                  )
              except Exception as e:
                  # Fall back to treating it as JSON; let the ETCH parser report a clear error
                  log_event(
                      f"workflowy_etch: failed to read candidate nodes file '{candidate}': {e}",
                      "ETCH",
                  )
  
      if payload_nodes is None:
          return {
              "success": False,
              "nodes_created": 0,
              "root_node_ids": [],
              "api_calls": 0,
              "retries": 0,
              "rate_limit_hits": 0,
              "errors": [
                  "Missing ETCH payload: provide either 'nodes' (list or stringified JSON) or 'nodes_file' (path to JSON file)",
              ],
          }
  
      # Rate limiter handled within workflowy_etch method due to recursive operations
  
      try:
          result = await client.workflowy_etch(parent_id, payload_nodes, replace_all=replace_all)
          # Annotate result to show where nodes came from (helps debugging real-world agent usage)
          if used_nodes_file:
              result.setdefault("_source", {})["nodes_file"] = nodes_file
          return result
      except Exception as e:
          # Top-level exception capture
          return {
              "success": False,
              "nodes_created": 0,
              "root_node_ids": [],
              "api_calls": 0,
              "retries": 0,
              "rate_limit_hits": 0,
              "errors": [f"An unexpected error occurred: {str(e)}"]
          }