Productive Simple MCP

Instructions

Implementation Reference

• tools.py:51-95 (handler)

  Core handler function that constructs API parameters based on inputs (pagination, sorting, filters), calls ProductiveClient.get_tasks, applies response filtering, handles API errors and logs via MCP Context.

  async def get_tasks(
      ctx: Context,
      page_number: int = None,
      page_size: int = config.items_per_page,
      sort: str = "-last_activity_at",
      project_id: int = None,
      user_id: int = None,
      extra_filters: dict = None
  ) -> ToolResult:
      """
      List tasks with optional filters and pagination.
  
      Developer notes:
      - project_id and user_id are converted to Productive API filters.
      - extra_filters is passed through directly to the API (e.g., filter[status][eq]).
      - Enforces a configurable default page[size] for consistency when not provided.
      - Sort supports Productive's allowed fields (e.g., last_activity_at, created_at, due_date).
      - Response is cleaned with utils.filter_task_list_response (excludes descriptions for lean lists).
      """
      try:
          await ctx.info("Fetching tasks")
          params = {}
          if page_number is not None:
              params["page[number]"] = page_number
          params["page[size]"] = page_size
          if sort:
              params["sort"] = sort
          if project_id is not None:
              params["filter[project_id][eq]"] = project_id
          if user_id is not None:
              params["filter[assignee_id][eq]"] = user_id
          if extra_filters:
              params.update(extra_filters)
  
          result = await client.get_tasks(params=params if params else None)
          await ctx.info("Successfully retrieved tasks")
  
          filtered = filter_task_list_response(result)
          return filtered
  
      except ProductiveAPIError as e:
          await _handle_productive_api_error(ctx, e, "tasks")
      except Exception as e:
          await ctx.error(f"Unexpected error fetching tasks: {str(e)}")
          raise e

• server.py:202-248 (registration)

  FastMCP tool registration (@mcp.tool) defining the 'get_tasks' tool name, input schema via Annotated[Pydantic Field] parameters with descriptions/validation, and delegation to the implementation in tools.get_tasks.

  async def get_tasks(
      ctx: Context,
      project_id: Annotated[int, Field(description="Filter tasks by project ID")] = None,
      user_id: Annotated[
          int, Field(description="Filter tasks by assignee/user ID")
      ] = None,
      page_number: Annotated[int, Field(description="Page number for pagination")] = None,
      page_size: Annotated[
          int, Field(description="Optional number of tasks per page (max 200)")
      ] = None,
      sort: Annotated[
          str,
          Field(
              description="Sort parameter (e.g., 'last_activity_at', '-last_activity_at', 'created_at', 'due_date'). Use '-' prefix for descending order. Defaults to '-last_activity_at' (most recent first)."
          ),
      ] = "-last_activity_at",
      extra_filters: Annotated[
          dict,
          Field(
              description="Additional Productive query filters using API syntax. Common filters: filter[status][eq] (1: open, 2: closed), filter[due_date][gte] (date), filter[workflow_status_category_id][eq] (1: not started, 2: started, 3: closed)."
          ),
      ] = None,
  ) -> Dict[str, Any]:
      """Get tasks with optional filtering and pagination.
  
      Supports filtering by project, assignee, status, and other criteria.
      All parameters are optional - omit to fetch all tasks.
  
      Example of extra_filters:
      - filter[status][eq]=1: Open tasks
      - filter[status][eq]=2: Closed tasks
      - filter[workflow_status_category_id][eq]=3: Workflow closed status
      - filter[board_status][eq]=1: Active board tasks
  
      Returns:
          Dictionary of tasks matching the provided filters
      """
      return await tools.get_tasks(
          ctx,
          page_number=page_number,
          page_size=page_size,
          sort=sort,
          project_id=project_id,
          user_id=user_id,
          extra_filters=extra_filters,
      )

• productive_client.py:81-87 (helper)

  ProductiveClient helper method that performs the actual HTTP API call to GET /tasks, automatically includes workflow_status relation, handles retries/errors.

  async def get_tasks(self, params: Optional[dict] = None) -> Dict[str, Any]:
      """Get all tasks with workflow_status always included
      """
      if params is None:
          params = {}
      params["include"] = "workflow_status"
      return await self._request("GET", "/tasks", params=params)