Productive Simple MCP

Instructions

Implementation Reference

• tools.py:614-653 (handler)

  Core handler function for the 'get_pages' MCP tool. Constructs API params from inputs, calls client.get_pages(), applies filter_page_list_response for list view sanitization, handles errors with ctx logging.

  async def get_pages(
      ctx: Context,
      project_id: int = None,
      creator_id: int = None,
      page_number: int = None,
      page_size: int = config.items_per_page
  ) -> ToolResult:
      """List pages (docs) with optional filters and pagination.
  
      Developer notes:
      - Supports project_id and creator_id filters.
      - Enforces configurable default page[size] if not provided.
      - Sorts by most recent updates first.
      - Applies utils.filter_response to sanitize (body excluded via type='pages').
      - Uses consistent scalar filters: filter[project_id][eq], filter[creator_id][eq]
      """
      try:
          await ctx.info("Fetching pages")
          params = {}
          if page_number is not None:
              params["page[number]"] = page_number
          params["page[size]"] = page_size
          if project_id is not None:
              params["filter[project_id][eq]"] = project_id
          if creator_id is not None:
              params["filter[creator_id][eq]"] = creator_id
          params["sort"] = "-updated_at"
          result = await client.get_pages(params=params if params else None)
          await ctx.info("Successfully retrieved pages")
          
          # For lists, remove heavy fields like body explicitly
          filtered = filter_page_list_response(result)
          
          return filtered
          
      except ProductiveAPIError as e:
          await _handle_productive_api_error(ctx, e, "pages")
      except Exception as e:
          await ctx.error(f"Unexpected error fetching pages: {str(e)}")
          raise e

• server.py:478-509 (registration)

  FastMCP tool registration for 'get_pages' including input schema via Annotated[Field] with descriptions and example usage. Delegates implementation to tools.get_pages.

  @mcp.tool
  async def get_pages(
      ctx: Context,
      project_id: Annotated[
          int, Field(description="Optional project ID to filter pages by")
      ] = None,
      creator_id: Annotated[
          int, Field(description="Optional creator ID to filter pages by")
      ] = None,
      page_number: Annotated[int, Field(description="Page number for pagination")] = None,
      page_size: Annotated[
          int, Field(description="Optional number of pages per page (max 200)")
      ] = None,
  ) -> Dict[str, Any]:
      """Get all pages/documents with optional filtering.
  
      Pages in Productive are documents that can contain rich text content,
      attachments, and are organized within projects.
  
      Returns:
          Dictionary containing pages with content, metadata, and relationships
  
      Example:
          get_pages(project_id=1234)  # Get all pages for a specific project
      """
      return await tools.get_pages(
          ctx,
          project_id=project_id,
          creator_id=creator_id,
          page_number=page_number,
          page_size=page_size,
      )

• productive_client.py:115-119 (helper)

  ProductiveClient HTTP API wrapper method called by the handler to fetch pages data from /pages endpoint with query params and full error/retry handling.

  async def get_pages(self, params: Optional[dict] = None) -> Dict[str, Any]:
      """Get all pages with optional filtering
      Supports filtering by project_id, creator_id, edited_at, id
      """
      return await self._request("GET", "/pages", params=params)

• utils.py:317-351 (helper)

  Supporting utility function applied in the handler to sanitize page list responses: removes heavy 'body' content, adds webapp_url, cleans nulls/empties, preserves essential metadata.

  def filter_page_list_response(response: Dict[str, Any]) -> Dict[str, Any]:
      """Filter page list responses to keep metadata and drop heavy body field.
  
      - Removes attributes.body from each page item
      - Keeps other attributes as-is after general cleaning
      - Preserves meta (cleaned) and adds webapp_url per item
      """
      if not isinstance(response, dict):
          return response
  
      filtered: Dict[str, Any] = {}
  
      # Process data array
      if "data" in response and isinstance(response["data"], list):
          filtered_data = []
          for item in response["data"]:
              if isinstance(item, dict) and item.get("type") == "pages":
                  new_item = {"id": item.get("id"), "type": item.get("type")}
                  attrs = item.get("attributes", {})
                  if isinstance(attrs, dict):
                      # Copy attributes without body
                      new_attrs = dict(attrs)
                      new_attrs.pop("body", None)
                      new_item["attributes"] = new_attrs
                  _add_webapp_url(new_item)
                  filtered_data.append(new_item)
              else:
                  filtered_data.append(item)
          filtered["data"] = filtered_data
  
      # Keep meta if present (has useful info like total_count)
      if "meta" in response:
          filtered["meta"] = _clean_meta_object(response["meta"])
  
      return remove_null_and_empty(filtered)