Skip to content

Streaming

Stream event types yielded by AsyncAgent.run_stream() and AsyncAgentSession.ask_stream(). The event protocol mirrors the Anthropic SDK's raw SSE shape so callers get the same discriminated-union stream whether they use data-harness or the SDK directly.

StreamEvent

StreamEvent is a union alias for all event types:

StreamEvent = (
    MessageStartEvent
    | ContentBlockStartEvent
    | ContentBlockDeltaEvent
    | ContentBlockStopEvent
    | MessageDeltaEvent
    | MessageStopEvent
    | ToolResultEvent
)

Event types

data_harness.MessageStartEvent dataclass 

MessageStartEvent()

data_harness.ContentBlockStartEvent dataclass 

ContentBlockStartEvent(
    index: int, content_block: TextBlock | ToolUseBlock
)

data_harness.ContentBlockDeltaEvent dataclass 

ContentBlockDeltaEvent(
    index: int, delta: TextDelta | InputJSONDelta
)

data_harness.ContentBlockStopEvent dataclass 

ContentBlockStopEvent(index: int)

data_harness.MessageDeltaEvent dataclass 

MessageDeltaEvent(
    stop_reason: StopReason,
    input_tokens: int,
    output_tokens: int,
    cache_read_tokens: int,
    cache_write_tokens: int,
)

data_harness.MessageStopEvent dataclass 

MessageStopEvent()

data_harness.ToolResultEvent dataclass 

ToolResultEvent(
    tool_use_id: str,
    tool_name: str,
    content: str,
    is_error: bool,
)

Emitted by the harness after a tool call is dispatched and returns.

This event has no equivalent in the raw provider stream. It signals that the harness has finished executing the tool and the next turn is starting.

Delta types

data_harness.TextDelta dataclass 

TextDelta(text: str)

data_harness.InputJSONDelta dataclass 

InputJSONDelta(partial_json: str)

Iteration pattern

async for event in agent.run_stream("Describe the dataset."):
    match event.type:
        case "content_block_delta":
            from data_harness import TextDelta
            if isinstance(event.delta, TextDelta):
                print(event.delta.text, end="", flush=True)
        case "tool_result":
            print(f"\n[{event.tool_name}] → {event.content[:80]}")
        case "message_stop":
            print()  # newline after final token