Async Jobs - PraisonAI

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.

How It Works

The user submits a job, the server runs it in the background, and the result comes back on completion.

Choose a Result Mode

Pick how the result is delivered once the job finishes.

Quick Start

Submit via recipe helper

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

Submit via HTTP API

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.

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:

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?

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.

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.

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. 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:

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:

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

praisonai serve jobs --port 8005

Advanced / factory mode (multiple workers, direct uvicorn control):

python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory

Submit Job

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.

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

response = httpx.post(
    "http://127.0.0.1:8005/api/v1/runs",
    json={"prompt": "Task"},
    headers={"Idempotency-Key": "unique-key-123"}
)

Since PR #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

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.

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:])

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:

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

response = httpx.post(
    "http://127.0.0.1:8005/api/v1/runs",
    json={
        "prompt": "Task",
        "webhook_url": "https://example.com/callback"
    }
)

Session Grouping

for task in tasks:
    httpx.post(
        "http://127.0.0.1:8005/api/v1/runs",
        json={"prompt": task, "session_id": "project-alpha"}
    )

Cancel Job

httpx.post(f"http://127.0.0.1:8005/api/v1/runs/{job_id}/cancel")

List Jobs

jobs = httpx.get("http://127.0.0.1:8005/api/v1/runs").json()

Complete Example

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

# 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 <job_id>

# Get result
praisonai run result <job_id>

# List jobs
praisonai run list

# Cancel job
praisonai run cancel <job_id>

Best Practices

Use idempotency keys for retries

Pass Idempotency-Key (HTTP) or idempotency_key= (recipe helper) so duplicate submits return the same job instead of duplicating work.

Prefer webhooks for long jobs

For runs over a few minutes, set webhook_url and let your service react to completion instead of holding an open poll loop.

Start the jobs server before integration tests

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.

Set PRAISONAI_JOBS_DB_PATH in production

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.

Handle 503 backpressure on the client

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.

Background Tasks

Run agent work in-process without a separate jobs server.

Async Jobs CLI

Submit, stream, and cancel jobs from the terminal.

Durable Tool Runs

Persist and resume tool executions across restarts.

Rate Limiter

Shape request bursts before they reach the admission ceiling.

​How It Works

​Choose a Result Mode

​Quick Start

​Persistent Store

​Which store should I use?

​Atomic Idempotency

​Restart Recovery

​Backpressure & Admission Control

​Start Server

​Submit Job

​Submit a Recipe Job

​Idempotency

​Polling

​SSE Streaming

​Webhook Callback

​Session Grouping

​Cancel Job

​List Jobs

​Complete Example

​CLI Usage

​Best Practices

​Related

Background Tasks

Async Jobs CLI

Durable Tool Runs

Rate Limiter

How It Works

Choose a Result Mode

Quick Start

Persistent Store

Which store should I use?

Atomic Idempotency

Restart Recovery

Backpressure & Admission Control

Start Server

Submit Job

Submit a Recipe Job

Idempotency

Polling

SSE Streaming

Webhook Callback

Session Grouping

Cancel Job

List Jobs

Complete Example

CLI Usage

Best Practices

Related