Async Jobs

Submit and manage long-running agent jobs and recipes via HTTP API

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

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

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

Which store should I use?

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 rows still in QUEUED or RUNNING as FAILED (error="Interrupted by service restart", completed_at=now) — but only when the process that wrote the row is gone.

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. Every write stamps the row with an owner id (<pid>:<proc_start>:<token>), so a second worker sharing the same jobs.db leaves a live cohabiting worker’s jobs alone.

A row is reconciled when its owner is:

• missing (a legacy row written before the owner column existed),
• a pid that is no longer running, or
• the same pid with a different process start time (the pid was reused, e.g. after a container 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.

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
)

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

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

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

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.

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.

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.

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

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

Related