list task query

README.md

@@ -110,11 +110,15 @@ If authentication is enabled (see below), send the taskowl API key as
 
 | Category | Tools |
 |---|---|
-| **Tasks** | `list_tasks`, `get_task`, `get_task_timeline`, `get_task_chain`, `get_task_summary`, `list_orphaned_tasks` |
+| **Tasks** | `list_tasks`, `get_task`, `get_task_timeline`, `get_task_chain`, `get_task_summary`, `list_task_types`, `list_orphaned_tasks` |
 | **Task actions** | `revoke_task`, `retry_task` |
 | **Workers** | `get_worker_status`, `list_workers`, `get_worker_stats`, `shutdown_worker`, `scale_worker_pool`, `get_active_tasks` |
 
-**Total: 14 tools**
+**Total: 15 tools**
+
+`list_tasks` supports exact filters (`state`, `name`, `worker`, `since`), a partial
+case-insensitive `search` on the task name, `offset` for pagination, and `sort_by`
+(`timestamp` [default, newest-first], `name`, `state`, `worker`).
 
 ## Examples
 
@@ -123,6 +127,7 @@ Questions you can ask your AI assistant when the MCP server is connected:
 | Question | Tools used |
 |---|---|
 | "Show me failed tasks from the last hour" | `list_tasks` |
+| "Which task types are running?" | `list_task_types` |
 | "Which tasks are orphaned?" | `list_orphaned_tasks` |
 | "Show me the timeline for task abc" | `get_task_timeline` |
 | "What's the retry chain for task abc?" | `get_task_chain` |
@@ -252,7 +257,7 @@ retries, and metrics. Interactive docs are available at:
 
 | Area | Endpoints |
 |------|-----------|
-| **Tasks** | `GET /api/tasks`, `GET /api/tasks/{id}`, `GET /api/tasks/{id}/timeline`, `GET /api/tasks/{id}/chain`, `GET /api/tasks/summary`, `GET /api/tasks/orphaned` |
+| **Tasks** | `GET /api/tasks`, `GET /api/tasks/{id}`, `GET /api/tasks/{id}/timeline`, `GET /api/tasks/{id}/chain`, `GET /api/tasks/summary`, `GET /api/tasks/types`, `GET /api/tasks/orphaned` |
 | **Task actions** | `POST /api/tasks/{id}/revoke`, `POST /api/tasks/{id}/retry` |
 | **Workers** | `GET /api/workers`, `GET /api/workers/list`, `GET /api/workers/{name}/stats`, `GET /api/workers/active-tasks` |
 | **Worker actions** | `POST /api/workers/{name}/shutdown`, `POST /api/workers/{name}/scale` |

src/taskowl/main.py

@@ -21,6 +21,7 @@
     get_task_timeline_query,
     get_worker_status_query,
     list_orphaned_tasks_query,
+    list_task_types_query,
     list_tasks_query,
 )
 from taskowl.workers import (
@@ -112,11 +113,29 @@ async def api_list_tasks(
     worker: str | None = None,
     since: str | None = None,
     limit: int = 100,
+    search: str | None = None,
+    offset: int = 0,
+    sort_by: str = "timestamp",
     session: AsyncSession = Depends(get_db),
     _: None = Depends(verify_api_key),
-) -> list[dict]:
+) -> list[dict] | dict:
     """List tasks with optional filters."""
-    return await list_tasks_query(state, name, worker, since, limit, session)
+    result = await list_tasks_query(
+        state, name, worker, since, limit, search, offset, sort_by, session
+    )
+    if isinstance(result, dict) and "error" in result:
+        raise HTTPException(status_code=400, detail=result["error"])
+    return result
+
+
+@app.get("/api/tasks/types")
+async def api_list_task_types(
+    limit: int = 50,
+    session: AsyncSession = Depends(get_db),
+    _: None = Depends(verify_api_key),
+) -> list[dict]:
+    """List distinct task names (types) with their task counts."""
+    return await list_task_types_query(limit, session)
 
 
 @app.get("/api/tasks/summary")

src/taskowl/mcp/tools.py

@@ -31,18 +31,24 @@ async def list_tasks(
         worker: str | None = None,
         since: str | None = None,
         limit: int = 100,
+        search: str | None = None,
+        offset: int = 0,
+        sort_by: str = "timestamp",
     ) -> list[dict]:
         """List tasks with optional filters.
 
         Args:
             state: Filter by state (received, started, succeeded, failed, retried, revoked)
-            name: Filter by task name
+            name: Filter by exact task name
             worker: Filter by worker hostname
             since: Only tasks created after this datetime (ISO 8601)
             limit: Max number of tasks to return (default: 100)
+            search: Partial, case-insensitive match on task name
+            offset: Number of tasks to skip (for pagination)
+            sort_by: Sort key (timestamp, name, state, worker). Defaults to timestamp, newest first
         """
         async with httpx.AsyncClient() as client:
-            params = {"limit": limit}
+            params = {"limit": limit, "offset": offset, "sort_by": sort_by}
             if state is not None:
                 params["state"] = state
             if name is not None:
@@ -51,6 +57,8 @@ async def list_tasks(
                 params["worker"] = worker
             if since is not None:
                 params["since"] = since
+            if search is not None:
+                params["search"] = search
 
             response = await client.get(
                 f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/tasks",
@@ -60,6 +68,25 @@ async def list_tasks(
             response.raise_for_status()
             return response.json()
 
+    @server.tool(
+        name="list_task_types",
+        description="List distinct task names (types) with their task counts",
+    )
+    async def list_task_types(limit: int = 50) -> list[dict]:
+        """List distinct task names with task counts.
+
+        Args:
+            limit: Max number of task types to return (default: 50)
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/tasks/types",
+                params={"limit": limit},
+                headers=_get_headers(),
+            )
+            response.raise_for_status()
+            return response.json()
+
     @server.tool(
         name="list_orphaned_tasks",
         description=(

src/taskowl/queries.py

@@ -21,25 +21,35 @@ async def list_tasks_query(
     worker: str | None = None,
     since: str | None = None,
     limit: int = 100,
+    search: str | None = None,
+    offset: int = 0,
+    sort_by: str = "timestamp",
     session: AsyncSession | None = None,
-) -> list[dict]:
+) -> list[dict] | dict:
     """List tasks with optional filters.
 
     Args:
         state: Filter by state (received, started, succeeded, failed, retried, revoked)
-        name: Filter by task name
+        name: Filter by exact task name
         worker: Filter by worker hostname
         since: Only tasks created after this datetime (ISO 8601)
         limit: Max number of tasks to return (default: 100)
+        search: Partial, case-insensitive match on task name
+        offset: Number of tasks to skip (for pagination)
+        sort_by: Sort key (timestamp, name, state, worker). Defaults to timestamp, newest first
         session: Optional database session (for testing)
 
     Returns:
         List of task dictionaries
     """
     if session is None:
         async with async_session_maker() as session:
-            return await _list_tasks_impl(session, state, name, worker, since, limit)
-    return await _list_tasks_impl(session, state, name, worker, since, limit)
+            return await _list_tasks_impl(
+                session, state, name, worker, since, limit, search, offset, sort_by
+            )
+    return await _list_tasks_impl(
+        session, state, name, worker, since, limit, search, offset, sort_by
+    )
 
 
 async def _list_tasks_impl(
@@ -49,7 +59,10 @@ async def _list_tasks_impl(
     worker: str | None,
     since: str | None,
     limit: int,
-) -> list[dict]:
+    search: str | None,
+    offset: int,
+    sort_by: str,
+) -> list[dict] | dict:
     """Internal implementation of list_tasks_query."""
     # This approach works with both PostgreSQL and SQLite
     from sqlalchemy import func as sql_func
@@ -104,14 +117,33 @@ async def _list_tasks_impl(
         query = query.where(TaskEvent.event_type == state.lower())
     if name:
         query = query.where(task_names.c.name == name)
+    if search:
+        query = query.where(task_names.c.name.ilike(f"%{search}%"))
     if worker:
         query = query.where(TaskEvent.hostname == worker)
     if since:
         since_dt = datetime.fromisoformat(since)
         query = query.where(TaskEvent.timestamp >= since_dt)
 
-    # Apply limit
-    query = query.limit(limit)
+    # Apply ordering
+    sort_keys = {
+        "timestamp": TaskEvent.timestamp,
+        "name": task_names.c.name,
+        "state": TaskEvent.event_type,
+        "worker": TaskEvent.hostname,
+    }
+    if sort_by not in sort_keys:
+        return {"error": f"Invalid sort_by: {sort_by}. Must be one of {list(sort_keys)}"}
+    column = sort_keys[sort_by]
+    if sort_by == "timestamp":
+        # Newest first by default
+        query = query.order_by(column.desc())
+    else:
+        query = query.order_by(column.asc())
+
+    # Apply limit/offset
+    offset = max(offset, 0)
+    query = query.offset(offset).limit(limit)
 
     result = await session.execute(query)
     rows = result.all()
@@ -133,6 +165,59 @@ async def _list_tasks_impl(
     return task_list
 
 
+async def list_task_types_query(
+    limit: int = 50,
+    session: AsyncSession | None = None,
+) -> list[dict]:
+    """List distinct task names (types) with their task counts.
+
+    Args:
+        limit: Max number of task types to return (default: 50)
+        session: Optional database session (for testing)
+
+    Returns:
+        List of dicts with name and count, ordered by count descending
+    """
+    if session is None:
+        async with async_session_maker() as session:
+            return await _list_task_types_impl(session, limit)
+    return await _list_task_types_impl(session, limit)
+
+
+async def _list_task_types_impl(session: AsyncSession, limit: int) -> list[dict]:
+    """Internal implementation of list_task_types_query."""
+    from sqlalchemy import func as sql_func
+
+    # Task name is only present on the earliest named event per task
+    earliest_named_ts = (
+        select(
+            TaskEvent.task_id,
+            sql_func.min(TaskEvent.timestamp).label("name_ts"),
+        )
+        .where(TaskEvent.name.isnot(None))
+        .group_by(TaskEvent.task_id)
+        .subquery()
+    )
+    task_names = (
+        select(TaskEvent.task_id, TaskEvent.name)
+        .join(
+            earliest_named_ts,
+            (TaskEvent.task_id == earliest_named_ts.c.task_id)
+            & (TaskEvent.timestamp == earliest_named_ts.c.name_ts),
+        )
+        .subquery()
+    )
+
+    query = (
+        select(task_names.c.name, sql_func.count().label("count"))
+        .group_by(task_names.c.name)
+        .order_by(sql_func.count().desc(), task_names.c.name.asc())
+        .limit(limit)
+    )
+    result = await session.execute(query)
+    return [{"name": row[0], "count": row[1]} for row in result.all()]
+
+
 async def get_task_query(task_id: str, session: AsyncSession | None = None) -> dict:
     """Get detailed information about a specific task.
 

tests/test_api.py

@@ -113,6 +113,105 @@ async def test_api_list_tasks_with_filters(client: AsyncClient, db_session: Asyn
     assert data[0]["worker"] == "worker1@localhost"
 
 
+@pytest.mark.asyncio
+async def test_api_list_tasks_search_offset_sort(client: AsyncClient, db_session: AsyncSession):
+    """Test GET /api/tasks with search, offset, and sort_by params."""
+    now = datetime.now(UTC)
+    db_session.add(
+        TaskEvent(
+            event_type="received",
+            task_id=uuid.uuid4(),
+            timestamp=now + timedelta(seconds=2),
+            hostname="worker1@localhost",
+            name="zeta_task",
+        )
+    )
+    db_session.add(
+        TaskEvent(
+            event_type="received",
+            task_id=uuid.uuid4(),
+            timestamp=now + timedelta(seconds=1),
+            hostname="worker1@localhost",
+            name="alpha_task",
+        )
+    )
+    await db_session.commit()
+
+    # search partial match
+    response = await client.get("/api/tasks?search=TASK")
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) == 2
+
+    # search no match
+    response = await client.get("/api/tasks?search=nope")
+    assert response.status_code == 200
+    assert response.json() == []
+
+    # sort_by name ascending
+    response = await client.get("/api/tasks?sort_by=name")
+    assert response.status_code == 200
+    data = response.json()
+    assert [t["name"] for t in data] == ["alpha_task", "zeta_task"]
+
+    # invalid sort_by -> 400
+    response = await client.get("/api/tasks?sort_by=bogus")
+    assert response.status_code == 400
+
+    # offset skips newest first (default timestamp desc)
+    response = await client.get("/api/tasks?offset=1")
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+
+
+@pytest.mark.asyncio
+async def test_api_list_task_types(client: AsyncClient, db_session: AsyncSession):
+    """Test GET /api/tasks/types returns distinct names with counts."""
+    now = datetime.now(UTC)
+    db_session.add(
+        TaskEvent(
+            event_type="received",
+            task_id=uuid.uuid4(),
+            timestamp=now,
+            hostname="worker1@localhost",
+            name="app.tasks.alpha",
+        )
+    )
+    db_session.add(
+        TaskEvent(
+            event_type="received",
+            task_id=uuid.uuid4(),
+            timestamp=now,
+            hostname="worker1@localhost",
+            name="app.tasks.alpha",
+        )
+    )
+    db_session.add(
+        TaskEvent(
+            event_type="received",
+            task_id=uuid.uuid4(),
+            timestamp=now,
+            hostname="worker1@localhost",
+            name="app.tasks.beta",
+        )
+    )
+    await db_session.commit()
+
+    response = await client.get("/api/tasks/types")
+    assert response.status_code == 200
+    data = response.json()
+    by_name = {t["name"]: t["count"] for t in data}
+    assert by_name == {"app.tasks.alpha": 2, "app.tasks.beta": 1}
+
+
+@pytest.mark.asyncio
+async def test_api_list_task_types_empty(client: AsyncClient):
+    """Test GET /api/tasks/types with no data."""
+    response = await client.get("/api/tasks/types")
+    assert response.status_code == 200
+    assert response.json() == []
+
+
 @pytest.mark.asyncio
 async def test_api_get_task_summary_empty(client: AsyncClient):
     """Test GET /api/tasks/summary with no data."""