Events & Types¶
Stream event types¶
Agent.astream() yields strongly-typed event objects, all subclasses of
BaseAgentStreamEvent. Import them from opendatasci.streaming.
from opendatasci.streaming import (
BaseAgentStreamEvent,
ReasoningEvent,
TokenEvent,
ToolCallEvent,
ToolCommunicationEvent,
ToolResultEvent,
WorkerDoneEvent,
SubagentEvent,
InputRequiredEvent,
ApprovalRequiredEvent,
UsageEvent,
ResponseEvent,
ErrorEvent,
)
Each class has a type class variable (matching the table below) so
existing event.type == "token" comparisons continue to work alongside
isinstance checks.
Event reference¶
| Class | type |
Key fields | Description |
|---|---|---|---|
ReasoningEvent |
"reasoning" |
content |
Thinking token (Anthropic / Bedrock only) |
TokenEvent |
"token" |
content |
Incremental response text |
ToolCallEvent |
"tool_call" |
content, tool, tool_call_id, summary, worker_summaries |
Agent is invoking a tool |
ToolCommunicationEvent |
"tool_communication" |
content, tool_call_id, tool_name |
In-progress status from a long-running tool |
ToolResultEvent |
"tool_result" |
content, tool_call_id, is_error |
Tool returned its result |
WorkerDoneEvent |
"worker_done" |
worker_idx, success |
A concurrent worker finished |
SubagentEvent |
"subagent_event" |
content, worker_idx, event_type, success, summary |
Lifecycle event from inside a running worker |
InputRequiredEvent |
"input_required" |
content, choices |
Agent paused at a free-text interrupt; call astream(answer) to resume |
ApprovalRequiredEvent |
"approval_required" |
command, description, heads_up |
Agent paused waiting for command approval; call astream("yes") or astream("no") to resume |
UsageEvent |
"usage" |
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens |
Token usage update (fields are int \| None) |
ResponseEvent |
"response" |
content |
End-of-turn marker with the full assembled response |
ErrorEvent |
"error" |
content |
Unrecoverable error |
AgentStreamEvent is exported as a union type alias of all the above for
use in type annotations.
Handling each event type¶
from opendatasci.streaming import (
ApprovalRequiredEvent, InputRequiredEvent, ResponseEvent, TokenEvent,
ToolCallEvent, UsageEvent, WorkerDoneEvent, ErrorEvent,
)
async for event in agent.astream("Analyse this dataset"):
if isinstance(event, TokenEvent):
print(event.content, end="", flush=True)
elif isinstance(event, ToolCallEvent):
print(f"\n→ {event.tool} …")
elif isinstance(event, WorkerDoneEvent):
status = "done" if event.success else "failed"
print(f"\n[worker {event.worker_idx}] {status}")
elif isinstance(event, InputRequiredEvent):
ans = input(f"{event.content} ({'/'.join(event.choices)}): ")
async for follow_up in agent.astream(ans):
...
elif isinstance(event, ApprovalRequiredEvent):
print(f"\nApproval required: {event.description}")
if event.heads_up:
print(f"Warning: {event.heads_up}")
answer = input("Allow? (yes/no): ")
async for follow_up in agent.astream(answer):
...
elif isinstance(event, UsageEvent):
print(f"tokens: {event.input_tokens} in / {event.output_tokens} out")
elif isinstance(event, ResponseEvent):
print()
elif isinstance(event, ErrorEvent):
print(f"\nError: {event.content}")
Reference¶
dataclass
¶
dataclass
¶
dataclass
¶
ToolCallEvent(content: str = '', tool: str = '', tool_call_id: str | None = None, summary: str = '', worker_summaries: list[str] = list())
Bases: BaseAgentStreamEvent
The agent is invoking a tool.
worker_summaries is populated only for spawn_workers tool calls;
summary carries the agent-provided summary argument for all other calls.
dataclass
¶
dataclass
¶
dataclass
¶
dataclass
¶
SubagentEvent(content: str = '', worker_idx: int | None = None, event_type: str = '', success: bool = True, summary: str = '')
Bases: BaseAgentStreamEvent
Lifecycle event from inside a running worker.
event_type is one of "worker_tool_call" or "worker_tool_result".
content carries the tool name for worker_tool_call events.
dataclass
¶
Bases: BaseAgentStreamEvent
The agent is paused at an interrupt and needs input from the user.
content is the question. Call astream again with the user's
answer to resume.
dataclass
¶
Bases: BaseAgentStreamEvent
The agent is paused waiting for the user to approve a command.
description is an LLM-generated plain-language summary of what the
command does; heads_up warns about potential negative impact and is
empty when none was identified. Call astream again with "yes" or
"no" to resume.
dataclass
¶
UsageEvent(input_tokens: int | None = None, output_tokens: int | None = None, cache_read_tokens: int | None = None, cache_creation_tokens: int | None = None)
Bases: BaseAgentStreamEvent
Per-call token usage.
All fields are None when not reported by the underlying provider for
this event (e.g. incremental estimates omit cache fields).