Execute a task

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` |

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)}"}

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),

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",

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"]

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