Run Stream Events

Follow a praisonai run live as a versioned NDJSON event stream

python
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.

bash
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.

Diagram

praisonai run –output stream-json

🤖 Agent

🔗 StreamEventBridge

📤 OutputController

📜 NDJSON stdout

🧰 CI / jq / script

Quick Start

  1. 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.

  2. 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 jq to watch only the events that matter to your pipeline.

  3. 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.

Diagram
NDJSON stdoutAgentStreamEventBridgepraisonai runUserNDJSON stdoutAgentStreamEventBridgepraisonai runUserpraisonai run –output stream-json “…”run.start (schema_version: 1)attach_bridge(agent, output)agent.messageagent.start(prompt)tool_call_starttool.starttool_call_resulttool.resultdelta_texttext.deltaresultdetach_bridgerun.result (ok: true)

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.

bash
praisonai run agents.yaml --output stream-json

Every 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.

python
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()
Diagram

AgentTeam Fan-In

events

events

🤖 Researcher

📡 team.stream_emitter

🤖 Writer

🔗 StreamEventBridge

📜 NDJSON stdout

Each member agent forwards its per-step events onto team.stream_emitter, which the bridge turns into NDJSON on stdout.

Diagram
NDJSON stdoutWriterResearcherteam.stream_emitterStreamEventBridgepraisonai run agents.yamlUserNDJSON stdoutWriterResearcherteam.stream_emitterStreamEventBridgepraisonai run agents.yamlUser–output stream-json_attach_stream_bridge(team)run.start (schema_version: 1)astart()tool_call_start (tagged agent_id=Researcher)forwardtool.start (agent_id: Researcher)text.deltaforwardtext.delta (agent_id: Researcher)astart()text.deltaforwardtext.delta (agent_id: Writer)_detach_stream_bridgerun.result (ok: true)

Example NDJSON for a team run

jsonl
{"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:

bash
praisonai run agents.yaml --output stream-json \
  | jq -c 'select(.data.agent_id == "Researcher")'

Route deltas to per-agent buffers in Python:

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.

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.retryBefore a retry wait, on rate-limit / transient-error backoffschema_version, attempt, max_attempts, delay, reason
run.resultSuccessful run endschema_version, ok: true, result
run.errorRun or transport failureschema_version, ok: false, error

Optional fields

FieldTypePresent onDescription
agent_idstr \nulltool.*, text.delta, reasoning.delta, run.retryEmitting 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

jsonl
{"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.

json
{
  "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.

json
{
  "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.

json
{
  "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.

json
{
  "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.

json
{"event":"text.delta","data":{"schema_version":1,"text":"The weather in London "}}
{"event":"text.delta","data":{"schema_version":1,"text":"is currently 15°C."}}
json
{"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.

json
{
  "event": "run.retry",
  "data": {
    "schema_version": 1,
    "attempt": 2,
    "max_attempts": 5,
    "delay": 4.0,
    "reason": "rate_limit"
  }
}
FieldTypeDescription
attemptintThe retry attempt about to run
max_attemptsintTotal attempts allowed
delayfloatSeconds the run will wait before retrying
reasonstrWhy 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.

json
{
  "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.

json
{
  "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.

python
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

Diagram

What do you need?

Human terminal output

–output text (default)

Single final JSON blob

–output json

Live per-step trace

–output stream-json

Just the exit code

–output silent

Diagnostic details

–output verbose


Common Patterns

Filter for specific events with jq

Watch only tool calls and the final result:

bash
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:

bash
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:

python
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

Always check schema_version

Before processing events, verify data.schema_version == 1. This guards your consumer against future breaking changes.

python
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.

python
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.


Related