Skip to main content

​
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

MethodDescription
job_idUnique job identifier
recipe_nameName of the recipe being executed
statusCurrent 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 VariableDefaultDescription
PRAISONAI_JOBS_PORT8005Server port
PRAISONAI_JOBS_HOST127.0.0.1Server host
PRAISONAI_JOBS_MAX_CONCURRENT10Max concurrent jobs
PRAISONAI_JOBS_DEFAULT_TIMEOUT3600Default timeout (seconds)

​
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

ScopeDescription
noneNo idempotency (default)
sessionUnique within session
globalUnique 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