> ## Documentation Index
> Fetch the complete documentation index at: https://thethirdpenco-feat-tool-lifecycle-events.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Live Turns

> Use AgentTurn for events, cancellation, and mid-run steering

`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:

```python theme={null}
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

```python theme={null}
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.

```python theme={null}
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()`:

```python theme={null}
await turn.steer("Change direction: check pending migrations.")
```

## Cancellation

```python theme={null}
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

```python theme={null}
async def handle_stream(agent: Agent, message: str):
    turn = agent.turn(message)

    yield "event: start\ndata: \n\n"

    async for event in turn.events():
        if event.type == "text.delta":
            yield f"event: chunk\ndata: {json.dumps(event.text)}\n\n"
        elif event.type.startswith("reasoning."):
            payload = {
                "kind": event.event.kind,
                "provider": event.event.provider,
                "text": event.event.text,
                "data": event.event.data,
            }
            yield f"event: reasoning\ndata: {json.dumps(payload)}\n\n"
        elif event.type in {
            "tool_call.started",
            "tool_call.delta",
            "tool_call.ready",
            "tool_execution.started",
            "tool_execution.finished",
            "tool_result",
        }:
            payload = {"type": event.type, "step": event.step_number, "index": event.index}
            yield f"event: tool\ndata: {json.dumps(payload)}\n\n"
        elif event.type == "step.started":
            payload = {"step": event.step_number}
            yield f"event: status\ndata: {json.dumps(payload)}\n\n"
        elif event.type == "turn.finished":
            yield f"event: end\ndata: {json.dumps(event.result.text)}\n\n"
```

## CLI Example

```python theme={null}
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