AgentTurn

API reference for live agent turn execution

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

id · str

Unique turn ID.

agent_id · str

ID of the owning agent.

agent · Agent[Any]

Owning agent instance.

message · Message

Initial user message for the turn.

options · TurnOptions

Per-turn options.

started · bool

True after the turn has started.

done · bool

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

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_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] · 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 · TurnResult

Final structured result.

Raises:

AbortError · Exception

Raised if the turn is aborted.

Exception · 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",
) -> None

Queues a steering message for the next safe step boundary.

Parameters:

message · Content · required

Message content to inject into the active turn.

role · 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",
) -> None

Alias for send().

abort

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

Aborts the turn.

Parameters:

reason · 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:

tools · ToolSet | None

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

stop_when · StopCondition | list[StopCondition] | None

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

reasoning · ReasoningConfig | None

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

provider_options · ProviderOptions | None

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

retry · RetryPolicy | None

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

signal · AbortSignal | None

External abort signal that also aborts the turn.

hooks · AgentHooks | None

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

metadata · 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

TurnEvent

TurnEvent = (
    TurnStarted
    | TextDelta
    | ReasoningDelta
    | StepStarted
    | StepRetrying
    | ToolCallStartedEvent
    | ToolCallDeltaEvent
    | ToolCallReadyEvent
    | ToolExecutionStartedEvent
    | 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, 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

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

See Also

• Hooks
• Agent
• Live Turns
• Results