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]: