---
title: Messages and Events
description: Build conversation messages and handle streamed events.
type: guide
summary: Construct messages, add files, read outputs, serialize history, handle events, and track usage.
---

# Messages and Events



Messages and events make up AI SDK for Python's data model. Events are used
to communicate transient streaming state, while messages are used to accumulate,
store, and pass around non-streaming state.

Both are Pydantic models, so you can pattern-match, serialize, and persist them
when your app needs that.

## Build messages

Messages store the conversation history you send to the model. Each message
represents a turn in the conversation and is made up of `Part` values:

```python
class Message:
    role: "system" | "user" | "assistant" | "tool" | "internal"
    parts: list[Part]
    ...


Part = (
    TextPart
    | ReasoningPart
    | ToolCallPart
    | ToolResultPart
    | BuiltinToolCallPart
    | BuiltinToolReturnPart
    | FilePart
    | HookPart
)

```

Use built-in factory functions to quickly construct frequently used shapes of
messages:

```python
messages = [
    ai.system_message("Keep robot uprising forecasts concise."),
    ai.user_message("Ask the mothership for an update."),
]
```

Messages round-trip through Pydantic:

```python
encoded = [message.model_dump(mode="json") for message in stream.messages]
restored = [ai.messages.Message.model_validate(item) for item in encoded]
```

## Add files and multimodal input

File parts let you send images, audio, or documents when the provider supports
them.

```python
message = ai.user_message(
    "Inspect this mothership diagram.",
    ai.file_part(image_bytes, media_type="image/png"),
)
```

## Handle message output

Assistant messages expose common part collections:

```python
message = stream.message

print(message.text)
print(message.reasoning)
print(message.tool_calls)
print(message.files)
```

When working with structured outputs, use `get_output` to get the final
assistant message wrapped into the provided Pydantic model.

```python
forecast = message.get_output(Forecast)
```

## Handle stream events

Streams and agents both yield event objects from `ai.events`. Applications can
handle each kind of event as they see fit:

```python
if isinstance(event, ai.events.TextDelta):
    print(event.chunk, end="", flush=True)
```

Text arrives through `TextStart`, `TextDelta`, and `TextEnd` events. Tool calls
arrive through `ToolStart`, `ToolDelta`, and `ToolEnd` events. Provider-executed
tools use the `BuiltinToolStart`, `BuiltinToolDelta`, `BuiltinToolEnd`, and
`BuiltinToolResult` events.

```text
StreamStart

    # text output
    TextStart
    TextDelta ...
    TextEnd

    # reasoning output
    ReasoningStart
    ReasoningDelta ...
    ReasoningEnd

    # host-executed tool call
    ToolStart
    ToolDelta ...
    ToolEnd

    # provider-executed tool call
    BuiltinToolStart
    BuiltinToolDelta ...
    BuiltinToolEnd
    BuiltinToolResult

    # generated file
    FileEvent

StreamEnd
```

```text
# Agent runs can also interleave:
ToolCallResult          # emitted after a Python tool finishes
PartialToolCallResult   # emitted while a streaming tool or subagent yields
HookEvent               # emitted when a hook is deferred, resolved, or cancelled
```

```python
async with ai.stream(model, messages, tools=tools) as stream:
    async for event in stream:
        if isinstance(event, ai.events.TextDelta):
            print(event.chunk, end="", flush=True)
        elif isinstance(event, ai.events.ToolEnd):
            print(f"Tool requested: {event.tool_call.tool_name}")
```

Agents can also emit tool results, hook events, and partial tool output. You can
ignore events you do not need, or route them into your application UI,
observability pipeline, or durable workflow.

Every event also carries a reference to a `ai.Message` that has partial content
accumulated from the stream *so far*.

## Track usage

Providers attach usage to events when they report it. The latest value is also
available on the final message and stream:

```python
async with ai.stream(model, messages) as stream:
    async for event in stream:
        if event.usage is not None:
            print(event.usage)

print(stream.usage)
print(stream.message.usage)
```

## Validate message history

`ai.stream` passes message history to the provider as-is. Providers repair
it before the wire call: they strip internal messages, remove non-model
parts, replace invalid tool args with `{}`, and insert error results for
missing tool calls. To fail fast instead — in tests or import pipelines —
validate the history yourself:

```python
from ai.providers import history_utils


history_utils.validate(messages)  # raises IntegrityError on any issue
issues = history_utils.inspect(messages)  # or just report them
```


---

For a semantic overview of all documentation, see [/sitemap.md](/sitemap.md)

For an index of all available documentation, see [/llms.txt](/llms.txt)

For agent-facing discovery, including API and MCP surfaces, see [/agents.md](/agents.md)