WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:167-190 (handler)

  The primary handler function decorated with @mcp.tool(name="workflowy_get_node"), implementing the tool logic: acquires client, handles rate limiting, calls client.get_node(), and returns the node.

  @mcp.tool(name="workflowy_get_node", description="Retrieve a specific WorkFlowy node by ID")
  async def get_node(node_id: str) -> WorkFlowyNode:
      """Retrieve a specific WorkFlowy node.
  
      Args:
          node_id: The ID of the node to retrieve
  
      Returns:
          The requested WorkFlowy node
      """
      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/models/node.py:8-137 (schema)

  Pydantic model WorkFlowyNode defining the structure and validation for the tool's return type.

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

• src/workflowy_mcp/server.py:25-30 (helper)

  Helper function that provides the shared WorkFlowyClient instance used in the tool handler.

  def get_client() -> WorkFlowyClient:
      """Get the global WorkFlowy client instance."""
      global _client
      if _client is None:
          raise RuntimeError("WorkFlowy client not initialized. Server not started properly.")
      return _client