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

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

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

• Async Jobs CLI - CLI commands for jobs
• Background Tasks - In-process background execution
• Scheduler - Periodic job scheduling