workflowy_delete_node
Remove a node and its sub-items from WorkFlowy outlines to manage hierarchical task lists and maintain organized workflows.
Instructions
Delete a WorkFlowy node and all its children
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| node_id | Yes |
Implementation Reference
- src/workflowy_mcp/server.py:1194-1218 (handler)MCP tool handler function that registers the 'workflowy_delete_node' tool via @mcp.tool decorator. Acquires rate limiter, calls WorkFlowyClient.delete_node(node_id), handles exceptions including rate limits, and returns success status with deleted ID.@mcp.tool(name="workflowy_delete_node", description="Delete a WorkFlowy node and all its children") async def delete_node(node_id: str) -> dict: """Delete a WorkFlowy node and all its children. Args: node_id: The ID of the node to delete Returns: Dictionary with success status """ client = get_client() if _rate_limiter: await _rate_limiter.acquire() try: success = await client.delete_node(node_id) if _rate_limiter: _rate_limiter.on_success() return {"success": success, "deleted_id": node_id} 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
- Core HTTP API implementation in WorkFlowyClientCore.delete_node: sends DELETE /nodes/{node_id}, implements exponential backoff retries for rate limits/timeouts/network errors, marks nodes_export cache dirty, logs to reconcile file on retries.async def delete_node(self, node_id: str, max_retries: int = 10) -> bool: """Delete a node and all its children with exponential backoff retry. Args: node_id: The ID of the node to delete max_retries: Maximum retry attempts (default 10) """ import asyncio from .api_client_etch import _log_to_file_helper logger = _ClientLogger() 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.delete(f"/nodes/{node_id}") # Delete endpoint returns just a message, not nested data await self._handle_response(response) # If we reached here after one or more retries, log success to reconcile log if retry_count > 0: success_msg = ( f"delete_node {node_id} succeeded after {retry_count + 1}/{max_retries} attempts " f"following rate limiting or transient errors." ) logger.info(success_msg) _log_to_file_helper(success_msg, "reconcile") # Best-effort: mark this node as dirty so any subsequent # /nodes-export-based operations that rely on it will 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 True except RateLimitError as e: retry_count += 1 retry_after = getattr(e, 'retry_after', None) or (base_delay * (2 ** retry_count)) retry_msg = ( f"Rate limited on delete_node {node_id}. Retry after {retry_after}s. " f"Attempt {retry_count}/{max_retries}" ) logger.warning(retry_msg) _log_to_file_helper(retry_msg, "reconcile") if retry_count < max_retries: await asyncio.sleep(retry_after) else: final_msg = ( f"delete_node {node_id} exhausted retries ({retry_count}/{max_retries}) " f"due to rate limiting – aborting." ) logger.error(final_msg) _log_to_file_helper(final_msg, "reconcile") raise except NetworkError as e: retry_count += 1 logger.warning( f"Network error on delete_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("delete_node") from err raise NetworkError("delete_node failed after maximum retries")