Live Turns

AgentTurn is the advanced agent primitive for interactive runtimes. Use it when chat() and stream() are too flat and your application needs lifecycle events, a structured final result, cancellation, or the ability to steer the model while it is working.

Why Turns Exist

Agents already provide simple methods:

text = await agent.chat("hi")
async for chunk in agent.stream("hi"):
    ...
result = await agent.run("hi")

Those are great for apps that only need a response. Live runtimes need more:

emit text as it streams
show step status
display reasoning events
cancel a run
inject new user guidance before the next model step
wait for the structured final result after streaming events

That is what agent.turn() is for.

Basic Turn

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

async for event in turn.events():
    if event.type == "text.delta":
        print(event.text, end="", flush=True)

result = await turn.result()

turn.events() starts the turn lazily. turn.result() also starts it if it has not started yet.

Event Stream

Current turn events:

Event	Meaning
`turn.started`	The turn started and the user message was appended
`text.delta`	A text chunk streamed from the assistant
`reasoning.delta`	Provider reasoning/thinking event
`step.started`	A model step is about to begin
`step.retrying`	A provider call failed and the runtime is waiting before another attempt
`tool_call.started`	A provider started streaming a tool call
`tool_call.delta`	A function-name or JSON-argument fragment arrived
`tool_call.ready`	A complete provider tool call is ready for execution
`tool_execution.started`	A local tool function started
`tool_execution.finished`	A local tool function finished or was aborted
`tool_result`	The normalized post-hook result is ready for the next model step
`step.finished`	A model step finished
`turn.finished`	The final `TurnResult` is available
`turn.failed`	The turn failed or was aborted

Steering

Use send() to steer a live turn.

await turn.send("Actually, focus on the database layer first.")

Steering is safe-boundary control. The message is queued and injected before the next model call. It does not mutate a provider request already in flight, and it does not interrupt a tool that is already running. steer() is an alias for send():

await turn.steer("Change direction: check pending migrations.")

Cancellation

turn.abort("user cancelled")

Aborting signals the active turn. Providers and tools must honor abort signals to stop immediately; otherwise cancellation happens at the next point the runtime checks the signal.

SSE Example

The stock AgentServer exposes the complete turn stream:

curl -N -X POST \
  "http://localhost:8080/agent/coder/chat?stream=events" \
  -H "Content-Type: application/json" \
  -d '{"message":"Fix the failing test"}'

Each SSE frame contains one public TurnEvent. The SSE event name and JSON type match:

event: tool_execution.started
data: {"type":"tool_execution.started","step_number":1,"index":0,"tool_call":{"id":"call_1","name":"read_file","arguments":{"path":"app.py"},"metadata":{}}}

Use ?stream=true when a client only needs the legacy start, chunk, and end text stream. Use ?stream=events for reasoning, retries, steps, tools, usage, structured completion, and typed failures. The /events endpoint is different: it replays durable application events created with Agent.emit(). Chat turn events are transient and belong to the single POST response that started the turn. Python clients reconstruct the wire payloads into the same event dataclasses:

from ai_query import connect

agent = connect("http://localhost:8080/agent/coder")

async for event in agent.events("Fix the failing test"):
    if event.type == "text.delta":
        print(event.text, end="", flush=True)
    elif event.type == "tool_execution.finished":
        print(event.tool_call.name, event.duration)

CLI Example

class CodingAgent(Agent[dict[str, Any]]):
    async def run_cli_turn(self, message: str) -> str:
        turn = self.turn(message)
        self._current_turn = turn

        try:
            async for event in turn.events():
                if event.type == "text.delta":
                    print(event.text, end="", flush=True)
                elif event.type.startswith("reasoning."):
                    await self.emit("reasoning", {
                        "kind": event.event.kind,
                        "text": event.event.text,
                        "data": event.event.data,
                    })
                elif event.type.startswith("tool_"):
                    await self.emit("tool", {
                        "type": event.type,
                        "step": event.step_number,
                        "index": event.index,
                    })
                elif event.type == "step.started":
                    await self.emit("status", {"step": event.step_number})

            result = await turn.result()
            return result.text
        finally:
            if self._current_turn is turn:
                self._current_turn = None

    async def steer_current_turn(self, message: str) -> None:
        if self._current_turn and not self._current_turn.done:
            await self._current_turn.send(message)

When Not To Use Turns

Do not start with AgentTurn for simple apps. Use:

chat() when you only need text
stream() when you only need streamed text
run() when you need metadata after completion
turn() when you need a live runtime

Start Here

Foundations

Agents

Deployment

Advanced

Why Turns Exist

Basic Turn

Event Stream

Steering

Cancellation

SSE Example

CLI Example

When Not To Use Turns

​Why Turns Exist

​Basic Turn

​Event Stream

​Steering

​Cancellation

​SSE Example

​CLI Example

​When Not To Use Turns

Why Turns Exist

Basic Turn

Event Stream

Steering

Cancellation

SSE Example

CLI Example

When Not To Use Turns