Stream Run API

Stream real-time progress updates via Server-Sent Events

GET /api/v1/runs/{job_id}/stream

cURL

# Stream with curl (use -N for unbuffered output)
curl -N http://127.0.0.1:8005/api/v1/runs/run_abc123/stream

import httpx

job_id = "run_abc123"

# Stream progress updates
with httpx.stream("GET", f"http://127.0.0.1:8005/api/v1/runs/{job_id}/stream") as response:
    for line in response.iter_lines():
        if line.startswith("data:"):
            data = line[5:].strip()
            if data == "[DONE]":
                print("Stream complete")
                break
            
            import json
            event = json.loads(data)
            
            if event["event"] == "progress":
                pct = event["data"]["percentage"]
                step = event["data"].get("current_step", "")
                print(f"Progress: {pct}% - {step}")
            elif event["event"] == "log":
                print(f"Log: {event['data']['message']}")
            elif event["event"] == "complete":
                print(f"Complete: {event['data']['status']}")

import httpx
from httpx_sse import connect_sse

with httpx.Client() as client:
    with connect_sse(client, "GET", f"http://127.0.0.1:8005/api/v1/runs/{job_id}/stream") as event_source:
        for sse in event_source.iter_sse():
            if sse.data == "[DONE]":
                break
            print(f"Event: {sse.event}, Data: {sse.data}")

praisonai run stream run_abc123

# JSON output (each line is a JSON event)
praisonai run stream run_abc123 --json

text/event-stream

data: {"event": "progress", "data": {"percentage": 0, "current_step": "Initializing"}}

data: {"event": "log", "data": {"message": "Agent started processing"}}

data: {"event": "progress", "data": {"percentage": 25, "current_step": "Analyzing input"}}

data: {"event": "progress", "data": {"percentage": 50, "current_step": "Processing data"}}

data: {"event": "log", "data": {"message": "Data analysis complete"}}

data: {"event": "progress", "data": {"percentage": 75, "current_step": "Generating response"}}

data: {"event": "progress", "data": {"percentage": 100, "current_step": "Complete"}}

data: {"event": "complete", "data": {"status": "succeeded", "job_id": "run_abc123"}}

data: [DONE]

Stream real-time progress updates for an async job using Server-Sent Events (SSE).

This endpoint returns a streaming response. The interactive playground is disabled for this endpoint. Use the examples below to test streaming.

job_id · string · required

The unique job identifier (e.g., run_abc123).

Response

The response is a text/event-stream with the following event types:

Event Types

Event	Description
progress	Progress update with percentage and current step
log	Log message from agent execution
complete	Job completed (succeeded, failed, or cancelled)
[DONE]	Stream ended

Event Format

data: {"event": "progress", "data": {"percentage": 50, "current_step": "Processing"}}

data: {"event": "log", "data": {"message": "Agent started task"}}

data: {"event": "complete", "data": {"status": "succeeded", "job_id": "run_abc123"}}

data: [DONE]

cURL

# Stream with curl (use -N for unbuffered output)
curl -N http://127.0.0.1:8005/api/v1/runs/run_abc123/stream

import httpx

job_id = "run_abc123"

# Stream progress updates
with httpx.stream("GET", f"http://127.0.0.1:8005/api/v1/runs/{job_id}/stream") as response:
    for line in response.iter_lines():
        if line.startswith("data:"):
            data = line[5:].strip()
            if data == "[DONE]":
                print("Stream complete")
                break
            
            import json
            event = json.loads(data)
            
            if event["event"] == "progress":
                pct = event["data"]["percentage"]
                step = event["data"].get("current_step", "")
                print(f"Progress: {pct}% - {step}")
            elif event["event"] == "log":
                print(f"Log: {event['data']['message']}")
            elif event["event"] == "complete":
                print(f"Complete: {event['data']['status']}")

import httpx
from httpx_sse import connect_sse

with httpx.Client() as client:
    with connect_sse(client, "GET", f"http://127.0.0.1:8005/api/v1/runs/{job_id}/stream") as event_source:
        for sse in event_source.iter_sse():
            if sse.data == "[DONE]":
                break
            print(f"Event: {sse.event}, Data: {sse.data}")

praisonai run stream run_abc123

# JSON output (each line is a JSON event)
praisonai run stream run_abc123 --json

text/event-stream

data: {"event": "progress", "data": {"percentage": 0, "current_step": "Initializing"}}

data: {"event": "log", "data": {"message": "Agent started processing"}}

data: {"event": "progress", "data": {"percentage": 25, "current_step": "Analyzing input"}}

data: {"event": "progress", "data": {"percentage": 50, "current_step": "Processing data"}}

data: {"event": "log", "data": {"message": "Data analysis complete"}}

data: {"event": "progress", "data": {"percentage": 75, "current_step": "Generating response"}}

data: {"event": "progress", "data": {"percentage": 100, "current_step": "Complete"}}

data: {"event": "complete", "data": {"status": "succeeded", "job_id": "run_abc123"}}

data: [DONE]

Complete Streaming Example

import httpx
import json

API_URL = "http://127.0.0.1:8005"

def submit_and_stream(prompt):
    # Submit job
    response = httpx.post(f"{API_URL}/api/v1/runs", json={"prompt": prompt})
    job = response.json()
    job_id = job["job_id"]
    print(f"Submitted job: {job_id}")
    
    # Stream progress
    final_status = None
    with httpx.stream("GET", f"{API_URL}/api/v1/runs/{job_id}/stream") as response:
        for line in response.iter_lines():
            if not line.startswith("data:"):
                continue
            
            data = line[5:].strip()
            if data == "[DONE]":
                break
            
            event = json.loads(data)
            
            if event["event"] == "progress":
                pct = event["data"]["percentage"]
                step = event["data"].get("current_step", "")
                print(f"\rProgress: {pct:5.1f}% - {step}", end="", flush=True)
            elif event["event"] == "complete":
                final_status = event["data"]["status"]
                print(f"\nJob {final_status}")
    
    # Get result if succeeded
    if final_status == "succeeded":
        result = httpx.get(f"{API_URL}/api/v1/runs/{job_id}/result").json()
        return result["result"]
    else:
        raise Exception(f"Job {final_status}")

result = submit_and_stream("Analyze quarterly sales data")
print(f"Result: {result}")

Error Responses

Status	Description
404	Job not found
400	Job already completed (no stream available)
500	Internal server error

Notes

• The stream automatically closes when the job completes
• If the connection is interrupted, you can reconnect and continue streaming
• For completed jobs, use Get Run Status instead

See Also

• Get Run Status
• Get Run Result
• Async Jobs Overview