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

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

Method	Description
`job_id`	Unique job identifier
`recipe_name`	Name of the recipe being executed
`status`	Current job status
`client`	Lazily-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

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

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:

from praisonai import recipe

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

A 10-minute wait(poll_interval=5) uses one keep-alive HTTP connection for all ~120 polls, not one fresh TCP+TLS handshake per poll. This applies to get_status(), get_result(), cancel(), and every iteration of wait().

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.

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

# 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 Variable	Default	Description
`PRAISONAI_JOBS_PORT`	8005	Server port
`PRAISONAI_JOBS_HOST`	127.0.0.1	Server host
`PRAISONAI_JOBS_MAX_CONCURRENT`	10	Max concurrent jobs
`PRAISONAI_JOBS_DEFAULT_TIMEOUT`	3600	Default 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_HOST`	127.0.0.1	Bind 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:

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

Webhook payload:

{
  "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:

# 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

Scope	Description
`none`	No idempotency (default)
`session`	Unique within session
`global`	Unique across all sessions

TEMPLATE.yaml Runtime Block

Configure job defaults in your recipe:

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

Error Handling

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

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

Guide

Reference

Async Jobs

Async Jobs

Python API

Submitting a Job

JobHandle Methods

Waiting for Completion

Deterministic cleanup with `with`

Using the Jobs API Directly

Starting the Jobs Server

Server Configuration

Job Status Values

Webhooks

Idempotency

Idempotency Scopes

TEMPLATE.yaml Runtime Block

Error Handling

Best Practices

See Also

​Async Jobs

​Python API

​Submitting a Job

​JobHandle Methods

​Waiting for Completion

​Deterministic cleanup with with

​Using the Jobs API Directly

​Starting the Jobs Server

​Server Configuration

​Job Status Values

​Webhooks

​Idempotency

​Idempotency Scopes

​TEMPLATE.yaml Runtime Block

​Error Handling

​Best Practices

​See Also

Async Jobs

Python API

Submitting a Job

JobHandle Methods

Waiting for Completion

Deterministic cleanup with `with`

Using the Jobs API Directly

Starting the Jobs Server

Server Configuration

Job Status Values

Webhooks

Idempotency

Idempotency Scopes

TEMPLATE.yaml Runtime Block

Error Handling

Best Practices

See Also