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)