Productive Simple MCP

Instructions

Implementation Reference

• tools.py:295-333 (handler)

  Core implementation of the get_todos tool handler. Builds parameters for the Productive API, fetches todos via the client, filters the response, and handles errors with context logging.

  async def get_todos(
      ctx: Context,
      task_id: int = None,
      page_number: int = None,
      page_size: int = config.items_per_page,
      extra_filters: dict = None
  ) -> ToolResult:
      """List todo checklist items with optional filters.
  
      Developer notes:
      - task_id is an int; API expects filter[task_id] to be array or scalar; we send scalar.
      - Enforces configurable default page[size] when not provided.
      - Use extra_filters for status ints (1=open, 2=closed) or assignee filters.
      - Sorting not supported by API - uses default order.
      - Applies utils.filter_response.
      """
      try:
          await ctx.info("Fetching todos")
          params = {}
          if page_number is not None:
              params["page[number]"] = page_number
          params["page[size]"] = page_size
          if task_id is not None:
              params["filter[task_id]"] = [task_id]
          if extra_filters:
              params.update(extra_filters)
  
          result = await client.get_todos(params=params if params else None)
          await ctx.info("Successfully retrieved todos")
          
          filtered = filter_response(result)
          
          return filtered
          
      except ProductiveAPIError as e:
          await _handle_productive_api_error(ctx, e, "todos")
      except Exception as e:
          await ctx.error(f"Unexpected error fetching todos: {str(e)}")
          raise e

• server.py:425-457 (registration)

  MCP tool registration for 'get_todos' using @mcp.tool decorator. Delegates to the handler in tools.py and includes input schema via Annotated Fields.

  @mcp.tool
  async def get_todos(
      ctx: Context,
      task_id: Annotated[
          int, Field(description="Productive task ID to filter todos by")
      ] = None,
      page_number: Annotated[int, Field(description="Page number for pagination")] = None,
      page_size: Annotated[
          int, Field(description="Optional number of todos per page (max 200)")
      ] = None,
      extra_filters: Annotated[
          dict,
          Field(
              description="Additional Productive query filters using API syntax. Common filters: filter[task_id][eq] (ID), filter[status][eq] (1: open, 2: closed), filter[assignee_id][eq] (ID)."
          ),
      ] = None,
  ) -> Dict[str, Any]:
      """Get all todo checklist items across all tasks and projects.
  
      Returns comprehensive todo data including:
      - Checkbox items within tasks for granular tracking
      - Completion status and assignee information
      - Parent task details with project context
      - Due dates and priority relative to parent task
      - Estimated vs actual time for checklist items
      """
      return await tools.get_todos(
          ctx,
          task_id=task_id,
          page_number=page_number,
          page_size=page_size,
          extra_filters=extra_filters,
      )

• server.py:427-441 (schema)

  Input schema definition using Pydantic Annotated and Field for the get_todos tool parameters, providing descriptions and types for MCP.

      ctx: Context,
      task_id: Annotated[
          int, Field(description="Productive task ID to filter todos by")
      ] = None,
      page_number: Annotated[int, Field(description="Page number for pagination")] = None,
      page_size: Annotated[
          int, Field(description="Optional number of todos per page (max 200)")
      ] = None,
      extra_filters: Annotated[
          dict,
          Field(
              description="Additional Productive query filters using API syntax. Common filters: filter[task_id][eq] (ID), filter[status][eq] (1: open, 2: closed), filter[assignee_id][eq] (ID)."
          ),
      ] = None,
  ) -> Dict[str, Any]:

• productive_client.py:102-105 (helper)

  Helper method in the ProductiveClient class that performs the actual HTTP GET request to the /todos endpoint with optional parameters.

  async def get_todos(self, params: Optional[dict] = None) -> Dict[str, Any]:
      """Get all todos
      """
      return await self._request("GET", "/todos", params=params)