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().

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

stream_text() raises RunFailed (or RunTruncated when the reply hit the token budget) (or another RunError) if the run does not complete, so a consumer that only wants text never has to inspect the result. The on_reset callback fires on ATTEMPT_RESET, when a retried model call restarts the answer; without it the answer simply starts over in the stream. Clear any provisional text you have shown when it fires.

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. Event types include TEXT_DELTA, TEXT_COMPLETE, TOOL_CALL_START, TOOL_CALL_COMPLETE, TOOL_RESULT, THINKING_DELTA, TURN_COMPLETE, ERROR, ATTEMPT_RESET, and RUN_RESULT. Durable runs also emit approval and recovery events. Text deltas are provisional; discard an interrupted attempt's text when ATTEMPT_RESET arrives. RunEvent(data=...) is required when you construct events yourself.

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(). The 0.3 harnessx.streaming module and StreamingAgent were removed in 0.4; they had been a shim over run_stream().

Usage stats accumulate on the guardrails engine:

agent.guardrails.usage_summary
# {'iterations': 3, 'input_tokens': 5120, 'output_tokens': 640, 'estimated_cost': '$0.0250'}