> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Async Jobs
> Submit and manage long-running agent jobs and recipes via HTTP API
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent
import asyncio
agent = Agent(name="job-agent", instructions="Run background jobs asynchronously.")
async def main():
job = await agent.astart("Start a long-running background job.")
print(job)
asyncio.run(main())
```
Submit long-running agent tasks and recipes, then retrieve results asynchronously via a jobs server.
The user submits a long job; the agent runs asynchronously and returns when the job completes.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
Client[📤 Submit] --> API[🛰️ Jobs API]
API --> Queue[(📋 Job store)]
Queue --> Worker[🤖 Agent run]
Worker --> Result[✅ Result / webhook]
Client -->|poll or SSE| Queue
classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff
classDef ok fill:#10B981,stroke:#7C90A0,color:#fff
classDef input fill:#6366F1,stroke:#7C90A0,color:#fff
class Worker agent
class API,Queue tool
class Result ok
class Client input
```
## How It Works
The user submits a job, the server runs it in the background, and the result comes back on completion.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant JobsAPI
participant Agent
User->>JobsAPI: Submit job
JobsAPI-->>User: job_id
JobsAPI->>Agent: Run in background
Agent-->>JobsAPI: Result
User->>JobsAPI: Poll / stream / webhook
JobsAPI-->>User: Completed result
```
## Choose a Result Mode
Pick how the result is delivered once the job finishes.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Q{How long is the job?} -->|Seconds| P[Poll status]
Q -->|Minutes, watch progress| S[SSE stream]
Q -->|Long, react later| W[Webhook callback]
classDef config fill:#6366F1,stroke:#7C90A0,color:#fff
classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff
class Q config
class P,S,W tool
```
## Quick Start
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai import recipe
job = recipe.submit_job(
"my-recipe",
input={"query": "What is AI?"},
config={"max_tokens": 1000},
session_id="session_123",
timeout_sec=3600,
api_url="http://127.0.0.1:8005",
)
print(f"Job ID: {job.job_id}")
result = job.wait(poll_interval=5, timeout=300)
print(f"Result: {result}")
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
API_URL = "http://127.0.0.1:8005"
response = httpx.post(f"{API_URL}/api/v1/runs", json={"prompt": "Analyze data"})
job_id = response.json()["job_id"]
status = httpx.get(f"{API_URL}/api/v1/runs/{job_id}").json()
result = httpx.get(f"{API_URL}/api/v1/runs/{job_id}/result").json()
```
## Start Server
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory
```
## Submit Job
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={"prompt": "Your task here"}
)
job_id = response.json()["job_id"]
```
## Idempotency
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={"prompt": "Task"},
headers={"Idempotency-Key": "unique-key-123"}
)
```
Since [PR #1673](https://github.com/MervinPraison/PraisonAI/pull/1673), the in-process store is safe to read concurrently with writes. You can safely share a single `InMemoryJobStore` instance between the FastAPI app and background tasks that periodically read stats.
## Polling
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import time
def wait_for_completion(job_id):
while True:
status = httpx.get(f"http://127.0.0.1:8005/api/v1/runs/{job_id}").json()
if status["status"] in ("succeeded", "failed", "cancelled"):
return status
time.sleep(status.get("retry_after", 2))
```
## SSE Streaming
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
with httpx.stream("GET", f"http://127.0.0.1:8005/api/v1/runs/{job_id}/stream") as r:
for line in r.iter_lines():
if line.startswith("data:"):
print(line[5:])
```
## Webhook Callback
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
response = httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={
"prompt": "Task",
"webhook_url": "https://example.com/callback"
}
)
```
## Session Grouping
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
for task in tasks:
httpx.post(
"http://127.0.0.1:8005/api/v1/runs",
json={"prompt": task, "session_id": "project-alpha"}
)
```
## Cancel Job
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
httpx.post(f"http://127.0.0.1:8005/api/v1/runs/{job_id}/cancel")
```
## List Jobs
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
jobs = httpx.get("http://127.0.0.1:8005/api/v1/runs").json()
```
## Complete Example
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
import httpx
import time
API_URL = "http://127.0.0.1:8005"
def submit_and_wait(prompt):
# Submit
response = httpx.post(f"{API_URL}/api/v1/runs", json={"prompt": prompt})
job_id = response.json()["job_id"]
# Wait
while True:
status = httpx.get(f"{API_URL}/api/v1/runs/{job_id}").json()
if status["status"] == "succeeded":
return httpx.get(f"{API_URL}/api/v1/runs/{job_id}/result").json()
elif status["status"] in ("failed", "cancelled"):
raise Exception(f"Job {status['status']}")
time.sleep(2)
result = submit_and_wait("What is 2+2?")
print(result["result"])
```
## CLI Usage
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# Start the jobs server
python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory
# Submit a job
praisonai run submit "Analyze this data"
# Submit a recipe as job
praisonai run submit "Analyze AI trends" --recipe news-analyzer
# With recipe config
praisonai run submit "Analyze" --recipe analyzer --recipe-config '{"format": "json"}'
# Wait for completion
praisonai run submit "Quick task" --wait
# Stream progress
praisonai run submit "Long task" --stream
# Check status
praisonai run status
# Get result
praisonai run result
# List jobs
praisonai run list
# Cancel job
praisonai run cancel
```
***
## Best Practices
Pass `Idempotency-Key` (HTTP) or `idempotency_key=` (recipe helper) so duplicate submits return the same job instead of duplicating work.
For runs over a few minutes, set `webhook_url` and let your service react to completion instead of holding an open poll loop.
Use a shared `session_id` per project so list/filter endpoints stay organised in dashboards.
`python -m uvicorn praisonai.jobs.server:create_app --port 8005 --factory` — the in-process store is safe for concurrent reads after PR #1673.
***
## Related
Run agent work in-process without a separate jobs server.
Submit, stream, and cancel jobs from the terminal.