diff --git a/README.md b/README.md index c0c1ed9..e97f103 100644 --- a/README.md +++ b/README.md @@ -111,11 +111,11 @@ 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_task_types`, `list_orphaned_tasks` | -| **Task actions** | `revoke_task`, `retry_task` | +| **Task actions** | `revoke_task`, `retry_task`, `execute_task` | | **Workers** | `get_worker_status`, `list_workers`, `get_worker_stats`, `shutdown_worker`, `scale_worker_pool`, `get_active_tasks` | | **Queues** | `list_queues` | -**Total: 16 tools** +**Total: 17 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` @@ -137,6 +137,7 @@ Questions you can ask your AI assistant when the MCP server is connected: | "How many messages are in each queue?" | `list_queues` | | "Shutdown worker celery@worker1" | `shutdown_worker` | | "Retry task abc" | `retry_task` | +| "Run myapp.tasks.process now" | `execute_task` | ## Architecture @@ -260,7 +261,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/types`, `GET /api/tasks/orphaned` | -| **Task actions** | `POST /api/tasks/{id}/revoke`, `POST /api/tasks/{id}/retry` | +| **Task actions** | `POST /api/tasks/{id}/revoke`, `POST /api/tasks/{id}/retry`, `POST /api/tasks/execute` | | **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` | | **Queues** | `GET /api/queues` | diff --git a/src/taskowl/actions.py b/src/taskowl/actions.py index a0f935d..dbaab7f 100644 --- a/src/taskowl/actions.py +++ b/src/taskowl/actions.py @@ -214,3 +214,66 @@ async def retry_task( } except Exception as e: return {"error": f"Failed to retry task: {str(e)}"} + + +def _parse_iso_datetime(value: str | None) -> datetime | None: + """Parse an ISO 8601 datetime string, raising ValueError on bad input.""" + if value is None: + return None + return _ensure_utc(datetime.fromisoformat(value)) + + +async def execute_task( + name: str, + args: list | None = None, + kwargs: dict | None = None, + queue: str | None = None, + countdown: int | None = None, + eta: str | None = None, + expires: str | None = None, + priority: int | None = None, +) -> dict: + """Execute a task by name, sending it to the Celery broker. + + Args: + name: Task name (e.g. 'myapp.tasks.process') + args: Positional arguments for the task + kwargs: Keyword arguments for the task + queue: Queue to send the task to (defaults to the app default) + countdown: Seconds to wait before the task runs + eta: ISO 8601 datetime before which the task should not run + expires: ISO 8601 datetime after which the task expires + priority: Queue priority (0-9, broker-dependent) + + Returns: + Dict with status, task_id, and name + """ + if not name: + return {"error": "Task name is required"} + + try: + eta_dt = _parse_iso_datetime(eta) + expires_dt = _parse_iso_datetime(expires) + except ValueError: + return {"error": "eta and expires must be valid ISO 8601 datetimes"} + + try: + app = _get_celery_app() + result = app.send_task( + name, + args=tuple(args or []), + kwargs=kwargs or {}, + queue=queue, + countdown=countdown, + eta=eta_dt, + expires=expires_dt, + priority=priority, + ) + return { + "status": "success", + "message": f"Task {name} has been sent", + "task_id": result.id, + "name": name, + } + except Exception as e: + return {"error": f"Failed to execute task: {str(e)}"} diff --git a/src/taskowl/main.py b/src/taskowl/main.py index 7e15298..de5c7a8 100644 --- a/src/taskowl/main.py +++ b/src/taskowl/main.py @@ -9,7 +9,7 @@ from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession -from taskowl.actions import retry_task, revoke_task +from taskowl.actions import execute_task, retry_task, revoke_task from taskowl.auth import verify_api_key from taskowl.config import settings from taskowl.database import close_db, get_db, init_db @@ -73,6 +73,19 @@ class TaskSummary(BaseModel): avg_runtime_seconds: float | None +class SendTaskRequest(BaseModel): + """Request model for executing a task by name.""" + + name: str + args: list | None = None + kwargs: dict | None = None + queue: str | None = None + countdown: int | None = None + eta: str | None = None + expires: str | None = None + priority: int | None = None + + @app.get("/health") async def health_check() -> dict[str, str]: """Health check endpoint.""" @@ -244,6 +257,31 @@ async def api_retry_task( return result +@app.post("/api/tasks/execute") +async def api_execute_task( + request: SendTaskRequest, + _: None = Depends(verify_api_key), +) -> dict: + """Execute a task by name, sending it to the Celery broker. + + Args: + request: Task name, args, kwargs, and scheduling options + """ + result = await execute_task( + request.name, + args=request.args, + kwargs=request.kwargs, + queue=request.queue, + countdown=request.countdown, + eta=request.eta, + expires=request.expires, + priority=request.priority, + ) + if "error" in result: + raise HTTPException(status_code=400, detail=result["error"]) + return result + + @app.get("/api/workers/list") async def api_list_workers( _: None = Depends(verify_api_key), diff --git a/src/taskowl/mcp/tools.py b/src/taskowl/mcp/tools.py index d9b86df..4b459d2 100644 --- a/src/taskowl/mcp/tools.py +++ b/src/taskowl/mcp/tools.py @@ -235,6 +235,57 @@ async def retry_task(task_id: str) -> dict: response.raise_for_status() return response.json() + @server.tool( + name="execute_task", + description="Execute a task by name, sending it to the Celery broker", + ) + async def execute_task( + name: str, + args: list | None = None, + kwargs: dict | None = None, + queue: str | None = None, + countdown: int | None = None, + eta: str | None = None, + expires: str | None = None, + priority: int | None = None, + ) -> dict: + """Execute a task by name. + + Args: + name: Task name (e.g. 'myapp.tasks.process') + args: Positional arguments for the task + kwargs: Keyword arguments for the task + queue: Queue to send the task to + countdown: Seconds to wait before the task runs + eta: ISO 8601 datetime before which the task should not run + expires: ISO 8601 datetime after which the task expires + priority: Queue priority (0-9, broker-dependent) + """ + async with httpx.AsyncClient() as client: + payload: dict = {"name": name} + if args is not None: + payload["args"] = args + if kwargs is not None: + payload["kwargs"] = kwargs + if queue is not None: + payload["queue"] = queue + if countdown is not None: + payload["countdown"] = countdown + if eta is not None: + payload["eta"] = eta + if expires is not None: + payload["expires"] = expires + if priority is not None: + payload["priority"] = priority + + response = await client.post( + f"http://{settings.taskowl_host}:{settings.taskowl_port}/api/tasks/execute", + json=payload, + headers=_get_headers(), + ) + response.raise_for_status() + return response.json() + @server.tool( name="list_workers", description="List all active Celery workers", diff --git a/tests/test_actions.py b/tests/test_actions.py index 1b43040..36b13e2 100644 --- a/tests/test_actions.py +++ b/tests/test_actions.py @@ -7,7 +7,7 @@ import pytest from sqlalchemy.ext.asyncio import AsyncSession -from taskowl.actions import retry_task, revoke_task +from taskowl.actions import execute_task, retry_task, revoke_task from taskowl.models import TaskEvent, WorkerEvent @@ -372,3 +372,91 @@ async def test_retry_orphaned_task(db_session: AsyncSession): assert result["status"] == "success" assert result["new_task_id"] == "new-task-id-orphan" + + +@pytest.mark.asyncio +async def test_execute_task_success(): + """Test executing a task by name.""" + with patch("taskowl.actions._get_celery_app") as mock_get_app: + mock_app = MagicMock() + mock_result = MagicMock() + mock_result.id = "task-id-123" + mock_app.send_task.return_value = mock_result + mock_get_app.return_value = mock_app + + result = await execute_task( + "myapp.tasks.process", + args=[1, "two"], + kwargs={"key": "value"}, + queue="high", + ) + + assert result["status"] == "success" + assert result["task_id"] == "task-id-123" + assert result["name"] == "myapp.tasks.process" + mock_app.send_task.assert_called_once_with( + "myapp.tasks.process", + args=(1, "two"), + kwargs={"key": "value"}, + queue="high", + countdown=None, + eta=None, + expires=None, + priority=None, + ) + + +@pytest.mark.asyncio +async def test_execute_task_defaults(): + """Test executing a task with only a name (defaults for everything else).""" + with patch("taskowl.actions._get_celery_app") as mock_get_app: + mock_app = MagicMock() + mock_result = MagicMock() + mock_result.id = "task-id-default" + mock_app.send_task.return_value = mock_result + mock_get_app.return_value = mock_app + + result = await execute_task("myapp.tasks.cleanup") + + assert result["status"] == "success" + assert result["task_id"] == "task-id-default" + mock_app.send_task.assert_called_once_with( + "myapp.tasks.cleanup", + args=(), + kwargs={}, + queue=None, + countdown=None, + eta=None, + expires=None, + priority=None, + ) + + +@pytest.mark.asyncio +async def test_execute_task_empty_name(): + """Test executing a task with an empty name.""" + result = await execute_task("") + assert "error" in result + assert "Task name is required" in result["error"] + + +@pytest.mark.asyncio +async def test_execute_task_invalid_eta(): + """Test executing a task with an invalid eta.""" + result = await execute_task("myapp.tasks.process", eta="not-a-date") + assert "error" in result + assert "valid ISO 8601" in result["error"] + + +@pytest.mark.asyncio +async def test_execute_task_celery_error(): + """Test executing a task when Celery raises an error.""" + with patch("taskowl.actions._get_celery_app") as mock_get_app: + mock_app = MagicMock() + mock_app.send_task.side_effect = Exception("Connection failed") + mock_get_app.return_value = mock_app + + result = await execute_task("myapp.tasks.process") + + assert "error" in result + assert "Failed to execute task" in result["error"] diff --git a/tests/test_api.py b/tests/test_api.py index bea2a46..e65af27 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -724,6 +724,53 @@ async def test_api_retry_task_wrong_state(client: AsyncClient, db_session: Async assert "succeeded" in data["detail"] +@pytest.mark.asyncio +async def test_api_execute_task_success(client: AsyncClient): + """Test POST /api/tasks/execute with a valid request.""" + with patch("taskowl.actions._get_celery_app") as mock_get_app: + mock_app = MagicMock() + mock_result = MagicMock() + mock_result.id = "task-id-123" + mock_app.send_task.return_value = mock_result + mock_get_app.return_value = mock_app + + response = await client.post( + "/api/tasks/execute", + json={ + "name": "myapp.tasks.process", + "args": [1, "two"], + "kwargs": {"key": "value"}, + "queue": "high", + }, + ) + assert response.status_code == 200 + data = response.json() + assert data["status"] == "success" + assert data["task_id"] == "task-id-123" + assert data["name"] == "myapp.tasks.process" + + +@pytest.mark.asyncio +async def test_api_execute_task_empty_name(client: AsyncClient): + """Test POST /api/tasks/execute with an empty name.""" + with patch("taskowl.actions._get_celery_app") as mock_get_app: + mock_get_app.return_value = MagicMock() + + response = await client.post("/api/tasks/execute", json={"name": ""}) + assert response.status_code == 400 + assert "Task name is required" in response.json()["detail"] + + +@pytest.mark.asyncio +async def test_api_execute_task_invalid_eta(client: AsyncClient): + """Test POST /api/tasks/execute with an invalid eta.""" + response = await client.post( + "/api/tasks/execute", json={"name": "myapp.tasks.process", "eta": "not-a-date"} + ) + assert response.status_code == 400 + assert "valid ISO 8601" in response.json()["detail"] + + # Worker management endpoint tests