Streaming · HarnessX

How it works

Two views exist over one run. Agent.stream_text() yields the answer's text as plain strings; Agent.run_stream() yields typed RunEvents and exposes the final RunResult. Both use the same engine, tools, and permission checks as agent.run().

Text deltas are provisional. A retried model call restarts the answer and emits ATTEMPT_RESET: discard the text already shown, and treat the final RunResult as authoritative.

Stalled streams and continued replies

A stream that never starts (RetryPolicy.stream_first_event_timeout_seconds, on Anthropic), stalls for RetryPolicy.stream_idle_timeout_seconds, or ends without the provider saying it is complete, is retried the same way, with an ATTEMPT_RESET. See Streams that stall or end early.

A reply cut off at its token budget and then continued is not a reset. The continuation streams after the text already shown, and the final output joins the pieces.

Stream text

import asyncio
from harnessx import Agent, AgentConfig, RunFailed

agent = Agent(config=AgentConfig(system_prompt="You are helpful."))

async def main():
    async with agent:
        try:
            async for chunk in agent.stream_text(
                "Tell me a story",
                on_reset=lambda: print("\n[answer restarted]"),
            ):
                print(chunk, end="", flush=True)
        except RunFailed as exc:
            print("run failed:", exc.error["message"])

asyncio.run(main())

A run that does not end with a whole answer raises a RunError out of stream_text(): RunFailed, RunTruncated when the reply hit the token budget, RunAwaitingInput when a tool needs approval, or RunCancelled. The on_reset callback fires on ATTEMPT_RESET; without it the answer simply starts over in the stream.

With fallbacks configured in AgentConfig.fallbacks, failover inside the chain happens only before the first chunk, and a provider switch is not itself a reset event. After a chunk arrives, a transient failure goes back to the engine, and any further attempt resets as above. See Retries & fallback models.

Refusals and the step limit

stream_text() also raises RunRefused when the model declined or a filter stopped it, and RunLimitReached when the run reached its step limit. RunTruncated is raised only after truncation recovery has run out.

Stop a stream early

Breaking out of the loop early leaves the run to be cancelled when the generator closes. Wrap the generator in contextlib.aclosing() to make that immediate:

import contextlib

async def first_line(agent, prompt):
    text = ""
    async with contextlib.aclosing(agent.stream_text(prompt)) as chunks:
        async for chunk in chunks:
            text += chunk
            if "\n" in text:
                break
    return text.partition("\n")[0]

Stream events

import asyncio
from harnessx import Agent, AgentConfig, RunEventType

agent = Agent(config=AgentConfig(system_prompt="You are helpful."))

async def main():
    async with agent, agent.run_stream("Tell me a story") as stream:
        async for event in stream:
            if event.type == RunEventType.TEXT_DELTA:
                print(event.data, end="", flush=True)
            elif event.type == RunEventType.TOOL_CALL_START:
                print(f"\n> calling {event.data.name}({event.data.input})")
            elif event.type == RunEventType.TOOL_RESULT:
                if event.data.is_error:
                    print(f"  tool error: {event.data.content}")
            elif event.type == RunEventType.TURN_COMPLETE:
                print()
        result = await stream.result()
        result.raise_for_status()

asyncio.run(main())

Use the stream's async context manager to close or cancel an abandoned stream. Durable runs also emit approval and recovery events. RunEventType members:

Event	`event.data`
`TEXT_DELTA`	Answer text fragment
`TEXT_COMPLETE`	Finished text of a turn
`THINKING_DELTA`	Reasoning text fragment
`TOOL_CALL_DELTA`	`ToolCallDelta(id, name, partial_json)` while the model writes a tool call; see below
`TOOL_CALL_START`	Tool name and input, when the call is complete and about to run
`TOOL_CALL_COMPLETE`	Finished call
`TOOL_RESULT`	Content and `is_error`
`TURN_COMPLETE`	End of a model turn
`ATTEMPT_RESET`	Restarted answer
`WAITING`	`{"on": "model", "seconds": ...}` while a model call is slow, `{"on": "retry", ...}` during a long wait before a retry; see below
`TODOS_UPDATED`	The whole task list; see below
`APPROVAL_REQUIRED`	A tool call that needs approval (durable runs)
`RECOVERY_REQUIRED`	A tool call whose outcome is unknown (durable runs)
`GAP`	`{"reason": "event_history_unavailable"}` when a Temporal run's event history cannot be read
`ERROR`	Run error
`RUN_RESULT`	Final `RunResult`

Show a tool call while it is being written

A model can spend a while writing a single tool call, say a whole report passed to write_file. TOOL_CALL_DELTA lets you show something in the meantime. The first delta for a call carries the tool's name and an empty partial_json, which is a good moment to show "calling write_file…". Each delta after that carries the next piece of the arguments, and joined together they make the call's JSON. TOOL_CALL_START still comes once the call is complete and about to run.

from harnessx import RunEventType

async def show_tool_calls(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.TOOL_CALL_DELTA and not event.data.partial_json:
                print(f"\n[writing a call to {event.data.name}...]")
            elif event.type == RunEventType.TOOL_CALL_START:
                print(f"[running {event.data.name}]")
        return await stream.result()

Anthropic, OpenAI, Azure and OpenRouter send the arguments in pieces. Gemini sends each call in one go, so you get two deltas: one with the name and one with all the arguments. Handle them like text deltas: when ATTEMPT_RESET arrives, drop anything you showed from that attempt. stream_text() leaves them out, and so does a stream whose reply is buffered by response middleware.

RunStream.text(on_reset=...) is the same text view over any event stream, including a durable runtime.run_stream():

async def text_only(stream):
    async with stream:
        async for chunk in stream.text():
            print(chunk, end="", flush=True)

AgentRuntime offers the same three verbs: runtime.run(), runtime.run_stream(), and runtime.stream_text().

Show that a slow call is still working

A provider that is slow rather than broken sends nothing: no text, no error, no retry. To the stream it looks like a hung agent. WAITING events fill that silence. The first arrives after first_after_seconds without a chunk, then one every repeat_every_seconds, each carrying the seconds waited so far.

from harnessx import Agent, AgentConfig, ProgressPolicy, RunEventType

agent = Agent(config=AgentConfig(
    progress=ProgressPolicy(first_after_seconds=10.0, repeat_every_seconds=15.0),
))

async def show_progress(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.WAITING:
                print(f"\n[still waiting on the {event.data['on']}: {event.data['seconds']}s]")
            elif event.type == RunEventType.TEXT_DELTA:
                print(event.data, end="", flush=True)
        return await stream.result()

Those values are the defaults. ProgressPolicy(first_after_seconds=None) turns the events off. You get the same events, with "on": "retry", while the agent waits before trying again, for example after a server's Retry-After. They are notices, not deadlines: nothing is cancelled, and the call is still bounded by RetryPolicy.call_timeout_seconds. stream_text() skips them.

Try it with no API key: uv run python examples/01-basics/progress_and_waiting.py.

Task list updates

RunEventType.TODOS_UPDATED fires when the agent changes its task list with write_todos. event.data is a dict with the whole list every time: todos (each {"content", "status"}), completed, total, and in_progress, the item being worked on or None.

from harnessx import RunEventType

async def show_plan(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.TODOS_UPDATED:
                plan = event.data
                print(f"[{plan['completed']}/{plan['total']}] {plan['in_progress'] or ''}")
        return await stream.result()

The event is checked once per batch of tool calls, so a durable or Temporal run reports the plan exactly as a direct one does.