WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1067-1087 (registration)

  MCP tool registration for 'workflowy_get_node' (deprecated handler that raises ValueError directing to use workflowy_glimpse)
  @mcp.tool(name="workflowy_get_node", description="DEPRECATED: Use workflowy_glimpse (GLIMPSE) instead") async def get_node_base(node_id: str) -> dict: """Deprecated - use GLIMPSE instead.""" raise ValueError("""⚠️ FUNCTION RENAMED The function 'workflowy_get_node' has been renamed to 'workflowy_get_node__WARNING__prefer_glimpse'. BUT MORE IMPORTANTLY: Use workflowy_glimpse (GLIMPSE command) instead! ✅ RECOMMENDED: workflowy_glimpse(node_id="...") Returns: {"root": {...}, "children": [...]} with complete tree structure. GLIMPSE is better: - Gets root node metadata (name, note) - Gets full children tree (not just direct children) - One call gets everything 📚 Build the GLIMPSE habit! """)
• src/workflowy_mcp/server.py:1090-1123 (handler)

  Functional MCP handler (under warning name) that validates secret_code and calls client.get_node(node_id)
  @mcp.tool(name="workflowy_get_node__WARNING__prefer_glimpse", description="⚠️ WARNING: Prefer workflowy_glimpse (GLIMPSE) for reading trees. Retrieve a specific WorkFlowy node by ID") async def get_node( node_id: str, secret_code: str | None = None, ) -> WorkFlowyNode: """Retrieve a specific WorkFlowy node. Args: node_id: The ID of the node to retrieve secret_code: Authorization code from Dan (required for WARNING functions) Returns: The requested WorkFlowy node """ # 🔐 SECRET CODE VALIDATION is_valid, error = validate_secret_code(secret_code, "workflowy_get_node__WARNING__prefer_glimpse") if not is_valid: raise ValueError(error) client = get_client() if _rate_limiter: await _rate_limiter.acquire() try: node = await client.get_node(node_id) 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/client/api_client_core.py:706-769 (handler)

  Core implementation of get_node: performs HTTP GET /nodes/{node_id}, handles retries, rate limits, errors, and parses into WorkFlowyNode
  async def get_node(self, node_id: str, max_retries: int = 10) -> WorkFlowyNode: """Retrieve a specific node by ID with exponential backoff retry. Args: node_id: The ID of the node to retrieve max_retries: Maximum retry attempts (default 10) """ import asyncio 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.get(f"/nodes/{node_id}") data = await self._handle_response(response) # API returns {"node": {...}} structure if isinstance(data, dict) and "node" in data: return WorkFlowyNode(**data["node"]) else: # Fallback for unexpected format return WorkFlowyNode(**data) 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 get_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 get_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("get_node") from err raise NetworkError("get_node failed after maximum retries")
• src/workflowy_mcp/models/node.py:8-137 (schema)

  Pydantic model defining the output schema WorkFlowyNode for get_node response
  class WorkFlowyNode(BaseModel): """Represents a single node in the WorkFlowy outline hierarchy.""" # API fields (what the API actually returns) id: str = Field(..., description="Unique identifier for the node") name: str | None = Field( None, validation_alias=AliasChoices("name", "nm"), description="Text content of the node" ) note: str | None = Field( None, validation_alias=AliasChoices("note", "no"), description="Note content attached to the node", ) priority: int | None = Field(None, description="Sort order") data: dict[str, Any] | None = Field(None, description="Node data including layoutMode") createdAt: int | None = Field( None, validation_alias=AliasChoices("createdAt", "created"), description="Creation timestamp (Unix timestamp)", ) modifiedAt: int | None = Field( None, validation_alias=AliasChoices("modifiedAt", "modified"), description="Last modification timestamp", ) completedAt: int | None = Field( None, description="Completion timestamp (null if not completed)" ) # Nested structure fields children: list["WorkFlowyNode"] | None = Field(None, alias="ch", description="Child nodes") parent_id: str | None = Field(None, alias="parentId", description="Parent node ID") # Handle 'cp' field for backward compatibility - we'll compute from completedAt completed_flag: bool | None = Field( None, alias="cp", description="Completion status (for tests)" ) @property def layoutMode(self) -> str | None: """Extract layoutMode from data field.""" if self.data and isinstance(self.data, dict): return self.data.get("layoutMode") return None # Backward compatibility aliases for tests @property def nm(self) -> str | None: """Backward compatibility for name field.""" return self.name @property def no(self) -> str | None: """Backward compatibility for note field.""" return self.note @property def cp(self) -> bool: """Backward compatibility for completed status.""" # Use completed_flag if it was set (from tests), otherwise check completedAt if self.completed_flag is not None: return self.completed_flag return self.completedAt is not None @property def ch(self) -> list["WorkFlowyNode"] | None: """Backward compatibility for children field.""" return self.children @property def created(self) -> int: """Backward compatibility for created timestamp.""" return self.createdAt or 0 @property def modified(self) -> int: """Backward compatibility for modified timestamp.""" return self.modifiedAt or 0 @field_validator("id") @classmethod def validate_id(cls, v: str) -> str: """Ensure ID is non-empty.""" if not v or not v.strip(): raise ValueError("Node ID must be non-empty") return v @field_validator("createdAt", "modifiedAt", "completedAt") @classmethod def validate_timestamp(cls, v: int | None) -> int | None: """Ensure timestamps are positive.""" if v is not None and v <= 0: raise ValueError("Timestamp must be positive") return v def model_dump(self, **kwargs: Any) -> dict[str, Any]: """Custom serialization to include backward compatibility fields.""" data: dict[str, Any] = super().model_dump(**kwargs) # Add backward compatibility fields for tests data["nm"] = self.name data["no"] = self.note data["cp"] = self.cp data["ch"] = self.children data["created"] = self.createdAt or 0 data["modified"] = self.modifiedAt or 0 return data class Config: """Pydantic model configuration.""" populate_by_name = True # Allow both field names and aliases json_schema_extra = { "example": { "id": "node-123", "name": "Example Node", "note": "This is a note", "priority": 1, "layoutMode": "bullets", "createdAt": 1704067200, "modifiedAt": 1704067200, "completedAt": None, "children": [], } } # Enable forward references for recursive model WorkFlowyNode.model_rebuild()