> ## 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
PraisonAI validates every webhook URL at submit time **and again immediately before delivery**, and pins the delivery connection to the exact IP it just screened.
```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"
}
)
```
### What gets rejected
A `webhook_url` is refused whenever the target โ or any address its hostname resolves to (A **or** AAAA) โ points at a non-public destination.
| Rejected because | Example |
| - | - |
| Non-http(s) scheme | `ftp://example.com/hook` |
| Localhost / loopback literals | `http://localhost/x`, `http://127.0.0.1/x`, `http://[::1]/x`, `http://0.0.0.0/x` |
| Private / RFC1918 | `http://10.0.0.5/x`, `http://192.168.1.1/x`, `http://172.16.0.1/x` |
| Link-local (incl. cloud metadata) | `http://169.254.169.254/latest/...` |
| Multicast / reserved / unspecified | `http://224.0.0.1/x`, `http://[fe80::1]/x` |
| Hostname resolves to any of the above (A **or** AAAA) | `attacker.example` whose AAAA is `::1` |
### Why it's re-validated at delivery
Jobs can run for minutes or hours, so a hostname that looked public at submit time could be flipped to an internal address before the result is sent (DNS rebinding). PraisonAI re-resolves and re-screens the target just before delivery, then connects to that exact screened IP.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Attacker as Attacker DNS
participant Jobs as Jobs API
participant Exec as JobExecutor
participant Target as Attacker webhook
Note over Jobs: submit_time
Jobs->>Attacker: Resolve attacker.example
Attacker-->>Jobs: 1.2.3.4 (public โ passes)
Jobs-->>Target: 202 Accepted (job queued)
Note over Jobs,Exec: job runs for minutes/hours
Note over Exec: delivery_time
Exec->>Attacker: Resolve attacker.example AGAIN
Attacker-->>Exec: 169.254.169.254 (rebind attempt)
Exec--xExec: rejected (link-local)
Note over Exec: no request sent โ job.result NOT exfiltrated
```
### What you see when a redirect is refused
A `3xx` response is never chased past the screened target โ this is deliberate. Instead of following the redirect, the executor logs (so you can grep for it) and does not deliver the result:
```
Webhook for {id} returned redirect {status}; not followed, result not delivered
```
## 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.
Webhook URLs are re-validated at delivery time and the connection is pinned to the screened IP. Point `webhook_url` at a **public** hostname reachable from the Jobs API host; internal/private/loopback targets and hostnames that rebind to them are rejected, and `3xx` redirects are not followed.
***
## 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.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.