Stream Run API

# 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

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]

GET

api

runs

{job_id}

stream

# 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

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.

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]

# 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

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

Deploy

Response

Event Types

Event Format

Complete Streaming Example

Error Responses

Notes

See Also

​Response

​Event Types

​Event Format

​Complete Streaming Example

​Error Responses

​Notes

​See Also

Response

Event Types

Event Format

Complete Streaming Example

Error Responses

Notes

See Also