Praison AI

Async Jobs

Server-based asynchronous job execution with persistence and webhooks

Async Jobs

Async Jobs provide a server-based approach to running recipes and agents. Jobs are submitted to an API server, persisted, and can be monitored, streamed, and cancelled. Ideal for production deployments with multiple clients.

Python API

Submitting a Job

python
from praisonai import recipe

# Submit recipe as async job
job = recipe.submit_job(
    "my-recipe",
    input={"query": "What is AI?"},
    config={"max_tokens": 1000},
    session_id="session_123",
    timeout_sec=3600,
    webhook_url="https://example.com/webhook",
    idempotency_key="unique-key-123",
    api_url="http://127.0.0.1:8005",
)

print(f"Job ID: {job.job_id}")
print(f"Status: {job.status}")

JobHandle Methods

MethodDescription
job_idUnique job identifier
recipe_nameName of the recipe being executed
statusCurrent job status
clientLazily-constructed shared httpx.Client bound to api_url — reused by every get_status() / get_result() / cancel() / wait() poll.
get_status()Fetch current status from API
get_result()Fetch job result from API
cancel()Cancel the job
wait(poll_interval, timeout)Wait for completion (reuses one keep-alive connection across all polls)
close()Close the shared HTTP client and release its connection pool
__enter__ / __exit__Context-manager protocol — calls close() on scope exit

Waiting for Completion

python
# Submit and wait
job = recipe.submit_job("my-recipe", input="test", wait=True)

# Or wait after submission
job = recipe.submit_job("my-recipe", input="test")
result = job.wait(poll_interval=5, timeout=300)
print(f"Result: {result}")

Deterministic cleanup with with

JobHandle is a long-lived handle: one wait() may poll the jobs API dozens of times. It holds a lazily-constructed shared httpx.Client, so those polls share one keep-alive connection instead of doing a fresh TLS handshake per call. Use the handle as a context manager — or call close() — to release the connection pool deterministically:

python
from praisonai import recipe

with recipe.submit_job("my-recipe", input="test") as job:
    result = job.wait(poll_interval=5, timeout=300)
    print(result)
# `job.client` is closed here — pooled sockets are released.

Not using with? Call close() explicitly:

python
from praisonai import recipe

job = recipe.submit_job("my-recipe", input="test")
try:
    result = job.wait(poll_interval=5, timeout=300)
finally:
    job.close()

Using the Jobs API Directly

recipe_name and recipe_config are honoured by the router as of PR #4085; earlier versions accepted the fields but did not forward them to the executor.

If your code creates its own client to hit the Jobs API directly (rather than using submit_job), keep the client alive across calls — the wrapper’s own JobHandle does exactly this.

python
from praisonai.jobs import Job, JobSubmitRequest, JobExecutor, create_app

# Create job request
request = JobSubmitRequest(
    prompt="Analyze this data",
    recipe_name="data-analyzer",
    recipe_config={"format": "json"},
    timeout=3600,
    webhook_url="https://example.com/callback",
    session_id="session_123",
)

# Submit via API client
import httpx

with httpx.Client() as client:
    response = client.post(
        "http://127.0.0.1:8005/api/v1/runs",
        json=request.model_dump(),
    )
    job_data = response.json()
    print(f"Job ID: {job_data['job_id']}")

Starting the Jobs Server

bash
# Using uvicorn directly
python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory

# Or using the CLI
praisonai jobs server --port 8005

Server Configuration

Environment VariableDefaultDescription
PRAISONAI_JOBS_PORT8005Server port
PRAISONAI_JOBS_HOST127.0.0.1Server host
PRAISONAI_JOBS_MAX_CONCURRENT10Max concurrent jobs
PRAISONAI_JOBS_DEFAULT_TIMEOUT3600Default timeout (seconds)
PRAISONAI_JOBS_API_KEY(unset)When set, all requests except /health must send the key via Authorization: Bearer <key> or X-API-Key: <key>.
PRAISONAI_JOBS_BIND_HOST127.0.0.1Bind host for the auth guard. Any value outside {127.0.0.1, localhost, ::1} requires PRAISONAI_JOBS_API_KEY.

Job Status Values

  • queued – Job is waiting to be processed
  • running – Job is currently executing
  • succeeded – Job completed successfully
  • failed – Job failed with an error
  • cancelled – Job was cancelled

Webhooks

Configure webhooks to receive notifications when jobs complete:

python
job = recipe.submit_job(
    "my-recipe",
    input="test",
    webhook_url="https://example.com/webhook",
)

Webhook payload:

json
{
  "job_id": "run_abc123",
  "status": "succeeded",
  "result": {"output": "..."},
  "completed_at": "2024-01-15T10:30:00Z",
  "duration_seconds": 45.2
}

Idempotency

Prevent duplicate job submissions with idempotency keys:

python
# Same key = same job (no duplicate)
job1 = recipe.submit_job("my-recipe", idempotency_key="order-123")
job2 = recipe.submit_job("my-recipe", idempotency_key="order-123")

assert job1.job_id == job2.job_id  # Same job returned

Idempotency Scopes

ScopeDescription
noneNo idempotency (default)
sessionUnique within session
globalUnique across all sessions

TEMPLATE.yaml Runtime Block

Configure job defaults in your recipe:

yaml
runtime:
  job:
    enabled: true
    timeout_sec: 3600
    webhook_url: "${WEBHOOK_URL}"
    idempotency_scope: "session"
    events:
      - completed
      - failed

Error Handling

python
from praisonai import recipe

try:
    job = recipe.submit_job("my-recipe", input="test")
    result = job.wait(timeout=60)

    if result.get("status") == "failed":
        print(f"Job failed: {result.get('error')}")
    else:
        print(f"Result: {result}")

except Exception as e:
    print(f"Error: {e}")

Best Practices

  1. Use idempotency keys – Prevent duplicate submissions
  2. Set appropriate timeouts – Match job complexity
  3. Configure webhooks – For async notification
  4. Monitor job status – Use streaming for real-time updates
  5. Handle failures gracefully – Implement retry logic

See Also