Skip to main content
Submit long-running agent tasks and retrieve results asynchronously.

​
Quick Start

import httpx

API_URL = "http://127.0.0.1:8005"

# Submit job
response = httpx.post(f"{API_URL}/api/v1/runs", json={"prompt": "Analyze data"})
job = response.json()
job_id = job["job_id"]

# Check status
status = httpx.get(f"{API_URL}/api/v1/runs/{job_id}").json()

# Get result
result = httpx.get(f"{API_URL}/api/v1/runs/{job_id}/result").json()

​
Start Server

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

​
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

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

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