Skip to main content
praisonai run --output stream-json emits a per-step NDJSON event stream you can pipe into CI pipelines, scripts, or observability tools.

​
Quick Start

1

Run with stream-json output

praisonai run --output stream-json "Find the weather in London"
Each line of stdout is one JSON event object. The stream ends when the run completes.
2

Filter events with jq

praisonai run --output stream-json "Find the weather in London" \
  | jq -c 'select(.event == "tool.start" or .event == "run.result")'
Use jq to watch only the events that matter to your pipeline.
3

Process events in Python

import json
import subprocess

proc = subprocess.Popen(
    ["praisonai", "run", "--output", "stream-json", "Find the weather in London"],
    stdout=subprocess.PIPE,
    text=True,
)
for line in proc.stdout:
    event = json.loads(line)
    print(event["event"], event["data"])

​
How It Works

Each run wires a StreamEventBridge between the agent’s internal StreamEventEmitter and the OutputController. Every agent action — tool calls, text deltas, errors — becomes an NDJSON line on stdout. Run-level lifecycle events (run.start, agent.message, run.result, run.error) are driven directly by the CLI. Per-step events (tool.*, text.delta, reasoning.delta) come through the StreamEventBridge callback.

​
Event Schema (schema_version = 1)

Every NDJSON line is a JSON object with event and data keys. Every data object contains schema_version: 1.
EventWhenKey fields in data
run.startOnce, at run startschema_version, target, model, framework
agent.messageOnce per agent invocationschema_version, agent (name or null)
tool.startEach tool callschema_version, tool, args
tool.resultEach tool returnschema_version, tool, result, ok
tool.errorTool-scoped failureschema_version, tool, error
text.deltaStreaming text chunkschema_version, text
reasoning.deltaStreaming reasoning chunkschema_version, text
run.resultSuccessful run endschema_version, ok: true, result
run.errorRun or transport failureschema_version, ok: false, error

​
Example NDJSON output

{"event":"run.start","data":{"schema_version":1,"target":"Find the weather in London","model":"gpt-4o-mini","framework":"praisonai"}}
{"event":"agent.message","data":{"schema_version":1,"agent":"Researcher"}}
{"event":"tool.start","data":{"schema_version":1,"tool":"web_search","args":{"query":"weather in London"}}}
{"event":"tool.result","data":{"schema_version":1,"tool":"web_search","result":"...","ok":true}}
{"event":"text.delta","data":{"schema_version":1,"text":"The weather in London is "}}
{"event":"text.delta","data":{"schema_version":1,"text":"currently 15°C..."}}
{"event":"run.result","data":{"schema_version":1,"ok":true,"result":"The weather in London is currently 15°C..."}}

​
Examples per Event Type

Emitted once at the start of every run. Contains the target prompt, model, and framework.
{
  "event": "run.start",
  "data": {
    "schema_version": 1,
    "target": "Find the weather in London",
    "model": "gpt-4o-mini",
    "framework": "praisonai"
  }
}
Emitted once per agent invocation, immediately before the agent starts processing.
{
  "event": "agent.message",
  "data": {
    "schema_version": 1,
    "agent": "Researcher"
  }
}
Emitted each time the agent calls a tool. args contains the tool’s input arguments.
{
  "event": "tool.start",
  "data": {
    "schema_version": 1,
    "tool": "web_search",
    "args": {"query": "weather in London"}
  }
}
Emitted when a tool returns. ok is true unless the tool reported an error.
{
  "event": "tool.result",
  "data": {
    "schema_version": 1,
    "tool": "web_search",
    "result": "London: 15°C, partly cloudy",
    "ok": true
  }
}
Streaming text chunks from the model. reasoning.delta is used when is_reasoning=True.
{"event":"text.delta","data":{"schema_version":1,"text":"The weather in London "}}
{"event":"text.delta","data":{"schema_version":1,"text":"is currently 15°C."}}

{"event":"reasoning.delta","data":{"schema_version":1,"text":"I should search for current conditions..."}}
Emitted once when the run completes successfully. result is the agent’s final output.
{
  "event": "run.result",
  "data": {
    "schema_version": 1,
    "ok": true,
    "result": "The weather in London is currently 15°C with partly cloudy skies."
  }
}
Emitted when a run-level, streaming, or transport failure occurs. Also emitted for core error events.
{
  "event": "run.error",
  "data": {
    "schema_version": 1,
    "ok": false,
    "error": "Rate limit exceeded"
  }
}

​
Stability and Versioning

Every event’s data object includes schema_version: 1. This version is bumped only on backward-incompatible schema changes. 
import json
import subprocess

proc = subprocess.Popen(
    ["praisonai", "run", "--output", "stream-json", "task"],
    stdout=subprocess.PIPE, text=True,
)
for line in proc.stdout:
    event = json.loads(line)
    version = event["data"].get("schema_version", 0)
    if version != 1:
        raise RuntimeError(f"Unexpected schema_version: {version}")
Pin your consumer against schema_version: 1. Future backward-incompatible changes will increment this value so you can detect and handle them gracefully.

​
When to Use Which Output Mode

​
Common Patterns

​
Filter for specific events with jq

Watch only tool calls and the final result: 
praisonai run --output stream-json "Research quantum computing" \
  | jq -c 'select(.event | test("^tool\\.|^run\\.result$"))'

​
Detect run.error in CI

Exit with a non-zero code if the run fails: 
OUTPUT=$(praisonai run --output stream-json "Run tests")
if echo "$OUTPUT" | jq -e 'select(.event == "run.error")' > /dev/null; then
  echo "Run failed"
  exit 1
fi

​
Build a progress UI fed by text.delta

Concatenate text.delta chunks to show a live streamed response: 
import json
import subprocess
import sys

proc = subprocess.Popen(
    ["praisonai", "run", "--output", "stream-json", "Explain quantum computing"],
    stdout=subprocess.PIPE,
    text=True,
)
for line in proc.stdout:
    event = json.loads(line)
    if event["event"] == "text.delta":
        sys.stdout.write(event["data"]["text"])
        sys.stdout.flush()
    elif event["event"] == "run.result":
        print()
        break

​
Best Practices

Before processing events, verify data.schema_version == 1. This guards your consumer against future breaking changes.
event = json.loads(line)
assert event["data"]["schema_version"] == 1, "Unsupported schema version"
A run.error event means the run failed. Check ok == false and surface the error field to your observability system.
if event["event"] == "run.error":
    logger.error("Agent run failed: %s", event["data"]["error"])
    sys.exit(1)
In text, json, silent, and verbose modes the bridge is a no-op — zero per-step callback overhead. Only enable stream-json when a downstream consumer actually reads the events.
Concatenating text.delta chunks is useful for live display, but run.result.data.result is the single authoritative final output string. Prefer it when you only need the end result.

CLI Run Reference

CLI Backend Protocol