Productive Simple MCP

Instructions

Implementation Reference

• tools.py:230-274 (handler)

  Core handler function implementing the get_comments tool logic: builds API parameters for pagination, filtering by project/task, adds sorting, calls the Productive client API, sanitizes response with filter_response, logs via ctx, handles API and general errors.
  async def get_comments( ctx: Context, project_id: int = None, task_id: int = None, page_number: int = None, page_size: int = config.items_per_page, extra_filters: dict = None ) -> ToolResult: """List comments with optional filters and pagination. Developer notes: - Pass-through for extra_filters (e.g., discussion_id, page_id, task_id). - Enforces configurable default page[size] if not provided. - Sort defaults to "-created_at" (most recent first). - Applies utils.filter_response to sanitize. - Uses consistent scalar filters: filter[project_id][eq], filter[task_id][eq] """ try: await ctx.info("Fetching comments") 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 task_id is not None: params["filter[task_id][eq]"] = task_id if extra_filters: params.update(extra_filters) # Add default sorting params["sort"] = "-created_at" result = await client.get_comments(params=params if params else None) await ctx.info("Successfully retrieved comments") filtered = filter_response(result) return filtered except ProductiveAPIError as e: await _handle_productive_api_error(ctx, e, "comments") except Exception as e: await ctx.error(f"Unexpected error fetching comments: {str(e)}") raise e
• server.py:368-403 (registration)

  MCP tool registration for 'get_comments' using @mcp.tool decorator. Defines input schema via Annotated parameters with descriptions and delegates to the core handler in tools.get_comments.
  async def get_comments( ctx: Context, project_id: Annotated[ int, Field(description="Productive project ID to filter comments by") ] = None, task_id: Annotated[ int, Field(description="Productive task ID to filter comments by") ] = None, page_number: Annotated[int, Field(description="Page number for pagination")] = None, page_size: Annotated[ int, Field(description="Optional number of comments per page (max 200)") ] = None, extra_filters: Annotated[ dict, Field( description="Additional Productive query filters using API syntax. Common filters: filter[project_id][eq] (ID), filter[task_id][eq] (ID), filter[discussion_id][eq] (ID)." ), ] = None, ) -> Dict[str, Any]: """Get all comments across projects and tasks with full context. Returns comprehensive comment data including: - Comment text, author, and timestamp - Parent entity (project, task, or other) with details - Discussion threads and replies - Attachments and file references - Mentions of team members or clients """ return await tools.get_comments( ctx, project_id=project_id, task_id=task_id, page_number=page_number, page_size=page_size, extra_filters=extra_filters, )
• productive_client.py:93-96 (helper)

  Low-level API client helper that performs the actual HTTP GET request to Productive's /comments endpoint with provided parameters.
  async def get_comments(self, params: Optional[dict] = None) -> Dict[str, Any]: """Get all comments """ return await self._request("GET", "/comments", params=params)
• server.py:368-386 (schema)

  Input schema definition using Pydantic Annotated fields for tool parameters in the MCP registration.
  async def get_comments( ctx: Context, project_id: Annotated[ int, Field(description="Productive project ID to filter comments by") ] = None, task_id: Annotated[ int, Field(description="Productive task ID to filter comments by") ] = None, page_number: Annotated[int, Field(description="Page number for pagination")] = None, page_size: Annotated[ int, Field(description="Optional number of comments per page (max 200)") ] = None, extra_filters: Annotated[ dict, Field( description="Additional Productive query filters using API syntax. Common filters: filter[project_id][eq] (ID), filter[task_id][eq] (ID), filter[discussion_id][eq] (ID)." ), ] = None, ) -> Dict[str, Any]: