WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1024-1064 (handler)

  MCP tool handler and registration for workflowy_update_node. Wraps client.update_node with rate limiting.

  @mcp.tool(name="workflowy_update_node", description="Update an existing WorkFlowy node")
  async def update_node(
      node_id: str,
      name: str | None = None,
      note: str | None = None,
      layout_mode: Literal["bullets", "todo", "h1", "h2", "h3"] | None = None,
      _completed: bool | None = None,
  ) -> WorkFlowyNode:
      """Update an existing WorkFlowy node.
  
      Args:
          node_id: The ID of the node to update
          name: New text content for the node (optional)
          note: New note/description (optional)
          layout_mode: New layout mode for the node (bullets, todo, h1, h2, h3) (optional)
          _completed: New completion status (not used - use complete_node/uncomplete_node)
  
      Returns:
          The updated WorkFlowy node
      """
      client = get_client()
  
      request = NodeUpdateRequest(  # type: ignore[call-arg]
          name=name,
          note=note,
          layoutMode=layout_mode,
      )
  
      if _rate_limiter:
          await _rate_limiter.acquire()
  
      try:
          node = await client.update_node(node_id, request)
          if _rate_limiter:
              _rate_limiter.on_success()
          return node
      except Exception as e:
          if _rate_limiter and hasattr(e, "__class__") and e.__class__.__name__ == "RateLimitError":
              _rate_limiter.on_rate_limit(getattr(e, "retry_after", None))
          raise

• src/workflowy_mcp/models/requests.py:38-50 (schema)

  Pydantic schema NodeUpdateRequest defining input parameters for node updates.

  class NodeUpdateRequest(BaseModel):
      """Request payload for updating an existing node."""
  
      name: str | None = Field(None, description="New text content")
      note: str | None = Field(None, description="New note content")
      layoutMode: Literal["bullets", "todo", "h1", "h2", "h3"] | None = Field(
          None, description="New display mode (bullets, todo, h1, h2, h3)"
      )
  
      def has_updates(self) -> bool:
          """Check if at least one field is provided for update."""
          return any(getattr(self, field) is not None for field in self.model_fields)

• src/workflowy_mcp/client/api_client_core.py:597-705 (handler)

  Core API client implementation of update_node: HTTP POST, validation, escaping, retries, error handling.

  async def update_node(self, node_id: str, request: NodeUpdateRequest, max_retries: int = 10) -> WorkFlowyNode:
      """Update an existing node with exponential backoff retry.
      
      Args:
          node_id: The ID of the node to update
          request: Node update request
          max_retries: Maximum retry attempts (default 10)
      """
      import asyncio
  
      logger = _ClientLogger()
      
      # Validate and escape name field if being updated
      if request.name is not None:
          processed_name, name_warning = self._validate_name_field(request.name)
          if processed_name is not None:
              request.name = processed_name
          if name_warning:
              logger.info(name_warning)
      
      # Validate and escape note field if being updated
      if request.note is not None:
          # Note: update_node doesn't have _internal_call flag yet, always validates
          processed_note, message = self._validate_note_field(request.note)
          
          if processed_note is None and message:  # Blocking error
              raise NetworkError(message)
          
          # Strip override token if present
          if processed_note and processed_note.startswith("<<<LITERAL_BACKSLASH_N_INTENTIONAL>>>"):
              processed_note = processed_note.replace("<<<LITERAL_BACKSLASH_N_INTENTIONAL>>>", "", 1)
          
          # Use processed (escaped) note
          request.note = processed_note
          
          # Log warning if escaping occurred
          if message and "AUTO-ESCAPED" in message:
              logger.info(message)
      
      retry_count = 0
      base_delay = 1.0
      
      while retry_count < max_retries:
          # Force delay at START of each iteration (rate limit protection)
          await asyncio.sleep(API_RATE_LIMIT_DELAY)
          
          try:
              response = await self.client.post(
                  f"/nodes/{node_id}", json=request.model_dump(exclude_none=True)
              )
              data = await self._handle_response(response)
              # API returns {"status": "ok"} - fetch updated node
              if isinstance(data, dict) and data.get('status') == 'ok':
                  get_response = await self.client.get(f"/nodes/{node_id}")
                  node_data = await self._handle_response(get_response)
                  node = WorkFlowyNode(**node_data["node"])
              else:
                  # Fallback for unexpected format
                  node = WorkFlowyNode(**data)
  
              # Best-effort: mark this node as dirty so that subsequent
              # /nodes-export-based operations touching this subtree can
              # trigger a refresh when needed.
              try:
                  self._mark_nodes_export_dirty([node_id])
              except Exception:
                  # Cache dirty marking must never affect API behavior
                  pass
  
              return node
                  
          except RateLimitError as e:
              retry_count += 1
              retry_after = getattr(e, 'retry_after', None) or (base_delay * (2 ** retry_count))
              logger.warning(
                  f"Rate limited on update_node. Retry after {retry_after}s. "
                  f"Attempt {retry_count}/{max_retries}"
              )
              
              if retry_count < max_retries:
                  await asyncio.sleep(retry_after)
              else:
                  raise
                  
          except NetworkError as e:
              retry_count += 1
              logger.warning(
                  f"Network error on update_node: {e}. Retry {retry_count}/{max_retries}"
              )
              
              if retry_count < max_retries:
                  await asyncio.sleep(base_delay * (2 ** retry_count))
              else:
                  raise
                  
          except httpx.TimeoutException as err:
              retry_count += 1
              
              logger.warning(
                  f"Timeout error: {err}. Retry {retry_count}/{max_retries}"
              )
              
              if retry_count < max_retries:
                  await asyncio.sleep(base_delay * (2 ** retry_count))
              else:
                  raise TimeoutError("update_node") from err
      
      raise NetworkError("update_node failed after maximum retries")