Stats API - PraisonAI

curl http://127.0.0.1:8005/stats

import httpx

response = httpx.get("http://127.0.0.1:8005/stats")
stats = response.json()

print(f"Total jobs: {stats['total_jobs']}")
print(f"Running: {stats['by_status'].get('running', 0)}")
print(f"Succeeded: {stats['by_status'].get('succeeded', 0)}")
print(f"Failed: {stats['by_status'].get('failed', 0)}")

curl http://127.0.0.1:8005/stats

{
  "total_jobs": 150,
  "by_status": {
    "queued": 5,
    "running": 3,
    "succeeded": 130,
    "failed": 10,
    "cancelled": 2
  }
}

GET

stats

curl http://127.0.0.1:8005/stats

import httpx

response = httpx.get("http://127.0.0.1:8005/stats")
stats = response.json()

print(f"Total jobs: {stats['total_jobs']}")
print(f"Running: {stats['by_status'].get('running', 0)}")
print(f"Succeeded: {stats['by_status'].get('succeeded', 0)}")
print(f"Failed: {stats['by_status'].get('failed', 0)}")

curl http://127.0.0.1:8005/stats

{
  "total_jobs": 150,
  "by_status": {
    "queued": 5,
    "running": 3,
    "succeeded": 130,
    "failed": 10,
    "cancelled": 2
  }
}

Get aggregate statistics about all async jobs.

Response

total_jobs

integer

required

Total number of jobs ever submitted.

by_status

object

required

Breakdown of jobs by status.

by_status.queued

integer

Number of jobs currently queued.

by_status.running

integer

Number of jobs currently running.

by_status.succeeded

integer

Number of jobs that completed successfully.

by_status.failed

integer

Number of jobs that failed.

by_status.cancelled

integer

Number of jobs that were cancelled.

curl http://127.0.0.1:8005/stats

import httpx

response = httpx.get("http://127.0.0.1:8005/stats")
stats = response.json()

print(f"Total jobs: {stats['total_jobs']}")
print(f"Running: {stats['by_status'].get('running', 0)}")
print(f"Succeeded: {stats['by_status'].get('succeeded', 0)}")
print(f"Failed: {stats['by_status'].get('failed', 0)}")

curl http://127.0.0.1:8005/stats

{
  "total_jobs": 150,
  "by_status": {
    "queued": 5,
    "running": 3,
    "succeeded": 130,
    "failed": 10,
    "cancelled": 2
  }
}

Usage

Use this endpoint for:

Monitoring dashboards
Capacity planning
Alerting on failure rates

Example: Monitor Failure Rate

import httpx
import time

API_URL = "http://127.0.0.1:8005"
FAILURE_THRESHOLD = 0.1  # 10% failure rate

def check_failure_rate():
    stats = httpx.get(f"{API_URL}/stats").json()
    
    total = stats["total_jobs"]
    failed = stats["by_status"].get("failed", 0)
    
    if total > 0:
        failure_rate = failed / total
        if failure_rate > FAILURE_THRESHOLD:
            print(f"WARNING: High failure rate: {failure_rate:.1%}")
            return False
    
    print(f"OK: {total} jobs, {failed} failed ({failed/max(total,1):.1%})")
    return True

# Periodic monitoring
while True:
    check_failure_rate()
    time.sleep(60)

​Response

​Usage

​Example: Monitor Failure Rate

​See Also

Response

Usage

Example: Monitor Failure Rate

See Also