get_tasks
Retrieve tasks from Productive.io with filtering by project, assignee, status, or custom criteria, plus pagination for organized task management.
Instructions
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | No | Filter tasks by project ID | |
| user_id | No | Filter tasks by assignee/user ID | |
| page_number | No | Page number for pagination | |
| page_size | No | Optional number of tasks per page (max 200) | |
| sort | No | 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 | No | 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). |
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)