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()