> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Async Jobs
> Submit and manage long-running agent jobs and recipes via HTTP API
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent
import asyncio
agent = Agent(name="job-agent", instructions="Run background jobs asynchronously.")
async def main():
job = await agent.astart("Start a long-running background job.")
print(job)
asyncio.run(main())
```
Submit long-running agent tasks and recipes, then retrieve results asynchronously via a jobs server.
The user submits a long job; the agent runs asynchronously and returns when the job completes.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
Client[๐ค Submit] --> API[๐ฐ๏ธ Jobs API]
API --> Store{๐ Job store}
Store -->|dev default| Mem[(๐งช InMemoryJobStore)]
Store -->|PRAISONAI_JOBS_DB_PATH set| SQL[(๐๏ธ SqliteJobStore)]
Mem --> Worker[๐ค Agent run]
SQL --> Worker
Worker --> Result[โ
Result / webhook]
Client -->|poll or SSE| Store
classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff
classDef ok fill:#10B981,stroke:#7C90A0,color:#fff
classDef input fill:#6366F1,stroke:#7C90A0,color:#fff
classDef warn fill:#F59E0B,stroke:#7C90A0,color:#fff
class Worker agent
class API,Mem,SQL tool
class Store warn
class Result ok
class Client input
```
## How It Works
The user submits a job, the server runs it in the background, and the result comes back on completion.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant JobsAPI
participant Agent
User->>JobsAPI: Submit job
JobsAPI-->>User: job_id
JobsAPI->>Agent: Run in background
Agent-->>JobsAPI: Result
User->>JobsAPI: Poll / stream / webhook
JobsAPI-->>User: Completed result
```
## Choose a Result Mode
Pick how the result is delivered once the job finishes.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Q{How long is the job?} -->|Seconds| P[Poll status]
Q -->|Minutes, watch progress| S[SSE stream]
Q -->|Long, react later| W[Webhook callback]
classDef config fill:#6366F1,stroke:#7C90A0,color:#fff
classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff
class Q config
class P,S,W tool
```
## Quick Start
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai import recipe
job = recipe.submit_job(
"my-recipe",
input={"query": "What is AI?"},
config={"max_tokens": 1000},
session_id="session_123",
timeout_sec=3600,
api_url="http://127.0.0.1:8005",
)
print(f"Job ID: {job.job_id}")
result = job.wait(poll_interval=5, timeout=300)
print(f"Result: {result}")
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
API_URL = "http://127.0.0.1:8005"
response = httpx.post(f"{API_URL}/api/v1/runs", json={"prompt": "Analyze data"})
job_id = response.json()["job_id"]
status = httpx.get(f"{API_URL}/api/v1/runs/{job_id}").json()
result = httpx.get(f"{API_URL}/api/v1/runs/{job_id}/result").json()
```
## Persistent Store
Point the Jobs API at a SQLite file so job state and idempotency keys survive restarts โ no code change, just one env var.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent
agent = Agent(
name="job-agent",
instructions="Answer research questions.",
)
# Run under a persistent Jobs API โ no code change needed.
# praisonai serve jobs --port 8005 --db /var/lib/praisonai/jobs.db
```
Construct the store yourself when you want explicit control:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.jobs import SqliteJobStore, create_app
app = create_app(store=SqliteJobStore(path="/var/lib/praisonai/jobs.db"))
```
`SqliteJobStore` and `create_app` are both exported from `praisonai.jobs`. `_build_default_store()` in `praisonai/jobs/server.py` selects the backend from the environment:
| Env var | Default | Effect |
| ------------------------------- | ----------- | --------------------------------------------------------------------------------------------------------------------------------------- |
| `PRAISONAI_JOBS_DB_PATH` | *(unset)* | When set, boot the Jobs API on `SqliteJobStore(path=...)`. When unset, fall back to `InMemoryJobStore` unless `ENVIRONMENT=production`. |
| `ENVIRONMENT` | *(unset)* | If `production` **and** `PRAISONAI_JOBS_DB_PATH` is unset, `create_app()` refuses to start (raises `RuntimeError`). |
| `PRAISONAI_MAX_CONCURRENT_JOBS` | `10` | Executor concurrency cap. |
| `PRAISONAI_JOB_TIMEOUT` | `3600` | Default per-job timeout, seconds. |
| `PRAISONAI_JOBS_API_KEY` | *(unset)* | Required when binding to a non-loopback host. |
| `PRAISONAI_JOBS_BIND_HOST` | `127.0.0.1` | Listen address. |
### Which store should I use?
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Start[๐ฆ Which store?] --> Prod{Production
deployment?}
Prod -->|Yes| SQL[SqliteJobStore
set PRAISONAI_JOBS_DB_PATH]
Prod -->|No| Dev{Need restart
survival?}
Dev -->|Yes| SQL
Dev -->|No| Mem[InMemoryJobStore
default in dev]
classDef question fill:#6366F1,stroke:#7C90A0,color:#fff
classDef answer fill:#10B981,stroke:#7C90A0,color:#fff
classDef store fill:#189AB4,stroke:#7C90A0,color:#fff
class Start,Prod,Dev question
class SQL answer
class Mem store
```
In production the Jobs API refuses to boot on the in-memory store. `_build_default_store()` raises `RuntimeError` with this message:
> Jobs API refused to start with the in-memory store in production. Set PRAISONAI\_JOBS\_DB\_PATH (e.g. /var/lib/praisonai/jobs.db) to persist jobs and idempotency keys across restarts, or explicitly pass store=InMemoryJobStore() to create\_app().
The check fires only when `ENVIRONMENT=production` **and** `PRAISONAI_JOBS_DB_PATH` is unset. Passing `store=InMemoryJobStore()` explicitly to `create_app()` bypasses the guard โ the documented escape hatch for ephemeral production workloads that accept losing job state on restart.
### Atomic Idempotency
`JobStore` exposes `save_if_absent(job)` so a duplicate submit resolves to a single job and side effects run exactly once.
* The default `InMemoryJobStore` implementation is a best-effort check-then-save: it looks up the idempotency key, then inserts if absent.
* `SqliteJobStore` overrides `save_if_absent()` to be **truly atomic** via a `UNIQUE` index on `idempotency_key`. Two concurrent submits with the same key race on the `INSERT`; the loser catches `IntegrityError` and returns the winning job, so both callers receive the same `job_id`.
* NULL idempotency keys are exempt from SQLite's `UNIQUE` constraint, so keyless jobs are never de-duplicated against each other.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant ClientA as Client A
participant ClientB as Client B
participant API as Jobs API
participant Store as SqliteJobStore
ClientA->>API: POST /runs (Idempotency-Key: k1)
ClientB->>API: POST /runs (Idempotency-Key: k1)
API->>Store: save_if_absent(job) [A]
API->>Store: save_if_absent(job) [B]
Store-->>API: job_id = J1 (A wins INSERT)
Store-->>API: job_id = J1 (B loses UNIQUE โ returns winner)
API-->>ClientA: job_id = J1
API-->>ClientB: job_id = J1
```
### Restart Recovery
On `SqliteJobStore` startup, `_reconcile_interrupted_jobs()` marks any row still in `QUEUED` or `RUNNING` as `FAILED`, with `error="Interrupted by service restart"` and `completed_at=now`.
A crashed executor leaves no worker to resume its in-flight jobs, so those rows would otherwise poll forever and pin an idempotency key to a job that can never complete. Terminal reconciliation gives callers โ and idempotent retries โ a definitive outcome.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Old as Process (crashed)
participant DB as jobs.db
participant New as Process (restart)
Old->>DB: job J1 = RUNNING
Note over Old: crash โ no executor left
New->>DB: _reconcile_interrupted_jobs()
DB-->>New: J1 was RUNNING
New->>DB: J1 = FAILED (error="Interrupted by service restart")
```
## Backpressure & Admission Control
A flood of submits now returns `503` instead of accepting an unbounded backlog.
The executor gates admission in two tiers: `max_concurrent` caps how many run bodies execute at once, while `max_queued` caps how many jobs are admitted (queued + running) before new submits are rejected.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
Submit[๐ค POST /runs] --> Check{running_tasks >= max_queued?}
Check -->|Yes| Reject[๐ 503 + Retry-After]
Check -->|No| Reserve[๐ Reserve slot]
Reserve --> Persist[๐พ Save QUEUED]
Persist --> Sem{Semaphore free?}
Sem -->|Yes| Run[๐ค Run body]
Sem -->|No| Wait[โณ Wait]
Wait --> Run
Run --> Done[โ
Complete]
classDef input fill:#6366F1,stroke:#7C90A0,color:#fff
classDef check fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef stop fill:#8B0000,stroke:#7C90A0,color:#fff
classDef run fill:#189AB4,stroke:#7C90A0,color:#fff
classDef ok fill:#10B981,stroke:#7C90A0,color:#fff
class Submit input
class Check,Sem check
class Reject stop
class Reserve,Persist,Wait,Run run
class Done ok
```
When admission is saturated, `JobExecutor.submit()` raises `JobQueueFull(retry_after=1.0)`. The router deletes the stranded `QUEUED` row via `store.delete(job.id)` โ freeing the idempotency key โ and returns `HTTPException(status_code=503, detail="Job queue is full; retry later.", headers={"Retry-After": "1"})`.
Set the ceiling on the constructor:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.jobs import JobExecutor, SqliteJobStore
executor = JobExecutor(
store=SqliteJobStore(path="/var/lib/praisonai/jobs.db"),
max_concurrent=10, # run at most 10 bodies at once
max_queued=50, # admit at most 50 (queued + running); default is max_concurrent * 10
)
```
| Constructor arg | Default | Effect |
| --------------- | ------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------ |
| `max_queued` | `max_concurrent * 10` (i.e. `100` with defaults) | Admission ceiling. New submits over this cap raise `JobQueueFull`, which the router maps to `503` with `Retry-After: 1`. |
**New in PR #4336:** the Jobs API returns `503 Service Unavailable` with a `Retry-After` header when the executor's admission ceiling (`max_queued`, default `max_concurrent ร 10`) is hit. The stranded `QUEUED` row is deleted, so retrying the same `Idempotency-Key` is safe. `max_queued` is a `JobExecutor` constructor argument โ pass a custom executor to `create_app(executor=...)` to change it (`get_executor()` reads `PRAISONAI_MAX_CONCURRENT_JOBS` and `PRAISONAI_JOB_TIMEOUT` from the environment, but not `max_queued`).
Honour `Retry-After` on the client and reuse the same `Idempotency-Key` so the retry de-duplicates cleanly:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
import time
def submit_with_retry(prompt: str, api_url: str = "http://127.0.0.1:8005"):
while True:
r = httpx.post(
f"{api_url}/api/v1/runs",
json={"prompt": prompt},
headers={"Idempotency-Key": "task-123"},
)
if r.status_code != 503:
r.raise_for_status()
return r.json()["job_id"]
# Queue is full โ respect Retry-After before retrying the same key.
time.sleep(float(r.headers.get("Retry-After", "1")))
```
## Start Server
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
praisonai serve jobs --port 8005
```
Advanced / factory mode (multiple workers, direct uvicorn control):
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory
```
## Submit Job
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={"prompt": "Your task here"}
)
job_id = response.json()["job_id"]
```
## Submit a Recipe Job
Point the same endpoint at an installed recipe by adding `recipe_name` (and, optionally, `recipe_config`). The `prompt` field becomes the recipe's input data.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={
"prompt": "review todays merges",
"recipe_name": "nightly-reviewer",
"recipe_config": {"model": "gpt-4o-mini"},
},
)
job_id = response.json()["job_id"]
```
## Idempotency
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={"prompt": "Task"},
headers={"Idempotency-Key": "unique-key-123"}
)
```
Since [PR #1673](https://github.com/MervinPraison/PraisonAI/pull/1673), the in-process store is safe to read concurrently with writes. You can safely share a single `InMemoryJobStore` instance between the FastAPI app and background tasks that periodically read stats.
## Polling
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import time
def wait_for_completion(job_id):
while True:
status = httpx.get(f"http://127.0.0.1:8005/api/v1/runs/{job_id}").json()
if status["status"] in ("succeeded", "failed", "cancelled"):
return status
time.sleep(status.get("retry_after", 2))
```
## SSE Streaming
Multiple viewers of the same `job_id` now each receive every progress update โ one disconnecting no longer freezes the others.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
with httpx.stream("GET", f"http://127.0.0.1:8005/api/v1/runs/{job_id}/stream") as r:
for line in r.iter_lines():
if line.startswith("data:"):
print(line[5:])
```
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant TabA as Viewer A
participant TabB as Viewer B
participant API as Jobs API
participant Exec as JobExecutor
TabA->>API: GET /runs/{id}/stream
API->>Exec: register_progress_callback(id, cb_A)
TabB->>API: GET /runs/{id}/stream
API->>Exec: register_progress_callback(id, cb_B)
Exec-->>TabA: progress event
Exec-->>TabB: progress event
TabB--xAPI: disconnect
API->>Exec: unregister_progress_callback(id, cb_B)
Exec-->>TabA: progress event (A still receives)
```
**New in PR #4336:** `/api/v1/runs/{id}/stream` now supports **multiple concurrent viewers of the same job**. Every subscriber receives every progress update. If you call `JobExecutor.register_progress_callback` / `unregister_progress_callback` directly, note that `unregister_progress_callback` now takes an optional second argument: pass the specific callback to remove only that subscriber; call without a callback to clear all subscribers (the shutdown path).
When driving `JobExecutor` directly, remove only your own subscriber so co-registered viewers keep streaming:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
def make_viewer():
async def on_progress(job):
print(job.progress_percentage, job.progress_step)
return on_progress
cb = make_viewer()
executor.register_progress_callback(job_id, cb)
try:
...
finally:
executor.unregister_progress_callback(job_id, cb) # remove ONLY this subscriber
```
## Webhook Callback
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={
"prompt": "Task",
"webhook_url": "https://example.com/callback"
}
)
```
## Session Grouping
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
for task in tasks:
httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={"prompt": task, "session_id": "project-alpha"}
)
```
## Cancel Job
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
httpx.post(f"http://127.0.0.1:8005/api/v1/runs/{job_id}/cancel")
```
## List Jobs
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
jobs = httpx.get("http://127.0.0.1:8005/api/v1/runs").json()
```
## Complete Example
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
import time
API_URL = "http://127.0.0.1:8005"
def submit_and_wait(prompt):
# Submit
response = httpx.post(f"{API_URL}/api/v1/runs", json={"prompt": prompt})
job_id = response.json()["job_id"]
# Wait
while True:
status = httpx.get(f"{API_URL}/api/v1/runs/{job_id}").json()
if status["status"] == "succeeded":
return httpx.get(f"{API_URL}/api/v1/runs/{job_id}/result").json()
elif status["status"] in ("failed", "cancelled"):
raise Exception(f"Job {status['status']}")
time.sleep(2)
result = submit_and_wait("What is 2+2?")
print(result["result"])
```
## CLI Usage
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# Start the jobs server
praisonai serve jobs --port 8005
# Advanced / factory mode: python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory
# Submit a job
praisonai run submit "Analyze this data"
# Submit a recipe as job
praisonai run submit "Analyze AI trends" --recipe news-analyzer
# With recipe config
praisonai run submit "Analyze" --recipe analyzer --recipe-config '{"format": "json"}'
# Wait for completion
praisonai run submit "Quick task" --wait
# Stream progress
praisonai run submit "Long task" --stream
# Check status
praisonai run status
# Get result
praisonai run result
# List jobs
praisonai run list
# Cancel job
praisonai run cancel
```
***
## Best Practices
Pass `Idempotency-Key` (HTTP) or `idempotency_key=` (recipe helper) so duplicate submits return the same job instead of duplicating work.
For runs over a few minutes, set `webhook_url` and let your service react to completion instead of holding an open poll loop.
Use a shared `session_id` per project so list/filter endpoints stay organised in dashboards.
`praisonai serve jobs --port 8005` (or, for factory mode, `python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory`) โ the in-process store is safe for concurrent reads after PR #1673.
Export `PRAISONAI_JOBS_DB_PATH=/var/lib/praisonai/jobs.db` so `SqliteJobStore` persists jobs and idempotency keys across restarts. With `ENVIRONMENT=production` and no path set, `create_app()` raises `RuntimeError` rather than silently losing state.
Size `max_queued` to your worker capacity and retry on `503` using the `Retry-After` header with the same `Idempotency-Key`. The rejected `QUEUED` row is deleted, so the retry starts clean instead of colliding with a stranded record.
***
## Related
Run agent work in-process without a separate jobs server.
Submit, stream, and cancel jobs from the terminal.
Persist and resume tool executions across restarts.
Shape request bursts before they reach the admission ceiling.