WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1126-1147 (registration)

  Registration and deprecated handler stub for 'workflowy_list_nodes' tool. Raises ValueError with deprecation notice recommending workflowy_glimpse.

  @mcp.tool(name="workflowy_list_nodes", description="DEPRECATED: Use workflowy_glimpse (GLIMPSE) instead")
  async def list_nodes_base(parent_id: str | None = None) -> dict:
      """Deprecated - use GLIMPSE instead."""
      raise ValueError("""⚠️ FUNCTION RENAMED
  
  The function 'workflowy_list_nodes' has been renamed to 'workflowy_list_nodes__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 full nested tree (not just direct children)
  - Gets root node metadata
  - More efficient
  
  📚 Build the GLIMPSE habit!
  """)

• src/workflowy_mcp/server.py:1149-1191 (handler)

  Full handler logic (under warning suffixed name). Validates secret_code, creates NodeListRequest, calls client.list_nodes, returns nodes list.

  @mcp.tool(name="workflowy_list_nodes__WARNING__prefer_glimpse", description="⚠️ WARNING: Prefer workflowy_glimpse (GLIMPSE) for reading trees. List WorkFlowy nodes (omit parent_id for root)")
  async def list_nodes(
      parent_id: str | None = None,
      secret_code: str | None = None,
  ) -> dict:
      """List WorkFlowy nodes.
  
      Args:
          parent_id: ID of parent node to list children for
                     (omit or pass None to list root nodes - parameter won't be sent to API)
          secret_code: Authorization code from Dan (required for WARNING functions)
  
      Returns:
          Dictionary with 'nodes' list and 'total' count
      """
      # 🔐 SECRET CODE VALIDATION
      is_valid, error = validate_secret_code(secret_code, "workflowy_list_nodes__WARNING__prefer_glimpse")
      if not is_valid:
          raise ValueError(error)
      
      client = get_client()
  
      request = NodeListRequest(  # type: ignore[call-arg]
          parentId=parent_id,
      )
  
      if _rate_limiter:
          await _rate_limiter.acquire()
  
      try:
          nodes, total = await client.list_nodes(request)
          if _rate_limiter:
              _rate_limiter.on_success()
          return {
              "nodes": [node.model_dump() for node in nodes],
              "total": total,
              "_warning": "⚠️ For reading multiple nodes or full trees, use workflowy_glimpse (GLIMPSE) instead for efficiency"
          }
      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:52-56 (schema)

  Pydantic input schema model NodeListRequest used by list_nodes handlers.

  class NodeListRequest(BaseModel):
      """Request parameters for listing nodes."""
  
      parentId: str | None = Field(None, description="Parent node ID to list children for")

• src/workflowy_mcp/client/api_client_core.py:771-846 (handler)

  Core client-side implementation: HTTP GET /nodes with parent_id param, response parsing to WorkFlowyNode list, full retry/rate-limit handling.

  async def list_nodes(self, request: NodeListRequest, max_retries: int = 10) -> tuple[list[WorkFlowyNode], int]:
      """List nodes with optional filtering and exponential backoff retry.
      
      Args:
          request: Node list request
          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:
              # exclude_none=True ensures parent_id is omitted entirely for root nodes
              # (API requires absence of parameter, not null value)
              # Build params manually to ensure snake_case (API expects parent_id not parentId)
              params = {}
              if request.parentId is not None:
                  params['parent_id'] = request.parentId
              response = await self.client.get("/nodes", params=params)
              response_data: list[Any] | dict[str, Any] = await self._handle_response(response)
  
              # Assuming API returns an array of nodes directly
              # (Need to verify actual response structure)
              nodes: list[WorkFlowyNode] = []
              if isinstance(response_data, dict):
                  if "nodes" in response_data:
                      nodes = [WorkFlowyNode(**node_data) for node_data in response_data["nodes"]]
              elif isinstance(response_data, list):
                  nodes = [WorkFlowyNode(**node_data) for node_data in response_data]
  
              total = len(nodes)  # API doesn't provide a total count
              return nodes, total
              
          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 list_nodes. 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 list_nodes: {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("list_nodes") from err
      
      raise NetworkError("list_nodes failed after maximum retries")