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

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

Using the Jobs API Directly

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