Streaming & events

Streaming as an iterator

import { Agent, stream } from "@rohhitrz/jonaki";

const agent = new Agent({ name: "Assistant" });

for await (const event of stream(agent, "Write a haiku about fireflies.")) {
  if (event.type === "text_delta") process.stdout.write(event.delta);
}

Or as a callback

import { Agent, run } from "@rohhitrz/jonaki";

await run(new Agent({ name: "Assistant" }), "hi", {
  stream: true,
  onEvent: (event) => {
    if (event.type === "text_delta") process.stdout.write(event.delta);
  },
});

One emitter, two shapes

onEvent and the stream() iterator are both listeners on the same EventEmitter. They are not two implementations, so they cannot report different things.

text_delta is only emitted when streaming — stream() turns it on by default; run() needs stream: true.

Getting the result too

stream() returns an object that is async-iterable and exposes the result:

import { stream } from "@rohhitrz/jonaki";

const running = stream(agent, "hi");

for await (const event of running) {
  if (event.type === "text_delta") process.stdout.write(event.delta);
}

const result = await running.result();
console.log(result.usage.costUSD);

If the run fails, the error is thrown from the iterator after the trailing run_failed event — so a consumer sees the failure explained before it is raised.

Event types

Every event carries { type, runId, at }. at is epoch milliseconds.

Event	Extra fields	When
`run_started`	`agentName`, `input`	Once, at the start
`model_call_started`	`agentName`, `provider`, `model`, `turn`	Before each model call
`text_delta`	`delta`	Per token chunk (streaming only)
`model_call_completed`	`step: ModelStep`	After each model call
`tool_started`	`toolCallId`, `name`, `args`	Before each tool
`tool_completed`	`step: ToolStep`	Tool succeeded
`tool_failed`	`step: ToolStep`	Tool errored, was repaired, or was rejected
`handoff_started`	`from`, `to`, `reason?`	Before a transfer
`handoff_completed`	`step: HandoffStep`	After a transfer
`guardrail_triggered`	`step: GuardrailStep`	After each guardrail decision
`approval_required`	`toolCall?`, `guardrail?`, `reason?`	The run is pausing
`run_completed`	`output`, `usage`	Finished successfully
`run_failed`	`code`, `message`	Stopped with an error

Step-backed events carry the trace entry itself, so you never have to reconstruct what the trace already recorded.

A progress UI

import { Agent, stream } from "@rohhitrz/jonaki";

const running = stream(agent, "Book me a flight to Oslo");

for await (const event of running) {
  switch (event.type) {
    case "model_call_started":
      console.log(`thinking (${event.model})…`);
      break;
    case "text_delta":
      process.stdout.write(event.delta);
      break;
    case "tool_started":
      console.log(`\n→ ${event.name}(${JSON.stringify(event.args)})`);
      break;
    case "tool_completed":
      console.log(`✔ ${event.step.name} (${event.step.durationMs}ms)`);
      break;
    case "tool_failed":
      console.log(`✘ ${event.step.name}: ${event.step.error?.message}`);
      break;
    case "handoff_started":
      console.log(`⇢ ${event.from} → ${event.to}`);
      break;
    case "approval_required":
      console.log(`⏸ needs approval: ${event.reason}`);
      break;
    case "run_completed":
      console.log(`\ndone · $${event.usage.costUSD.toFixed(4)}`);
      break;
    case "run_failed":
      console.log(`\nfailed: ${event.code}`);
      break;
  }
}

Collecting cost as you go

import { stream } from "@rohhitrz/jonaki";

let spent = 0;

for await (const event of stream(agent, "hi")) {
  if (event.type === "model_call_completed") {
    spent += event.step.usage.costUSD;
    console.log(`spent so far: $${spent.toFixed(4)}`);
  }
}

Narrowing types

The union is discriminated on type, so a switch or an if narrows it:

import type { JonakiEvent } from "@rohhitrz/jonaki";

function describe(event: JonakiEvent): string {
  if (event.type === "tool_completed") {
    return `${event.step.name} → ${String(event.step.result)}`;
  }
  return event.type;
}

Streaming to a web response

import { Agent, stream } from "@rohhitrz/jonaki";

export async function POST(request: Request) {
  const { prompt } = (await request.json()) as { prompt: string };
  const agent = new Agent({ name: "Assistant" });

  const body = new ReadableStream<Uint8Array>({
    async start(controller) {
      const encoder = new TextEncoder();
      for await (const event of stream(agent, prompt)) {
        if (event.type === "text_delta") {
          controller.enqueue(encoder.encode(event.delta));
        }
      }
      controller.close();
    },
  });

  return new Response(body, {
    headers: { "content-type": "text/plain; charset=utf-8" },
  });
}

Streaming & events

On this page