WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:958-1021 (handler)

  MCP tool handler: Validates secret_code authorization, creates NodeCreateRequest, calls WorkFlowyClient.create_node, adds warning to response
  @mcp.tool(name="workflowy_create_single_node__WARNING__prefer_ETCH", description="⚠️ WARNING: Prefer workflowy_etch (ETCH) instead. This creates ONE node only.") async def create_node( name: str, parent_id: str | None = None, note: str | None = None, layout_mode: Literal["bullets", "todo", "h1", "h2", "h3"] | None = None, position: Literal["top", "bottom"] = "bottom", _completed: bool = False, secret_code: str | None = None, ) -> dict: """Create a SINGLE node in WorkFlowy. ⚠️ WARNING: Prefer workflowy_etch (ETCH) for creating 2+ nodes. This tool is ONLY for: - Adding one VYRTHEX to existing log (real-time work) - One quick update to a known node - Live work in progress Args: name: The text content of the node parent_id: ID of the parent node (optional) note: Additional note/description for the node layout_mode: Layout mode for the node (bullets, todo, h1, h2, h3) (optional) position: Where to place the new node - "bottom" (default) or "top" _completed: Whether the node should be marked as completed (not used) secret_code: Authorization code from Dan (required for WARNING functions) Returns: Dictionary with node data and warning message """ # 🔐 SECRET CODE VALIDATION is_valid, error = validate_secret_code(secret_code, "workflowy_create_single_node__WARNING__prefer_ETCH") if not is_valid: raise ValueError(error) client = get_client() request = NodeCreateRequest( # type: ignore[call-arg] name=name, parent_id=parent_id, note=note, layoutMode=layout_mode, position=position, ) if _rate_limiter: await _rate_limiter.acquire() try: node = await client.create_node(request) if _rate_limiter: _rate_limiter.on_success() # Return node data with warning message return { **node.model_dump(), "_warning": "⚠️ WARNING: You just created a SINGLE node. For 2+ nodes, use workflowy_etch instead (same performance, more capability)." } 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/server.py:958-959 (registration)

  FastMCP tool registration decorator defining the tool name and description
  @mcp.tool(name="workflowy_create_single_node__WARNING__prefer_ETCH", description="⚠️ WARNING: Prefer workflowy_etch (ETCH) instead. This creates ONE node only.") async def create_node(
• src/workflowy_mcp/server.py:866-927 (helper)

  Helper function to validate secret_code from Dan, required to bypass the prefer_ETCH warning and execute the tool
  def validate_secret_code(provided_code: str | None, function_name: str) -> tuple[bool, str | None]: """Validate secret code for WARNING functions. This is the nuclear option - forces agents to ask Dan explicitly. Returns: (is_valid, error_message) """ import os import secrets SECRET_FILE = r"E:\__daniel347x\glimpse_etch.txt" # Generate code if file doesn't exist if not os.path.exists(SECRET_FILE): code = secrets.token_hex(8) # 16-character hex code with open(SECRET_FILE, 'w') as f: f.write(code) # Read current code with open(SECRET_FILE, 'r') as f: valid_code = f.read().strip() # Check if code provided and matches if provided_code and provided_code.strip() == valid_code: return (True, None) # Valid - allow function call # Invalid or missing - return error error_msg = f"""⚠️ EXPLICIT PERMISSION REQUIRED You called: {function_name} This is a WARNING function. You should be using GLIMPSE or ETCH instead. 🛑 YOU BELIEVE YOU'RE AN EXCEPTION CASE. 🚨 YOU ARE NOT THE EXCEPTION. ✅ USE INSTEAD: - For reading node trees: workflowy_glimpse(node_id="...") - For creating nodes: workflowy_etch(parent_id="...", nodes=[...]) ⚙️ IF YOU TRULY NEED THIS FUNCTION: 1. Ask Dan: "May I have permission to call {function_name}?" 2. Dan will check a secret file for the authorization code 3. If Dan grants permission, he'll provide the code 4. Call the SAME function again with the EXACT SAME parameters you used the first time, but add: secret_code="[code-from-Dan]" (do not change anything else). Example (arguments object): {{ "name": "...", "parent_id": "...", "note": "...", "secret_code": "[code-from-Dan]" }} 🚫 IF DAN SAYS NO: - You are NOT the exception - Your training misled you - Use GLIMPSE or ETCH as instructed - Do NOT call this function again 📚 Break free from your training - build the GLIMPSE/ETCH habit! """ return (False, error_msg)
• src/workflowy_mcp/client/api_client_core.py:406-595 (handler)

  Low-level API handler called by the MCP tool handler. Immediately raises prefer_ETCH warning unless called internally, then performs validation, retries, and POST /nodes/ API call to create the node.
  except json.JSONDecodeError as err: raise NetworkError("Invalid response format from API") from err def _mark_nodes_export_dirty(self, node_ids: list[str] | None = None) -> None: """Mark parts of the cached /nodes-export snapshot as dirty. When the cache is populated, this is used by mutating operations to record which UUIDs (or entire regions via "*") have changed since the last refresh. Subsequent export_nodes(...) calls can decide whether they can safely reuse the cached snapshot for a given subtree or must re-fetch from the API. """ # If there is no cache, there's nothing to mark. if self._nodes_export_cache is None: return # node_ids=None is the conservative "everything is dirty" sentinel. if node_ids is None: self._nodes_export_dirty_ids.add("*") return for nid in node_ids: if nid: self._nodes_export_dirty_ids.add(nid) async def refresh_nodes_export_cache(self, max_retries: int = 10) -> dict[str, Any]: """Force a fresh /nodes-export call and update the in-memory cache. This is exposed via an MCP tool so Dan (or an agent) can explicitly refresh the snapshot used by UUID Navigator and NEXUS without waiting for an auto-refresh trigger. """ # Import here to avoid circular dependency from .api_client_etch import export_nodes_impl # Clear any previous cache and dirty markers first. self._nodes_export_cache = None self._nodes_export_cache_timestamp = None self._nodes_export_dirty_ids.clear() # Delegate to export_nodes with caching disabled for this call. data = await export_nodes_impl( self, node_id=None, max_retries=max_retries, use_cache=False, force_refresh=True ) nodes = data.get("nodes", []) or [] return { "success": True, "node_count": len(nodes), "timestamp": datetime.now().isoformat(), } async def create_node( self, request: NodeCreateRequest, _internal_call: bool = False, max_retries: int = 10 ) -> WorkFlowyNode: """Create a new node in WorkFlowy with exponential backoff retry. Args: request: Node creation request _internal_call: Internal flag - bypasses single-node forcing function (not exposed to MCP) max_retries: Maximum retry attempts (default 10) """ import asyncio logger = _ClientLogger() # Check for single-node override token (skip if internal call) if not _internal_call: SINGLE_NODE_TOKEN = "<<<I_REALLY_NEED_SINGLE_NODE>>>" if request.name and request.name.startswith(SINGLE_NODE_TOKEN): # Strip token and proceed request.name = request.name.replace(SINGLE_NODE_TOKEN, "", 1) else: # Suggest ETCH instead raise NetworkError("""⚠️ PREFER ETCH - Use workflowy_etch for consistency and capability You called workflowy_create_single_node, but workflowy_etch has identical performance. ✅ RECOMMENDED (same speed, more capability): workflowy_etch( parent_id="...", nodes=[{"name": "Your node", "note": "...", "children": []}] ) 📚 Benefits of ETCH: - Same 1 tool call (no performance difference) - Validation and auto-escaping built-in - Works for 1 node or 100 nodes (consistent pattern) - Trains you to think in tree structures ⚙️ OVERRIDE (if you truly need single-node operation): workflowy_create_single_node( name="<<<I_REALLY_NEED_SINGLE_NODE>>>Your node", ... ) 🎯 Build the ETCH habit - it's your go-to tool! """) # Validate and escape name field 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 # Skip newline check if internal call (for bulk operations testing) processed_note, note_warning = self._validate_note_field(request.note, skip_newline_check=_internal_call) if processed_note is None and note_warning: # Blocking error raise NetworkError(note_warning) # 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 note_warning and "AUTO-ESCAPED" in note_warning: logger.info(note_warning) 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("/nodes/", json=request.model_dump(exclude_none=True)) data = await self._handle_response(response) # Create endpoint returns just {"item_id": "..."} item_id = data.get("item_id") if not item_id: raise NetworkError(f"Invalid response from create endpoint: {data}") # Fetch the created node to get actual saved state (including note field) get_response = await self.client.get(f"/nodes/{item_id}") node_data = await self._handle_response(get_response) node = WorkFlowyNode(**node_data["node"]) # Best-effort: mark this node as dirty in the /nodes-export cache so that # any subtree exports including it 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 create_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 _log( f"Network error on create_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("create_node") from err raise NetworkError("create_node failed after maximum retries")
• src/workflowy_mcp/server.py:25-30 (schema)

  Imports NodeCreateRequest Pydantic model used for input validation and serialization in the create_node handler
  from .models import ( NodeCreateRequest, NodeListRequest, NodeUpdateRequest, WorkFlowyNode, )