AgentTurn

AgentTurn represents one live execution of an Agent. Create one with agent.turn(...). Do not instantiate AgentTurn directly unless you are extending the framework.

Import

from ai_query.agents import AgentTurn, TurnOptions, TurnResult

Constructor

AgentTurn(
    agent: Agent[Any],
    message: Content,
    *,
    options: TurnOptions | None = None,
) -> None

Normally called internally by:

turn = agent.turn("Fix the failing test")

Properties

str

Unique turn ID.

str

ID of the owning agent.

Agent[Any]

Owning agent instance.

Message

Initial user message for the turn.

TurnOptions

Per-turn options.

bool

True after the turn has started.

bool

True after the turn finishes, fails, or is aborted.

AbortSignal

Public abort signal for tools and other work tied to this turn.

tuple[SteeringEntry, ...]

Immutable ordered snapshot of steering messages waiting for a safe step boundary.

events

async def events(self) -> AsyncIterator[TurnEvent]

Starts the turn if needed and yields typed lifecycle events.

async for event in turn.events():
    ...

Common event types:

Type	Meaning
`turn.started`	The turn was created and user input was appended.
`step.started`	A model step is about to run.
`text.delta`	Assistant answer text.
`reasoning.delta`	Provider reasoning output or metadata; inspect `event.event.kind` for `summary`, `delta`, `signature`, or `state`.
`tool_call.started` / `tool_call.delta`	Provider tool-call fragments, when available.
`tool_call.ready`	A normalized `ToolCall` is ready for execution or error handling.
`tool_execution.started` / `tool_execution.finished`	Local tool function execution lifecycle.
`tool_execution.progress`	Correlated progress emitted by `ToolExecutionContext`.
`tool_result`	Final tool result after hooks, in provider order.
`step.finished`	The model step was persisted.
`turn.finished` / `turn.failed`	Terminal turn outcome.

Returns:

AsyncIterator[TurnEvent]

Stream of turn events. The iterator ends after turn.finished or turn.failed.

text_stream

async def text_stream(self) -> AsyncIterator[str]

Convenience stream over events() that yields only text.delta text.

async for chunk in turn.text_stream():
    print(chunk, end="")

result

async def result(self) -> TurnResult

Starts the turn if needed and waits for the structured result. Returns:

TurnResult

Final structured result.

Raises:

Exception

Raised if the turn is aborted.

Exception

Propagates provider, tool, or runtime failures.

wait

async def wait(self) -> TurnResult

Alias for result().

send

async def send(
    self,
    message: Content,
    *,
    role: Literal["user", "system"] = "user",
) -> SteeringEntry

Queues a steering message for the next safe step boundary and returns its stable ID. Parameters:

Content

required

Message content to inject into the active turn.

Literal['user', 'system']

Role for the injected message. Defaults to "user".

Notes:

send() does not mutate a provider request already in flight.
queued messages are injected before the next model call.
queued messages are scoped to this turn.

steer

async def steer(
    self,
    message: Content,
    *,
    role: Literal["user", "system"] = "user",
) -> SteeringEntry

Alias for send().

cancel_steering

async def cancel_steering(self, steering_id: str) -> bool

Removes a queued steering entry by its stable ID. Returns False if the entry was already claimed or canceled.

abort

def abort(self, reason: str | None = None) -> None

Aborts the turn. Parameters:

str | None

Optional abort reason.

TurnOptions

@dataclass
class TurnOptions:
    tools: ToolSet | None = None
    stop_when: StopCondition | list[StopCondition] | None = None
    reasoning: ReasoningConfig | None = None
    provider_options: ProviderOptions | None = None
    retry: RetryPolicy | None = None
    signal: AbortSignal | None = None
    hooks: AgentHooks | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

Fields:

ToolSet | None

Per-turn tools. Defaults to the agent’s tools.

StopCondition | list[StopCondition] | None

Per-turn stop conditions. Defaults to the agent’s stop_when.

ReasoningConfig | None

Per-turn reasoning controls. Defaults to the agent’s reasoning.

ProviderOptions | None

Per-turn provider options. Defaults to the agent’s provider_options.

RetryPolicy | None

Per-turn provider-call retry policy. Defaults to the agent’s retry.

AbortSignal | None

External abort signal that also aborts the turn.

AgentHooks | None

Per-turn hook overrides. These merge over agent-level default hooks.

dict[str, Any]

Application metadata. Not sent to the provider by default.

TurnResult

@dataclass
class TurnResult:
    turn_id: str
    agent_id: str
    text: str
    finish_reason: str | None
    usage: Usage | None
    steps: list[StepResult]
    started_at: float
    ended_at: float
    output_message: Message
    termination: TurnTermination | None = None

TurnEvent

TurnEvent = (
    TurnStarted
    | TextDelta
    | ReasoningDelta
    | StepStarted
    | StepRetrying
    | ToolCallStartedEvent
    | ToolCallDeltaEvent
    | ToolCallReadyEvent
    | ToolExecutionStartedEvent
    | ToolExecutionProgressEvent
    | ToolExecutionFinishedEvent
    | ToolResultEvent
    | StepFinished
    | TurnFinished
    | TurnFailed
)

TurnStarted

@dataclass
class TurnStarted:
    type: Literal["turn.started"]
    turn_id: str
    agent_id: str
    message: Message
    created_at: float

TextDelta

@dataclass
class TextDelta:
    type: Literal["text.delta"]
    text: str

ReasoningDelta

@dataclass
class ReasoningDelta:
    type: Literal["reasoning.delta"]
    event: ReasoningEvent

StepStarted

@dataclass
class StepStarted:
    type: Literal["step.started"]
    step_number: int

StepFinished

@dataclass
class StepFinished:
    type: Literal["step.finished"]
    step_number: int
    step: StepResult
    usage: Usage | None

usage is the provider-reported usage for this step.

StepRetrying

@dataclass
class StepRetrying:
    type: Literal["step.retrying"]
    step_number: int
    attempt: int
    max_attempts: int
    delay: float
    error: str

Tool lifecycle events

AgentTurn forwards the same ToolCallStartedEvent, ToolCallDeltaEvent, ToolCallReadyEvent, ToolExecutionStartedEvent, ToolExecutionProgressEvent, ToolExecutionFinishedEvent, and ToolResultEvent objects exposed by stream_text().event_stream. This preserves tool call IDs, provider indexes, execution completion order, and normalized results across core and agent APIs.

TurnFinished

@dataclass
class TurnFinished:
    type: Literal["turn.finished"]
    result: TurnResult

TurnFailed

@dataclass
class TurnFailed:
    type: Literal["turn.failed"]
    error: str
    error_type: str | None = None
    aborted: bool = False
    termination: TurnTermination | None = None

TurnResult.termination and TurnFailed.termination expose the same structured diagnostic used by core generation and streaming. Aborted and failed turns also attach it to the exception raised by result().

Wire Codec

from ai_query import turn_event_from_dict, turn_event_to_dict

turn_event_to_dict(event) converts every public TurnEvent, including nested messages, steps, reasoning, tools, results, usage, and final TurnResult, to a stable JSON-compatible object. turn_event_from_dict(payload) reconstructs the matching event dataclass. The stock AgentServer uses this codec for POST /agent/{id}/chat?stream=events. Byte content is represented with an ai-query wire marker and restored as bytes. ToolOutput image and file bytes use the same JSON-safe round trip. Other tool result values must be JSON-compatible. Structured termination diagnostics round-trip on both turn.finished and turn.failed.

Example

turn = agent.turn("Fix the failing test")

async for event in turn.events():
    match event.type:
        case "text.delta":
            print(event.text, end="", flush=True)
        case "step.started":
            print(f"step {event.step_number}")
        case "step.retrying":
            print(f"retrying step {event.step_number}, attempt {event.attempt}")
        case "turn.failed":
            print(event.error)

result = await turn.result()

Core Functions

Agent Classes

Transport Layer

Providers

Types

Import

Constructor

Properties

events

text_stream

result

wait

send

steer

cancel_steering

abort

TurnOptions

TurnResult

TurnEvent

TurnStarted

TextDelta

ReasoningDelta

StepStarted

StepFinished

StepRetrying

Tool lifecycle events

TurnFinished

TurnFailed

Wire Codec

Example

See Also

​Import

​Constructor

​Properties

​events

​text_stream

​result

​wait

​send

​steer

​cancel_steering

​abort

​TurnOptions

​TurnResult

​TurnEvent

​TurnStarted

​TextDelta

​ReasoningDelta

​StepStarted

​StepFinished

​StepRetrying

​Tool lifecycle events

​TurnFinished

​TurnFailed

​Wire Codec

​Example

​See Also

Import

Constructor

Properties

events

text_stream

result

wait

send

steer

cancel_steering

abort

TurnOptions

TurnResult

TurnEvent

TurnStarted

TextDelta

ReasoningDelta

StepStarted

StepFinished

StepRetrying

Tool lifecycle events

TurnFinished

TurnFailed

Wire Codec

Example

See Also