Run Stream Events
Follow a praisonai run live as a versioned NDJSON event stream
from praisonaiagents import Agent
agent = Agent(name="stream-agent", instructions="Stream agent run events in real time.")
agent.start("Process this request and stream each step as it completes.")praisonai run --output stream-json emits a per-step NDJSON event stream for CI pipelines, scripts, and observability tools. This works for single-agent runs and YAML / AgentTeam runs — team member events are fanned in on a single stream and tagged with the emitting agent’s agent_id.
praisonai run --output stream-json "Find the weather in London"The user runs the CLI with stream-json output; each agent step emits a versioned NDJSON line for scripts and CI.
Quick Start
- Run with stream-json output
bash 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.
- Filter events with jq
bash praisonai run --output stream-json "Find the weather in London" \ | jq -c 'select(.event == "tool.start" or .event == "run.result")'Use
jqto watch only the events that matter to your pipeline. - Process events in Python
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.
On YAML / AgentTeam runs, the bridge attaches to team.stream_emitter — a lazy aggregate StreamEventEmitter that fans in every member agent’s events and tags each with the emitting agent_id.
YAML / Multi-Agent Runs
YAML and AgentTeam runs emit the same NDJSON contract as single-agent runs, with an extra agent_id field on per-step events.
praisonai run agents.yaml --output stream-jsonEvery team member’s events fan in on one stream, and each per-step event carries an agent_id so consumers can attribute activity to a specific member.
from praisonaiagents import Agent, Task, PraisonAIAgents
researcher = Agent(name="Researcher", instructions="Research the topic.")
writer = Agent(name="Writer", instructions="Summarise the findings.")
team = PraisonAIAgents(
agents=[researcher, writer],
tasks=[
Task(description="Find recent news on quantum computing", agent=researcher),
Task(description="Write a 3-sentence summary", agent=writer),
],
)
# The team's aggregate stream_emitter fans in per-agent events.
# The CLI `--output stream-json` attaches to this automatically.
team.start()Each member agent forwards its per-step events onto team.stream_emitter, which the bridge turns into NDJSON on stdout.
Example NDJSON for a team run
{"event":"run.start","data":{"schema_version":1,"target":"agents.yaml","model":"gpt-4o-mini","framework":"praisonai"}}
{"event":"agent.message","data":{"schema_version":1,"agent":"Researcher"}}
{"event":"tool.start","data":{"schema_version":1,"agent_id":"Researcher","tool":"web_search","args":{"query":"quantum computing news"}}}
{"event":"tool.result","data":{"schema_version":1,"agent_id":"Researcher","tool":"web_search","result":"...","ok":true}}
{"event":"text.delta","data":{"schema_version":1,"agent_id":"Writer","text":"Recent breakthroughs include "}}
{"event":"text.delta","data":{"schema_version":1,"agent_id":"Writer","text":"error-corrected qubits..."}}
{"event":"run.result","data":{"schema_version":1,"ok":true,"result":"Recent breakthroughs include error-corrected qubits..."}}Group events by agent
Filter one member’s events with jq:
praisonai run agents.yaml --output stream-json \
| jq -c 'select(.data.agent_id == "Researcher")'Route deltas to per-agent buffers in Python:
import json, subprocess, collections
buffers = collections.defaultdict(list)
proc = subprocess.Popen(
["praisonai", "run", "agents.yaml", "--output", "stream-json"],
stdout=subprocess.PIPE, text=True,
)
for line in proc.stdout:
ev = json.loads(line)
if ev["event"] == "text.delta":
buffers[ev["data"].get("agent_id", "?")].append(ev["data"]["text"])
for agent_id, chunks in buffers.items():
print(f"--- {agent_id} ---")
print("".join(chunks))Event Schema (schema_version = 1)
Every NDJSON line is a JSON object with event and data keys. Every data object contains schema_version: 1.
| Event | When | Key fields in data |
|---|---|---|
run.start | Once, at run start | schema_version, target, model, framework |
agent.message | Once per agent invocation | schema_version, agent (name or null) |
tool.start | Each tool call | schema_version, tool, args |
tool.result | Each tool return | schema_version, tool, result, ok |
tool.error | Tool-scoped failure | schema_version, tool, error |
text.delta | Streaming text chunk | schema_version, text |
reasoning.delta | Streaming reasoning chunk | schema_version, text |
run.retry | Before a retry wait, on rate-limit / transient-error backoff | schema_version, attempt, max_attempts, delay, reason |
run.result | Successful run end | schema_version, ok: true, result |
run.error | Run or transport failure | schema_version, ok: false, error |
Optional fields
| Field | Type | Present on | Description | |
|---|---|---|---|---|
agent_id | str \ | null | tool.*, text.delta, reasoning.delta, run.retry | Emitting team member’s agent_id (falls back to display_name). Present on YAML/team runs; absent (or null) on single-agent runs. |
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":"run.retry","data":{"schema_version":1,"attempt":2,"max_attempts":5,"delay":4.0,"reason":"rate_limit"}}
{"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
run.start
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"
}
}agent.message
Emitted once per agent invocation, immediately before the agent starts processing.
{
"event": "agent.message",
"data": {
"schema_version": 1,
"agent": "Researcher"
}
}tool.start
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"}
}
}tool.result
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
}
}text.delta and reasoning.delta
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..."}}reasoning.delta events now appear reliably for async agent invocations that hit the OpenAI Responses API. Previously they surfaced only for sync invocations, so long-running async streams looked silent during the thinking phase.
run.retry
Emitted before the retry wait, so consumers can render a live countdown instead of a silent-looking hang. Fires on rate-limit / transient-error backoff.
{
"event": "run.retry",
"data": {
"schema_version": 1,
"attempt": 2,
"max_attempts": 5,
"delay": 4.0,
"reason": "rate_limit"
}
}| Field | Type | Description |
|---|---|---|
attempt | int | The retry attempt about to run |
max_attempts | int | Total attempts allowed |
delay | float | Seconds the run will wait before retrying |
reason | str | Why the retry fired (e.g. rate_limit) |
This is the CLI-facing form of the RETRY stream event — same signal, surfaced as NDJSON.
run.result
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."
}
}run.error
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"
}
}On YAML / AgentTeam runs, run.error is now emitted when team.astart() raises. This lets stream-json consumers distinguish a failed team run from an incomplete / still-running one. The original exception is always re-raised — the run.error emit is best-effort and does not mask it.
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}")The agent_id field on per-step events is a new optional field — a forward-compatible addition that does not change schema_version. Consumers can ignore it and keep working on both single-agent and team runs.
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
fiBuild 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()
breakBest Practices
Always check schema_version
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"Handle run.error explicitly
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)Use stream-json only when you need per-step trace
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.
Use run.result as the authoritative final output
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.