`pydantic_ai.ui`

StateDeps `dataclass`

Bases: Generic[StateT]

Dependency type that holds state.

This class is used to manage the state of an agent run. It allows setting the state of the agent run with a specific type of state model, which must be a subclass of BaseModel.

The state is set using the state setter by the Adapter when the run starts.

Implements the StateHandler protocol.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@dataclass
class StateDeps(Generic[StateT]):
    """Dependency type that holds state.

    This class is used to manage the state of an agent run. It allows setting
    the state of the agent run with a specific type of state model, which must
    be a subclass of `BaseModel`.

    The state is set using the `state` setter by the `Adapter` when the run starts.

    Implements the `StateHandler` protocol.
    """

    state: StateT

StateHandler

Bases: Protocol

Protocol for state handlers in agent runs. Requires the class to be a dataclass with a state field.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@runtime_checkable
class StateHandler(Protocol):
    """Protocol for state handlers in agent runs. Requires the class to be a dataclass with a `state` field."""

    # Has to be a dataclass so we can use `replace` to update the state.
    # From https://github.com/python/typeshed/blob/9ab7fde0a0cd24ed7a72837fcb21093b811b80d8/stdlib/_typeshed/__init__.pyi#L352
    __dataclass_fields__: ClassVar[dict[str, Field[Any]]]

    @property
    def state(self) -> Any:
        """Get the current state of the agent run."""
        ...

    @state.setter
    def state(self, state: Any) -> None:
        """Set the state of the agent run.

        This method is called to update the state of the agent run with the
        provided state.

        Args:
            state: The run state.
        """
        ...

state `property` `writable`

state: Any

Get the current state of the agent run.

UIAdapter `dataclass`

Bases: ABC, Generic[RunInputT, MessageT, EventT, AgentDepsT, OutputDataT]

Base class for UI adapters.

This class is responsible for transforming agent run input received from the frontend into arguments for Agent.run_stream_events(), running the agent, and then transforming Pydantic AI events into protocol-specific events.

The event stream transformation is handled by a protocol-specific UIEventStream subclass.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@dataclass
class UIAdapter(ABC, Generic[RunInputT, MessageT, EventT, AgentDepsT, OutputDataT]):
    """Base class for UI adapters.

    This class is responsible for transforming agent run input received from the frontend into arguments for [`Agent.run_stream_events()`][pydantic_ai.agent.Agent.run_stream_events], running the agent, and then transforming Pydantic AI events into protocol-specific events.

    The event stream transformation is handled by a protocol-specific [`UIEventStream`][pydantic_ai.ui.UIEventStream] subclass.
    """

    agent: AbstractAgent[AgentDepsT, OutputDataT]
    """The Pydantic AI agent to run."""

    run_input: RunInputT
    """The protocol-specific run input object."""

    _: KW_ONLY

    accept: str | None = None
    """The `Accept` header value of the request, used to determine how to encode the protocol-specific events for the streaming response."""

    manage_system_prompt: Literal['server', 'client'] = 'server'
    """Who owns the system prompt.

    Only affects `system_prompt` — [`instructions`][pydantic_ai.Agent.instructions]
    are always injected by the agent on every request regardless of this setting.

    `'server'` (default): the agent's configured `system_prompt` is authoritative.
    Any `SystemPromptPart` sent by the frontend is stripped with a warning (since a
    malicious client could otherwise inject arbitrary instructions via crafted API
    requests), and the agent's own system prompt is reinjected at the head of the
    first request via the
    [`ReinjectSystemPrompt`][pydantic_ai.capabilities.ReinjectSystemPrompt] capability.

    `'client'`: the frontend owns the system prompt. Frontend `SystemPromptPart`s
    are preserved as-is, and the agent's configured `system_prompt` is not injected
    — the caller is fully responsible for sending it on every turn if desired. To
    opt into the same fallback-to-configured behavior as server mode, add the
    [`ReinjectSystemPrompt`][pydantic_ai.capabilities.ReinjectSystemPrompt] capability
    to your agent.
    """

    allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'})
    """URL schemes that are allowed for [`FileUrl`][pydantic_ai.messages.FileUrl] parts
    ([`ImageUrl`][pydantic_ai.messages.ImageUrl], [`DocumentUrl`][pydantic_ai.messages.DocumentUrl],
    [`VideoUrl`][pydantic_ai.messages.VideoUrl], [`AudioUrl`][pydantic_ai.messages.AudioUrl])
    in client-submitted messages.

    Defaults to `{'http', 'https'}`. Parts whose URL scheme is not in this set are
    dropped with a warning before the messages are passed to the agent.

    Non-HTTP schemes like `s3://` (Bedrock) or `gs://` (Google Cloud) cause the model
    provider to fetch the object using the server-side IAM role or service account,
    so a client that can supply arbitrary URLs can read anything that identity can
    reach. HTTPS URLs are safe to forward because the provider fetches them with
    its own public credentials, and the library's own [`download_item`][pydantic_ai.models.download_item]
    path applies SSRF protection when it has to download them itself.

    For uploads initiated in the browser, prefer pre-signed `https://` URLs over
    cloud-storage schemes. To opt into a cloud-storage scheme after auditing your
    frontend, add it to this set, e.g. `frozenset({'http', 'https', 's3'})`.
    """

    @classmethod
    async def from_request(
        cls,
        request: Request,
        *,
        agent: AbstractAgent[AgentDepsT, OutputDataT],
        manage_system_prompt: Literal['server', 'client'] = 'server',
        allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),
        **kwargs: Any,
    ) -> Self:
        """Create an adapter from a request.

        Extra keyword arguments are forwarded to the adapter constructor, allowing subclasses
        to accept additional adapter-specific parameters.
        """
        return cls(
            agent=agent,
            run_input=cls.build_run_input(await request.body()),
            accept=request.headers.get('accept'),
            manage_system_prompt=manage_system_prompt,
            allowed_file_url_schemes=allowed_file_url_schemes,
            **kwargs,
        )

    @classmethod
    @abstractmethod
    def build_run_input(cls, body: bytes) -> RunInputT:
        """Build a protocol-specific run input object from the request body."""
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def load_messages(cls, messages: Sequence[MessageT]) -> list[ModelMessage]:
        """Transform protocol-specific messages into Pydantic AI messages."""
        raise NotImplementedError

    @classmethod
    def dump_messages(cls, messages: Sequence[ModelMessage]) -> list[MessageT]:
        """Transform Pydantic AI messages into protocol-specific messages."""
        raise NotImplementedError

    @abstractmethod
    def build_event_stream(self) -> UIEventStream[RunInputT, EventT, AgentDepsT, OutputDataT]:
        """Build a protocol-specific event stream transformer."""
        raise NotImplementedError

    @cached_property
    @abstractmethod
    def messages(self) -> list[ModelMessage]:
        """Pydantic AI messages from the protocol-specific run input."""
        raise NotImplementedError

    @cached_property
    def toolset(self) -> AbstractToolset[AgentDepsT] | None:
        """Toolset representing frontend tools from the protocol-specific run input."""
        return None

    @cached_property
    def state(self) -> dict[str, Any] | None:
        """Frontend state from the protocol-specific run input."""
        return None

    @cached_property
    def deferred_tool_results(self) -> DeferredToolResults | None:
        """Deferred tool results extracted from the request, used for tool approval workflows."""
        return None

    @cached_property
    def conversation_id(self) -> str | None:
        """Conversation ID extracted from the protocol-specific run input.

        Used to correlate multiple agent runs that share message history. Returned as
        the `gen_ai.conversation.id` OpenTelemetry span attribute on each run.

        Subclasses for protocols that carry a conversation/thread/chat ID should override this
        (e.g. AG-UI's `RunAgentInput.threadId`, Vercel AI's top-level chat `id`).
        """
        return None

    def sanitize_messages(
        self,
        messages: Sequence[ModelMessage],
        *,
        deferred_tool_results: DeferredToolResults | None = None,
    ) -> list[ModelMessage]:
        """Strip parts of client-submitted messages that aren't trusted from the client.

        Called on the messages produced from the protocol-specific run input before
        they're passed to the agent. Caller-supplied `message_history` is not passed
        through this method — it is trusted as coming from server-side persistence.

        Currently strips:

        - [`SystemPromptPart`][pydantic_ai.messages.SystemPromptPart]s when
          [`manage_system_prompt`][pydantic_ai.ui.UIAdapter.manage_system_prompt] is
          `'server'`. The agent's configured `system_prompt` is reinjected by
          [`ReinjectSystemPrompt`][pydantic_ai.capabilities.ReinjectSystemPrompt] on
          the next model request. If stripping leaves a `ModelRequest` with no parts,
          the request is dropped from history entirely.
        - [`FileUrl`][pydantic_ai.messages.FileUrl] parts whose URL scheme is not in
          [`allowed_file_url_schemes`][pydantic_ai.ui.UIAdapter.allowed_file_url_schemes].
          Non-HTTP schemes like `s3://` or `gs://` cause the model provider to fetch
          the object using the server-side IAM role, so they should only be accepted
          from trusted frontends.
        - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] and
          [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] entries at
          the end of the history that don't have a matching entry in
          `deferred_tool_results`. Tool calls are produced by the model on the server
          side, so an unresolved tool call at the end of client-supplied history doesn't
          correspond to a paused agent run and shouldn't be executed. Tool calls that
          correspond to a resolution in `deferred_tool_results` are preserved so that
          human-in-the-loop resumption continues to work. If stripping leaves the final
          response with no parts, the response is dropped from history entirely.
        """
        resolved_tool_call_ids: set[str] = set()
        if deferred_tool_results is not None:
            resolved_tool_call_ids.update(deferred_tool_results.approvals)
            resolved_tool_call_ids.update(deferred_tool_results.calls)

        strip_system_prompt = self.manage_system_prompt == 'server'
        stripped_system_prompt = False
        disallowed_url_schemes: set[str] = set()
        dangling_tool_call_names: list[str] = []
        last_index = len(messages) - 1

        sanitized: list[ModelMessage] = []
        for index, message in enumerate(messages):
            if isinstance(message, ModelRequest):
                new_request_parts, request_stripped_system_prompt = self._sanitize_request_parts(
                    message.parts, strip_system_prompt=strip_system_prompt, disallowed_schemes=disallowed_url_schemes
                )
                stripped_system_prompt = stripped_system_prompt or request_stripped_system_prompt
                if new_request_parts:
                    sanitized.append(replace(message, parts=new_request_parts))
                # Otherwise drop the request entirely so we don't leave an empty
                # `ModelRequest(parts=[])` in history.
            elif isinstance(message, ModelResponse) and index == last_index:
                new_response_parts = self._sanitize_last_response_parts(
                    message.parts,
                    resolved_tool_call_ids=resolved_tool_call_ids,
                    dangling_names=dangling_tool_call_names,
                )
                if new_response_parts:
                    sanitized.append(replace(message, parts=new_response_parts))
                # Otherwise drop the final response entirely so we don't leave an empty
                # `ModelResponse(parts=[])` in history.
            else:
                sanitized.append(message)

        if stripped_system_prompt:
            warnings.warn(
                "Client-submitted system prompts were stripped because `manage_system_prompt` is `'server'` "
                "(the default). Set `manage_system_prompt='client'` to let the frontend own the system prompt.",
                UserWarning,
                stacklevel=2,
            )

        if disallowed_url_schemes:
            warnings.warn(
                f'Client-submitted file URLs with scheme(s) {sorted(disallowed_url_schemes)!r} '
                f'were dropped because those schemes are not in `allowed_file_url_schemes` '
                f'(currently {sorted(self.allowed_file_url_schemes)!r}). Non-HTTP schemes like '
                f'`s3://` or `gs://` are fetched by the model provider using the server-side IAM role, '
                f'so they should only be accepted from trusted frontends. To allow a scheme, add it to '
                f'`allowed_file_url_schemes` on the adapter.',
                UserWarning,
                stacklevel=2,
            )

        if dangling_tool_call_names:
            warnings.warn(
                f'Client-submitted history ended with unresolved tool call(s) '
                f'{sorted(set(dangling_tool_call_names))!r}, which were stripped. Tool calls are '
                f'produced by the model on the server side, so an unresolved tool call at the end '
                f'of client-supplied history does not correspond to a paused agent run. For '
                f'human-in-the-loop resumption, pass matching `deferred_tool_results` to the run '
                f'method.',
                UserWarning,
                stacklevel=2,
            )

        return sanitized

    def _sanitize_request_parts(
        self,
        parts: Sequence[ModelRequestPart],
        *,
        strip_system_prompt: bool,
        disallowed_schemes: set[str],
    ) -> tuple[list[ModelRequestPart], bool]:
        """Sanitize the parts of a client-submitted [`ModelRequest`][pydantic_ai.messages.ModelRequest].

        `disallowed_schemes` is updated in place with any non-allowlisted file URL schemes encountered.
        Returns the kept parts and whether any [`SystemPromptPart`][pydantic_ai.messages.SystemPromptPart]s were stripped.
        """
        stripped_system_prompt = False
        new_parts: list[ModelRequestPart] = []
        for part in parts:
            if strip_system_prompt and isinstance(part, SystemPromptPart):
                stripped_system_prompt = True
                continue
            if isinstance(part, UserPromptPart) and not isinstance(part.content, str):
                new_parts.append(replace(part, content=self._filter_user_content(part.content, disallowed_schemes)))
            else:
                new_parts.append(part)
        return new_parts, stripped_system_prompt

    def _filter_user_content(
        self,
        content: Sequence[UserContent],
        disallowed_schemes: set[str],
    ) -> list[UserContent]:
        """Drop [`FileUrl`][pydantic_ai.messages.FileUrl] items whose scheme isn't in the allowlist.

        `disallowed_schemes` is updated in place with any disallowed schemes encountered.
        """
        filtered: list[UserContent] = []
        for item in content:
            if isinstance(item, FileUrl):
                scheme = urlparse(item.url).scheme.lower()
                if scheme and scheme not in self.allowed_file_url_schemes:
                    disallowed_schemes.add(scheme)
                    continue
            filtered.append(item)
        return filtered

    def _sanitize_last_response_parts(
        self,
        parts: Sequence[ModelResponsePart],
        *,
        resolved_tool_call_ids: set[str],
        dangling_names: list[str],
    ) -> list[ModelResponsePart]:
        """Sanitize the parts of the trailing client-submitted [`ModelResponse`][pydantic_ai.messages.ModelResponse].

        Drops tool calls that aren't resolved by `deferred_tool_results`. `dangling_names`
        is appended to with the names of any stripped calls.
        """
        new_parts: list[ModelResponsePart] = []
        for part in parts:
            if isinstance(part, BaseToolCallPart) and part.tool_call_id not in resolved_tool_call_ids:
                dangling_names.append(part.tool_name)
                continue
            new_parts.append(part)
        return new_parts

    def transform_stream(
        self,
        stream: AsyncIterator[NativeEvent],
        on_complete: OnCompleteFunc[EventT] | None = None,
    ) -> AsyncIterator[EventT]:
        """Transform a stream of Pydantic AI events into protocol-specific events.

        Args:
            stream: The stream of Pydantic AI events to transform.
            on_complete: Optional callback function called when the agent run completes successfully.
                The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
        """
        return self.build_event_stream().transform_stream(stream, on_complete=on_complete)

    def encode_stream(self, stream: AsyncIterator[EventT]) -> AsyncIterator[str]:
        """Encode a stream of protocol-specific events as strings according to the `Accept` header value.

        Args:
            stream: The stream of protocol-specific events to encode.
        """
        return self.build_event_stream().encode_stream(stream)

    def streaming_response(self, stream: AsyncIterator[EventT]) -> StreamingResponse:
        """Generate a streaming response from a stream of protocol-specific events.

        Args:
            stream: The stream of protocol-specific events to encode.
        """
        return self.build_event_stream().streaming_response(stream)

    def run_stream_native(
        self,
        *,
        output_type: OutputSpec[Any] | None = None,
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        model: Model | KnownModelName | str | None = None,
        instructions: _instructions.AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: ModelSettings | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
    ) -> AsyncIterator[NativeEvent]:
        """Run the agent with the protocol-specific run input and stream Pydantic AI events.

        Args:
            output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
                output validators since output validators would expect an argument that matches the agent's output type.
            message_history: History of the conversation so far.
            deferred_tool_results: Optional results for deferred tool calls in the message history.
            conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
            model: Optional model to use for this run, required if `model` was not set when creating the agent.
            instructions: Optional additional instructions to use for this run.
            deps: Optional dependencies to use for this run.
            model_settings: Optional settings to use for this model's request.
            usage_limits: Optional limits on model request count or token usage.
            usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
            metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
                [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
            infer_name: Whether to try to infer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.
            capabilities: Optional additional [capabilities](https://ai.pydantic.dev/capabilities/) for this run, merged with the agent's configured capabilities.
                Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.
        """
        if deferred_tool_results is None:
            deferred_tool_results = self.deferred_tool_results
        if conversation_id is None:
            conversation_id = self.conversation_id

        frontend_messages = self.sanitize_messages(self.messages, deferred_tool_results=deferred_tool_results)
        message_history = [*(message_history or []), *frontend_messages]

        toolset = self.toolset
        if toolset:
            output_type = [output_type or self.agent.output_type, DeferredToolRequests]
            toolsets = [*(toolsets or []), toolset]

        if isinstance(deps, StateHandler):
            raw_state = self.state or {}
            if isinstance(deps.state, BaseModel):
                state = type(deps.state).model_validate(raw_state)
            else:
                state = raw_state

            deps.state = state
        elif self.state:
            warnings.warn(
                f'State was provided but `deps` of type `{type(deps).__name__}` does not implement the `StateHandler` protocol, so the state was ignored. Use `StateDeps[...]` or implement `StateHandler` to receive AG-UI state.',
                UserWarning,
                stacklevel=2,
            )

        run_capabilities: list[AbstractCapability[AgentDepsT]] = []
        if self.manage_system_prompt == 'server':
            run_capabilities.append(ReinjectSystemPrompt(replace_existing=True))
        if capabilities:
            run_capabilities.extend(capabilities)

        async def stream_events() -> AsyncIterator[NativeEvent]:
            async with self.agent.run_stream_events(
                output_type=output_type,
                message_history=message_history,
                deferred_tool_results=deferred_tool_results,
                conversation_id=conversation_id,
                model=model,
                deps=deps,
                model_settings=model_settings,
                instructions=instructions,
                usage_limits=usage_limits,
                usage=usage,
                metadata=metadata,
                infer_name=infer_name,
                toolsets=toolsets,
                capabilities=run_capabilities,
            ) as events:
                async for event in events:
                    yield event

        return stream_events()

    def run_stream(
        self,
        *,
        output_type: OutputSpec[Any] | None = None,
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        model: Model | KnownModelName | str | None = None,
        instructions: _instructions.AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: ModelSettings | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
        on_complete: OnCompleteFunc[EventT] | None = None,
    ) -> AsyncIterator[EventT]:
        """Run the agent with the protocol-specific run input and stream protocol-specific events.

        Args:
            output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
                output validators since output validators would expect an argument that matches the agent's output type.
            message_history: History of the conversation so far.
            deferred_tool_results: Optional results for deferred tool calls in the message history.
            conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
            model: Optional model to use for this run, required if `model` was not set when creating the agent.
            instructions: Optional additional instructions to use for this run.
            deps: Optional dependencies to use for this run.
            model_settings: Optional settings to use for this model's request.
            usage_limits: Optional limits on model request count or token usage.
            usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
            metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
                [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
            infer_name: Whether to try to infer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.
            capabilities: Optional additional [capabilities](https://ai.pydantic.dev/capabilities/) for this run, merged with the agent's configured capabilities.
                Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.
            on_complete: Optional callback function called when the agent run completes successfully.
                The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
        """
        return self.transform_stream(
            self.run_stream_native(
                output_type=output_type,
                message_history=message_history,
                deferred_tool_results=deferred_tool_results,
                conversation_id=conversation_id,
                model=model,
                instructions=instructions,
                deps=deps,
                model_settings=model_settings,
                usage_limits=usage_limits,
                usage=usage,
                metadata=metadata,
                infer_name=infer_name,
                toolsets=toolsets,
                capabilities=capabilities,
            ),
            on_complete=on_complete,
        )

    @classmethod
    async def dispatch_request(
        cls,
        request: Request,
        *,
        agent: AbstractAgent[DispatchDepsT, DispatchOutputDataT],
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        model: Model | KnownModelName | str | None = None,
        instructions: _instructions.AgentInstructions[DispatchDepsT] = None,
        deps: DispatchDepsT = None,
        output_type: OutputSpec[Any] | None = None,
        model_settings: ModelSettings | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        metadata: AgentMetadata[DispatchDepsT] | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[DispatchDepsT]] | None = None,
        capabilities: Sequence[AbstractCapability[DispatchDepsT]] | None = None,
        on_complete: OnCompleteFunc[EventT] | None = None,
        manage_system_prompt: Literal['server', 'client'] = 'server',
        allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),
        **kwargs: Any,
    ) -> Response:
        """Handle a protocol-specific HTTP request by running the agent and returning a streaming response of protocol-specific events.

        Extra keyword arguments are forwarded to [`from_request`][pydantic_ai.ui.UIAdapter.from_request],
        allowing subclasses to accept additional adapter-specific parameters.

        Args:
            request: The incoming Starlette/FastAPI request.
            agent: The agent to run.
            output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
                output validators since output validators would expect an argument that matches the agent's output type.
            message_history: History of the conversation so far.
            deferred_tool_results: Optional results for deferred tool calls in the message history.
            conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
            model: Optional model to use for this run, required if `model` was not set when creating the agent.
            instructions: Optional additional instructions to use for this run.
            deps: Optional dependencies to use for this run.
            model_settings: Optional settings to use for this model's request.
            usage_limits: Optional limits on model request count or token usage.
            usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
            metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
                [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
            infer_name: Whether to try to infer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.
            capabilities: Optional additional [capabilities](https://ai.pydantic.dev/capabilities/) for this run, merged with the agent's configured capabilities.
                Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.
            on_complete: Optional callback function called when the agent run completes successfully.
                The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
            manage_system_prompt: Who owns the system prompt. See
                [`UIAdapter.manage_system_prompt`][pydantic_ai.ui.UIAdapter.manage_system_prompt].
            allowed_file_url_schemes: URL schemes allowed for file URL parts from the client. See
                [`UIAdapter.allowed_file_url_schemes`][pydantic_ai.ui.UIAdapter.allowed_file_url_schemes].
            **kwargs: Additional keyword arguments forwarded to [`from_request`][pydantic_ai.ui.UIAdapter.from_request].

        Returns:
            A streaming Starlette response with protocol-specific events encoded per the request's `Accept` header value.
        """
        try:
            from starlette.responses import Response
        except ImportError as e:  # pragma: no cover
            raise ImportError(
                'Please install the `starlette` package to use `dispatch_request()` method, '
                'you can use the `ui` optional group — `pip install "pydantic-ai-slim[ui]"`'
            ) from e

        try:
            # The DepsT and OutputDataT come from `agent`, not from `cls`; the cast is necessary to explain this to pyright
            adapter = cast(
                UIAdapter[RunInputT, MessageT, EventT, DispatchDepsT, DispatchOutputDataT],
                await cls.from_request(
                    request,
                    agent=cast(AbstractAgent[AgentDepsT, OutputDataT], agent),
                    manage_system_prompt=manage_system_prompt,
                    allowed_file_url_schemes=allowed_file_url_schemes,
                    **kwargs,
                ),
            )
        except ValidationError as e:  # pragma: no cover
            return Response(
                content=e.json(),
                media_type='application/json',
                status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
            )

        return adapter.streaming_response(
            adapter.run_stream(
                message_history=message_history,
                deferred_tool_results=deferred_tool_results,
                conversation_id=conversation_id,
                deps=deps,
                output_type=output_type,
                model=model,
                instructions=instructions,
                model_settings=model_settings,
                usage_limits=usage_limits,
                usage=usage,
                metadata=metadata,
                infer_name=infer_name,
                toolsets=toolsets,
                capabilities=capabilities,
                on_complete=on_complete,
            ),
        )

agent `instance-attribute`

agent: AbstractAgent[AgentDepsT, OutputDataT]

The Pydantic AI agent to run.

run_input `instance-attribute`

run_input: RunInputT

The protocol-specific run input object.

accept `class-attribute` `instance-attribute`

accept: str | None = None

The Accept header value of the request, used to determine how to encode the protocol-specific events for the streaming response.

manage_system_prompt `class-attribute` `instance-attribute`

manage_system_prompt: Literal["server", "client"] = "server"

Who owns the system prompt.

Only affects system_prompt — [instructions][pydantic_ai.Agent.instructions] are always injected by the agent on every request regardless of this setting.

'server' (default): the agent's configured system_prompt is authoritative. Any SystemPromptPart sent by the frontend is stripped with a warning (since a malicious client could otherwise inject arbitrary instructions via crafted API requests), and the agent's own system prompt is reinjected at the head of the first request via the ReinjectSystemPrompt capability.

'client': the frontend owns the system prompt. Frontend SystemPromptParts are preserved as-is, and the agent's configured system_prompt is not injected — the caller is fully responsible for sending it on every turn if desired. To opt into the same fallback-to-configured behavior as server mode, add the ReinjectSystemPrompt capability to your agent.

allowed_file_url_schemes `class-attribute` `instance-attribute`

allowed_file_url_schemes: frozenset[str] = frozenset(
    {"http", "https"}
)

URL schemes that are allowed for FileUrl parts (ImageUrl, DocumentUrl, VideoUrl, AudioUrl) in client-submitted messages.

Defaults to {'http', 'https'}. Parts whose URL scheme is not in this set are dropped with a warning before the messages are passed to the agent.

Non-HTTP schemes like s3:// (Bedrock) or gs:// (Google Cloud) cause the model provider to fetch the object using the server-side IAM role or service account, so a client that can supply arbitrary URLs can read anything that identity can reach. HTTPS URLs are safe to forward because the provider fetches them with its own public credentials, and the library's own [download_item][pydantic_ai.models.download_item] path applies SSRF protection when it has to download them itself.

For uploads initiated in the browser, prefer pre-signed https:// URLs over cloud-storage schemes. To opt into a cloud-storage scheme after auditing your frontend, add it to this set, e.g. frozenset({'http', 'https', 's3'}).

from_request `async` `classmethod`

from_request(
    request: Request,
    *,
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    manage_system_prompt: Literal[
        "server", "client"
    ] = "server",
    allowed_file_url_schemes: frozenset[str] = frozenset(
        {"http", "https"}
    ),
    **kwargs: Any
) -> Self

Create an adapter from a request.

Extra keyword arguments are forwarded to the adapter constructor, allowing subclasses to accept additional adapter-specific parameters.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@classmethod
async def from_request(
    cls,
    request: Request,
    *,
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    manage_system_prompt: Literal['server', 'client'] = 'server',
    allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),
    **kwargs: Any,
) -> Self:
    """Create an adapter from a request.

    Extra keyword arguments are forwarded to the adapter constructor, allowing subclasses
    to accept additional adapter-specific parameters.
    """
    return cls(
        agent=agent,
        run_input=cls.build_run_input(await request.body()),
        accept=request.headers.get('accept'),
        manage_system_prompt=manage_system_prompt,
        allowed_file_url_schemes=allowed_file_url_schemes,
        **kwargs,
    )

build_run_input `abstractmethod` `classmethod`

build_run_input(body: bytes) -> RunInputT

Build a protocol-specific run input object from the request body.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@classmethod
@abstractmethod
def build_run_input(cls, body: bytes) -> RunInputT:
    """Build a protocol-specific run input object from the request body."""
    raise NotImplementedError

load_messages `abstractmethod` `classmethod`

load_messages(
    messages: Sequence[MessageT],
) -> list[ModelMessage]

Transform protocol-specific messages into Pydantic AI messages.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@classmethod
@abstractmethod
def load_messages(cls, messages: Sequence[MessageT]) -> list[ModelMessage]:
    """Transform protocol-specific messages into Pydantic AI messages."""
    raise NotImplementedError

dump_messages `classmethod`

dump_messages(
    messages: Sequence[ModelMessage],
) -> list[MessageT]

Transform Pydantic AI messages into protocol-specific messages.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@classmethod
def dump_messages(cls, messages: Sequence[ModelMessage]) -> list[MessageT]:
    """Transform Pydantic AI messages into protocol-specific messages."""
    raise NotImplementedError

build_event_stream `abstractmethod`

build_event_stream() -> (
    UIEventStream[
        RunInputT, EventT, AgentDepsT, OutputDataT
    ]
)

Build a protocol-specific event stream transformer.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@abstractmethod
def build_event_stream(self) -> UIEventStream[RunInputT, EventT, AgentDepsT, OutputDataT]:
    """Build a protocol-specific event stream transformer."""
    raise NotImplementedError

messages `abstractmethod` `cached` `property`

messages: list[ModelMessage]

Pydantic AI messages from the protocol-specific run input.

toolset `cached` `property`

toolset: AbstractToolset[AgentDepsT] | None

Toolset representing frontend tools from the protocol-specific run input.

state `cached` `property`

state: dict[str, Any] | None

Frontend state from the protocol-specific run input.

deferred_tool_results `cached` `property`

deferred_tool_results: DeferredToolResults | None

Deferred tool results extracted from the request, used for tool approval workflows.

conversation_id `cached` `property`

conversation_id: str | None

Conversation ID extracted from the protocol-specific run input.

Used to correlate multiple agent runs that share message history. Returned as the gen_ai.conversation.id OpenTelemetry span attribute on each run.

Subclasses for protocols that carry a conversation/thread/chat ID should override this (e.g. AG-UI's RunAgentInput.threadId, Vercel AI's top-level chat id).

sanitize_messages

sanitize_messages(
    messages: Sequence[ModelMessage],
    *,
    deferred_tool_results: DeferredToolResults | None = None
) -> list[ModelMessage]

Strip parts of client-submitted messages that aren't trusted from the client.

Called on the messages produced from the protocol-specific run input before they're passed to the agent. Caller-supplied message_history is not passed through this method — it is trusted as coming from server-side persistence.

Currently strips:

SystemPromptParts when manage_system_prompt is 'server'. The agent's configured system_prompt is reinjected by ReinjectSystemPrompt on the next model request. If stripping leaves a ModelRequest with no parts, the request is dropped from history entirely.
FileUrl parts whose URL scheme is not in allowed_file_url_schemes. Non-HTTP schemes like s3:// or gs:// cause the model provider to fetch the object using the server-side IAM role, so they should only be accepted from trusted frontends.
ToolCallPart and NativeToolCallPart entries at the end of the history that don't have a matching entry in deferred_tool_results. Tool calls are produced by the model on the server side, so an unresolved tool call at the end of client-supplied history doesn't correspond to a paused agent run and shouldn't be executed. Tool calls that correspond to a resolution in deferred_tool_results are preserved so that human-in-the-loop resumption continues to work. If stripping leaves the final response with no parts, the response is dropped from history entirely.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

def sanitize_messages(
    self,
    messages: Sequence[ModelMessage],
    *,
    deferred_tool_results: DeferredToolResults | None = None,
) -> list[ModelMessage]:
    """Strip parts of client-submitted messages that aren't trusted from the client.

    Called on the messages produced from the protocol-specific run input before
    they're passed to the agent. Caller-supplied `message_history` is not passed
    through this method — it is trusted as coming from server-side persistence.

    Currently strips:

    - [`SystemPromptPart`][pydantic_ai.messages.SystemPromptPart]s when
      [`manage_system_prompt`][pydantic_ai.ui.UIAdapter.manage_system_prompt] is
      `'server'`. The agent's configured `system_prompt` is reinjected by
      [`ReinjectSystemPrompt`][pydantic_ai.capabilities.ReinjectSystemPrompt] on
      the next model request. If stripping leaves a `ModelRequest` with no parts,
      the request is dropped from history entirely.
    - [`FileUrl`][pydantic_ai.messages.FileUrl] parts whose URL scheme is not in
      [`allowed_file_url_schemes`][pydantic_ai.ui.UIAdapter.allowed_file_url_schemes].
      Non-HTTP schemes like `s3://` or `gs://` cause the model provider to fetch
      the object using the server-side IAM role, so they should only be accepted
      from trusted frontends.
    - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] and
      [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] entries at
      the end of the history that don't have a matching entry in
      `deferred_tool_results`. Tool calls are produced by the model on the server
      side, so an unresolved tool call at the end of client-supplied history doesn't
      correspond to a paused agent run and shouldn't be executed. Tool calls that
      correspond to a resolution in `deferred_tool_results` are preserved so that
      human-in-the-loop resumption continues to work. If stripping leaves the final
      response with no parts, the response is dropped from history entirely.
    """
    resolved_tool_call_ids: set[str] = set()
    if deferred_tool_results is not None:
        resolved_tool_call_ids.update(deferred_tool_results.approvals)
        resolved_tool_call_ids.update(deferred_tool_results.calls)

    strip_system_prompt = self.manage_system_prompt == 'server'
    stripped_system_prompt = False
    disallowed_url_schemes: set[str] = set()
    dangling_tool_call_names: list[str] = []
    last_index = len(messages) - 1

    sanitized: list[ModelMessage] = []
    for index, message in enumerate(messages):
        if isinstance(message, ModelRequest):
            new_request_parts, request_stripped_system_prompt = self._sanitize_request_parts(
                message.parts, strip_system_prompt=strip_system_prompt, disallowed_schemes=disallowed_url_schemes
            )
            stripped_system_prompt = stripped_system_prompt or request_stripped_system_prompt
            if new_request_parts:
                sanitized.append(replace(message, parts=new_request_parts))
            # Otherwise drop the request entirely so we don't leave an empty
            # `ModelRequest(parts=[])` in history.
        elif isinstance(message, ModelResponse) and index == last_index:
            new_response_parts = self._sanitize_last_response_parts(
                message.parts,
                resolved_tool_call_ids=resolved_tool_call_ids,
                dangling_names=dangling_tool_call_names,
            )
            if new_response_parts:
                sanitized.append(replace(message, parts=new_response_parts))
            # Otherwise drop the final response entirely so we don't leave an empty
            # `ModelResponse(parts=[])` in history.
        else:
            sanitized.append(message)

    if stripped_system_prompt:
        warnings.warn(
            "Client-submitted system prompts were stripped because `manage_system_prompt` is `'server'` "
            "(the default). Set `manage_system_prompt='client'` to let the frontend own the system prompt.",
            UserWarning,
            stacklevel=2,
        )

    if disallowed_url_schemes:
        warnings.warn(
            f'Client-submitted file URLs with scheme(s) {sorted(disallowed_url_schemes)!r} '
            f'were dropped because those schemes are not in `allowed_file_url_schemes` '
            f'(currently {sorted(self.allowed_file_url_schemes)!r}). Non-HTTP schemes like '
            f'`s3://` or `gs://` are fetched by the model provider using the server-side IAM role, '
            f'so they should only be accepted from trusted frontends. To allow a scheme, add it to '
            f'`allowed_file_url_schemes` on the adapter.',
            UserWarning,
            stacklevel=2,
        )

    if dangling_tool_call_names:
        warnings.warn(
            f'Client-submitted history ended with unresolved tool call(s) '
            f'{sorted(set(dangling_tool_call_names))!r}, which were stripped. Tool calls are '
            f'produced by the model on the server side, so an unresolved tool call at the end '
            f'of client-supplied history does not correspond to a paused agent run. For '
            f'human-in-the-loop resumption, pass matching `deferred_tool_results` to the run '
            f'method.',
            UserWarning,
            stacklevel=2,
        )

    return sanitized

transform_stream

transform_stream(
    stream: AsyncIterator[NativeEvent],
    on_complete: OnCompleteFunc[EventT] | None = None,
) -> AsyncIterator[EventT]

Transform a stream of Pydantic AI events into protocol-specific events.

Parameters:

Name	Type	Description	Default
`stream`	`AsyncIterator[NativeEvent]`	The stream of Pydantic AI events to transform.	required
`on_complete`	`OnCompleteFunc[EventT] \| None`	Optional callback function called when the agent run completes successfully. The callback receives the completed `AgentRunResult` and can optionally yield additional protocol-specific events.	`None`

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

def transform_stream(
    self,
    stream: AsyncIterator[NativeEvent],
    on_complete: OnCompleteFunc[EventT] | None = None,
) -> AsyncIterator[EventT]:
    """Transform a stream of Pydantic AI events into protocol-specific events.

    Args:
        stream: The stream of Pydantic AI events to transform.
        on_complete: Optional callback function called when the agent run completes successfully.
            The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
    """
    return self.build_event_stream().transform_stream(stream, on_complete=on_complete)

encode_stream

encode_stream(
    stream: AsyncIterator[EventT],
) -> AsyncIterator[str]

Encode a stream of protocol-specific events as strings according to the Accept header value.

Parameters:

Name	Type	Description	Default
`stream`	`AsyncIterator[EventT]`	The stream of protocol-specific events to encode.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

def encode_stream(self, stream: AsyncIterator[EventT]) -> AsyncIterator[str]:
    """Encode a stream of protocol-specific events as strings according to the `Accept` header value.

    Args:
        stream: The stream of protocol-specific events to encode.
    """
    return self.build_event_stream().encode_stream(stream)

streaming_response

streaming_response(
    stream: AsyncIterator[EventT],
) -> StreamingResponse

Generate a streaming response from a stream of protocol-specific events.

Parameters:

Name	Type	Description	Default
`stream`	`AsyncIterator[EventT]`	The stream of protocol-specific events to encode.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

def streaming_response(self, stream: AsyncIterator[EventT]) -> StreamingResponse:
    """Generate a streaming response from a stream of protocol-specific events.

    Args:
        stream: The stream of protocol-specific events to encode.
    """
    return self.build_event_stream().streaming_response(stream)

run_stream_native

run_stream_native(
    *,
    output_type: OutputSpec[Any] | None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: (
        DeferredToolResults | None
    ) = None,
    conversation_id: str | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    infer_name: bool = True,
    toolsets: (
        Sequence[AbstractToolset[AgentDepsT]] | None
    ) = None,
    capabilities: (
        Sequence[AbstractCapability[AgentDepsT]] | None
    ) = None
) -> AsyncIterator[NativeEvent]

Run the agent with the protocol-specific run input and stream Pydantic AI events.

Parameters:

Name	Type	Description	Default
`output_type`	`OutputSpec[Any] \| None`	Custom output type to use for this run, `output_type` may only be used if the agent has no output validators since output validators would expect an argument that matches the agent's output type.	`None`
`message_history`	`Sequence[ModelMessage] \| None`	History of the conversation so far.	`None`
`deferred_tool_results`	`DeferredToolResults \| None`	Optional results for deferred tool calls in the message history.	`None`
`conversation_id`	`str \| None`	ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.	`None`
`model`	`Model \| KnownModelName \| str \| None`	Optional model to use for this run, required if `model` was not set when creating the agent.	`None`
`instructions`	`AgentInstructions[AgentDepsT]`	Optional additional instructions to use for this run.	`None`
`deps`	`AgentDepsT`	Optional dependencies to use for this run.	`None`
`model_settings`	`ModelSettings \| None`	Optional settings to use for this model's request.	`None`
`usage_limits`	`UsageLimits \| None`	Optional limits on model request count or token usage.	`None`
`usage`	`RunUsage \| None`	Optional usage to start with, useful for resuming a conversation or agents used in tools.	`None`
`metadata`	`AgentMetadata[AgentDepsT] \| None`	Optional metadata to attach to this run. Accepts a dictionary or a callable taking `RunContext`; merged with the agent's configured metadata.	`None`
`infer_name`	`bool`	Whether to try to infer the agent name from the call frame if it's not set.	`True`
`toolsets`	`Sequence[AbstractToolset[AgentDepsT]] \| None`	Optional additional toolsets for this run.	`None`
`capabilities`	`Sequence[AbstractCapability[AgentDepsT]] \| None`	Optional additional capabilities for this run, merged with the agent's configured capabilities. Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.	`None`

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

def run_stream_native(
    self,
    *,
    output_type: OutputSpec[Any] | None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    conversation_id: str | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: _instructions.AgentInstructions[AgentDepsT] = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
    capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
) -> AsyncIterator[NativeEvent]:
    """Run the agent with the protocol-specific run input and stream Pydantic AI events.

    Args:
        output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
            output validators since output validators would expect an argument that matches the agent's output type.
        message_history: History of the conversation so far.
        deferred_tool_results: Optional results for deferred tool calls in the message history.
        conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
        model: Optional model to use for this run, required if `model` was not set when creating the agent.
        instructions: Optional additional instructions to use for this run.
        deps: Optional dependencies to use for this run.
        model_settings: Optional settings to use for this model's request.
        usage_limits: Optional limits on model request count or token usage.
        usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
        metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
            [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
        infer_name: Whether to try to infer the agent name from the call frame if it's not set.
        toolsets: Optional additional toolsets for this run.
        capabilities: Optional additional [capabilities](https://ai.pydantic.dev/capabilities/) for this run, merged with the agent's configured capabilities.
            Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.
    """
    if deferred_tool_results is None:
        deferred_tool_results = self.deferred_tool_results
    if conversation_id is None:
        conversation_id = self.conversation_id

    frontend_messages = self.sanitize_messages(self.messages, deferred_tool_results=deferred_tool_results)
    message_history = [*(message_history or []), *frontend_messages]

    toolset = self.toolset
    if toolset:
        output_type = [output_type or self.agent.output_type, DeferredToolRequests]
        toolsets = [*(toolsets or []), toolset]

    if isinstance(deps, StateHandler):
        raw_state = self.state or {}
        if isinstance(deps.state, BaseModel):
            state = type(deps.state).model_validate(raw_state)
        else:
            state = raw_state

        deps.state = state
    elif self.state:
        warnings.warn(
            f'State was provided but `deps` of type `{type(deps).__name__}` does not implement the `StateHandler` protocol, so the state was ignored. Use `StateDeps[...]` or implement `StateHandler` to receive AG-UI state.',
            UserWarning,
            stacklevel=2,
        )

    run_capabilities: list[AbstractCapability[AgentDepsT]] = []
    if self.manage_system_prompt == 'server':
        run_capabilities.append(ReinjectSystemPrompt(replace_existing=True))
    if capabilities:
        run_capabilities.extend(capabilities)

    async def stream_events() -> AsyncIterator[NativeEvent]:
        async with self.agent.run_stream_events(
            output_type=output_type,
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
            conversation_id=conversation_id,
            model=model,
            deps=deps,
            model_settings=model_settings,
            instructions=instructions,
            usage_limits=usage_limits,
            usage=usage,
            metadata=metadata,
            infer_name=infer_name,
            toolsets=toolsets,
            capabilities=run_capabilities,
        ) as events:
            async for event in events:
                yield event

    return stream_events()

run_stream

run_stream(
    *,
    output_type: OutputSpec[Any] | None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: (
        DeferredToolResults | None
    ) = None,
    conversation_id: str | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[AgentDepsT] = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    infer_name: bool = True,
    toolsets: (
        Sequence[AbstractToolset[AgentDepsT]] | None
    ) = None,
    capabilities: (
        Sequence[AbstractCapability[AgentDepsT]] | None
    ) = None,
    on_complete: OnCompleteFunc[EventT] | None = None
) -> AsyncIterator[EventT]

Run the agent with the protocol-specific run input and stream protocol-specific events.

Parameters:

Name	Type	Description	Default
`output_type`	`OutputSpec[Any] \| None`	Custom output type to use for this run, `output_type` may only be used if the agent has no output validators since output validators would expect an argument that matches the agent's output type.	`None`
`message_history`	`Sequence[ModelMessage] \| None`	History of the conversation so far.	`None`
`deferred_tool_results`	`DeferredToolResults \| None`	Optional results for deferred tool calls in the message history.	`None`
`conversation_id`	`str \| None`	ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.	`None`
`model`	`Model \| KnownModelName \| str \| None`	Optional model to use for this run, required if `model` was not set when creating the agent.	`None`
`instructions`	`AgentInstructions[AgentDepsT]`	Optional additional instructions to use for this run.	`None`
`deps`	`AgentDepsT`	Optional dependencies to use for this run.	`None`
`model_settings`	`ModelSettings \| None`	Optional settings to use for this model's request.	`None`
`usage_limits`	`UsageLimits \| None`	Optional limits on model request count or token usage.	`None`
`usage`	`RunUsage \| None`	Optional usage to start with, useful for resuming a conversation or agents used in tools.	`None`
`metadata`	`AgentMetadata[AgentDepsT] \| None`	Optional metadata to attach to this run. Accepts a dictionary or a callable taking `RunContext`; merged with the agent's configured metadata.	`None`
`infer_name`	`bool`	Whether to try to infer the agent name from the call frame if it's not set.	`True`
`toolsets`	`Sequence[AbstractToolset[AgentDepsT]] \| None`	Optional additional toolsets for this run.	`None`
`capabilities`	`Sequence[AbstractCapability[AgentDepsT]] \| None`	Optional additional capabilities for this run, merged with the agent's configured capabilities. Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.	`None`
`on_complete`	`OnCompleteFunc[EventT] \| None`	Optional callback function called when the agent run completes successfully. The callback receives the completed `AgentRunResult` and can optionally yield additional protocol-specific events.	`None`

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

def run_stream(
    self,
    *,
    output_type: OutputSpec[Any] | None = None,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    conversation_id: str | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: _instructions.AgentInstructions[AgentDepsT] = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[AgentDepsT] | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
    capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
    on_complete: OnCompleteFunc[EventT] | None = None,
) -> AsyncIterator[EventT]:
    """Run the agent with the protocol-specific run input and stream protocol-specific events.

    Args:
        output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
            output validators since output validators would expect an argument that matches the agent's output type.
        message_history: History of the conversation so far.
        deferred_tool_results: Optional results for deferred tool calls in the message history.
        conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
        model: Optional model to use for this run, required if `model` was not set when creating the agent.
        instructions: Optional additional instructions to use for this run.
        deps: Optional dependencies to use for this run.
        model_settings: Optional settings to use for this model's request.
        usage_limits: Optional limits on model request count or token usage.
        usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
        metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
            [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
        infer_name: Whether to try to infer the agent name from the call frame if it's not set.
        toolsets: Optional additional toolsets for this run.
        capabilities: Optional additional [capabilities](https://ai.pydantic.dev/capabilities/) for this run, merged with the agent's configured capabilities.
            Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.
        on_complete: Optional callback function called when the agent run completes successfully.
            The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
    """
    return self.transform_stream(
        self.run_stream_native(
            output_type=output_type,
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
            conversation_id=conversation_id,
            model=model,
            instructions=instructions,
            deps=deps,
            model_settings=model_settings,
            usage_limits=usage_limits,
            usage=usage,
            metadata=metadata,
            infer_name=infer_name,
            toolsets=toolsets,
            capabilities=capabilities,
        ),
        on_complete=on_complete,
    )

dispatch_request `async` `classmethod`

dispatch_request(
    request: Request,
    *,
    agent: AbstractAgent[
        DispatchDepsT, DispatchOutputDataT
    ],
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: (
        DeferredToolResults | None
    ) = None,
    conversation_id: str | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: AgentInstructions[DispatchDepsT] = None,
    deps: DispatchDepsT = None,
    output_type: OutputSpec[Any] | None = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[DispatchDepsT] | None = None,
    infer_name: bool = True,
    toolsets: (
        Sequence[AbstractToolset[DispatchDepsT]] | None
    ) = None,
    capabilities: (
        Sequence[AbstractCapability[DispatchDepsT]] | None
    ) = None,
    on_complete: OnCompleteFunc[EventT] | None = None,
    manage_system_prompt: Literal[
        "server", "client"
    ] = "server",
    allowed_file_url_schemes: frozenset[str] = frozenset(
        {"http", "https"}
    ),
    **kwargs: Any
) -> Response

Handle a protocol-specific HTTP request by running the agent and returning a streaming response of protocol-specific events.

Extra keyword arguments are forwarded to from_request, allowing subclasses to accept additional adapter-specific parameters.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The incoming Starlette/FastAPI request.	required
`agent`	`AbstractAgent[DispatchDepsT, DispatchOutputDataT]`	The agent to run.	required
`output_type`	`OutputSpec[Any] \| None`	Custom output type to use for this run, `output_type` may only be used if the agent has no output validators since output validators would expect an argument that matches the agent's output type.	`None`
`message_history`	`Sequence[ModelMessage] \| None`	History of the conversation so far.	`None`
`deferred_tool_results`	`DeferredToolResults \| None`	Optional results for deferred tool calls in the message history.	`None`
`conversation_id`	`str \| None`	ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.	`None`
`model`	`Model \| KnownModelName \| str \| None`	Optional model to use for this run, required if `model` was not set when creating the agent.	`None`
`instructions`	`AgentInstructions[DispatchDepsT]`	Optional additional instructions to use for this run.	`None`
`deps`	`DispatchDepsT`	Optional dependencies to use for this run.	`None`
`model_settings`	`ModelSettings \| None`	Optional settings to use for this model's request.	`None`
`usage_limits`	`UsageLimits \| None`	Optional limits on model request count or token usage.	`None`
`usage`	`RunUsage \| None`	Optional usage to start with, useful for resuming a conversation or agents used in tools.	`None`
`metadata`	`AgentMetadata[DispatchDepsT] \| None`	Optional metadata to attach to this run. Accepts a dictionary or a callable taking `RunContext`; merged with the agent's configured metadata.	`None`
`infer_name`	`bool`	Whether to try to infer the agent name from the call frame if it's not set.	`True`
`toolsets`	`Sequence[AbstractToolset[DispatchDepsT]] \| None`	Optional additional toolsets for this run.	`None`
`capabilities`	`Sequence[AbstractCapability[DispatchDepsT]] \| None`	Optional additional capabilities for this run, merged with the agent's configured capabilities. Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.	`None`
`on_complete`	`OnCompleteFunc[EventT] \| None`	Optional callback function called when the agent run completes successfully. The callback receives the completed `AgentRunResult` and can optionally yield additional protocol-specific events.	`None`
`manage_system_prompt`	`Literal['server', 'client']`	Who owns the system prompt. See `UIAdapter.manage_system_prompt`.	`'server'`
`allowed_file_url_schemes`	`frozenset[str]`	URL schemes allowed for file URL parts from the client. See `UIAdapter.allowed_file_url_schemes`.	`frozenset({'http', 'https'})`
`**kwargs`	`Any`	Additional keyword arguments forwarded to `from_request`.	`{}`

Returns:

Type	Description
`Response`	A streaming Starlette response with protocol-specific events encoded per the request's `Accept` header value.

Source code in pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@classmethod
async def dispatch_request(
    cls,
    request: Request,
    *,
    agent: AbstractAgent[DispatchDepsT, DispatchOutputDataT],
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    conversation_id: str | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: _instructions.AgentInstructions[DispatchDepsT] = None,
    deps: DispatchDepsT = None,
    output_type: OutputSpec[Any] | None = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    metadata: AgentMetadata[DispatchDepsT] | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[DispatchDepsT]] | None = None,
    capabilities: Sequence[AbstractCapability[DispatchDepsT]] | None = None,
    on_complete: OnCompleteFunc[EventT] | None = None,
    manage_system_prompt: Literal['server', 'client'] = 'server',
    allowed_file_url_schemes: frozenset[str] = frozenset({'http', 'https'}),
    **kwargs: Any,
) -> Response:
    """Handle a protocol-specific HTTP request by running the agent and returning a streaming response of protocol-specific events.

    Extra keyword arguments are forwarded to [`from_request`][pydantic_ai.ui.UIAdapter.from_request],
    allowing subclasses to accept additional adapter-specific parameters.

    Args:
        request: The incoming Starlette/FastAPI request.
        agent: The agent to run.
        output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
            output validators since output validators would expect an argument that matches the agent's output type.
        message_history: History of the conversation so far.
        deferred_tool_results: Optional results for deferred tool calls in the message history.
        conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
        model: Optional model to use for this run, required if `model` was not set when creating the agent.
        instructions: Optional additional instructions to use for this run.
        deps: Optional dependencies to use for this run.
        model_settings: Optional settings to use for this model's request.
        usage_limits: Optional limits on model request count or token usage.
        usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
        metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
            [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
        infer_name: Whether to try to infer the agent name from the call frame if it's not set.
        toolsets: Optional additional toolsets for this run.
        capabilities: Optional additional [capabilities](https://ai.pydantic.dev/capabilities/) for this run, merged with the agent's configured capabilities.
            Use `capabilities=[NativeTool(...)]` to add provider-side native tools per request.
        on_complete: Optional callback function called when the agent run completes successfully.
            The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
        manage_system_prompt: Who owns the system prompt. See
            [`UIAdapter.manage_system_prompt`][pydantic_ai.ui.UIAdapter.manage_system_prompt].
        allowed_file_url_schemes: URL schemes allowed for file URL parts from the client. See
            [`UIAdapter.allowed_file_url_schemes`][pydantic_ai.ui.UIAdapter.allowed_file_url_schemes].
        **kwargs: Additional keyword arguments forwarded to [`from_request`][pydantic_ai.ui.UIAdapter.from_request].

    Returns:
        A streaming Starlette response with protocol-specific events encoded per the request's `Accept` header value.
    """
    try:
        from starlette.responses import Response
    except ImportError as e:  # pragma: no cover
        raise ImportError(
            'Please install the `starlette` package to use `dispatch_request()` method, '
            'you can use the `ui` optional group — `pip install "pydantic-ai-slim[ui]"`'
        ) from e

    try:
        # The DepsT and OutputDataT come from `agent`, not from `cls`; the cast is necessary to explain this to pyright
        adapter = cast(
            UIAdapter[RunInputT, MessageT, EventT, DispatchDepsT, DispatchOutputDataT],
            await cls.from_request(
                request,
                agent=cast(AbstractAgent[AgentDepsT, OutputDataT], agent),
                manage_system_prompt=manage_system_prompt,
                allowed_file_url_schemes=allowed_file_url_schemes,
                **kwargs,
            ),
        )
    except ValidationError as e:  # pragma: no cover
        return Response(
            content=e.json(),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    return adapter.streaming_response(
        adapter.run_stream(
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
            conversation_id=conversation_id,
            deps=deps,
            output_type=output_type,
            model=model,
            instructions=instructions,
            model_settings=model_settings,
            usage_limits=usage_limits,
            usage=usage,
            metadata=metadata,
            infer_name=infer_name,
            toolsets=toolsets,
            capabilities=capabilities,
            on_complete=on_complete,
        ),
    )

SSE_CONTENT_TYPE `module-attribute`

SSE_CONTENT_TYPE = 'text/event-stream'

Content type header value for Server-Sent Events (SSE).

NativeEvent `module-attribute`

NativeEvent: TypeAlias = (
    AgentStreamEvent | AgentRunResultEvent[Any]
)

Type alias for the native event type, which is either an AgentStreamEvent or an AgentRunResultEvent.

OnCompleteFunc `module-attribute`

OnCompleteFunc: TypeAlias = (
    Callable[[AgentRunResult[Any]], None]
    | Callable[[AgentRunResult[Any]], Awaitable[None]]
    | Callable[[AgentRunResult[Any]], AsyncIterator[EventT]]
)

Callback function type that receives the AgentRunResult of the completed run. Can be sync, async, or an async generator of protocol-specific events.

UIEventStream `dataclass`

Bases: ABC, Generic[RunInputT, EventT, AgentDepsT, OutputDataT]

Base class for UI event stream transformers.

This class is responsible for transforming Pydantic AI events into protocol-specific events.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

@dataclass
class UIEventStream(ABC, Generic[RunInputT, EventT, AgentDepsT, OutputDataT]):
    """Base class for UI event stream transformers.

    This class is responsible for transforming Pydantic AI events into protocol-specific events.
    """

    run_input: RunInputT

    accept: str | None = None
    """The `Accept` header value of the request, used to determine how to encode the protocol-specific events for the streaming response."""

    message_id: str = field(default_factory=lambda: str(uuid4()))
    """The message ID to use for the next event."""

    _turn: Literal['request', 'response'] | None = None

    _result: AgentRunResult[OutputDataT] | None = None
    _final_result_event: FinalResultEvent | None = None
    _pending_tool_calls: dict[str, _PendingToolCall] = field(default_factory=dict[str, '_PendingToolCall'])
    """Tool calls dispatched but not yet completed, indexed by `tool_call_id`."""

    def new_message_id(self) -> str:
        """Generate and store a new message ID."""
        self.message_id = str(uuid4())
        return self.message_id

    @property
    def response_headers(self) -> Mapping[str, str] | None:
        """Response headers to return to the frontend."""
        return None

    @property
    def content_type(self) -> str:
        """Get the content type for the event stream, compatible with the `Accept` header value.

        By default, this returns the Server-Sent Events content type (`text/event-stream`).
        If a subclass supports other types as well, it should consider `self.accept` in [`encode_event()`][pydantic_ai.ui.UIEventStream.encode_event] and return the resulting content type.
        """
        return SSE_CONTENT_TYPE

    @abstractmethod
    def encode_event(self, event: EventT) -> str:
        """Encode a protocol-specific event as a string."""
        raise NotImplementedError

    async def encode_stream(self, stream: AsyncIterator[EventT]) -> AsyncIterator[str]:
        """Encode a stream of protocol-specific events as strings according to the `Accept` header value."""
        async for event in stream:
            yield self.encode_event(event)

    def streaming_response(self, stream: AsyncIterator[EventT]) -> StreamingResponse:
        """Generate a streaming response from a stream of protocol-specific events."""
        try:
            from starlette.responses import StreamingResponse
        except ImportError as e:  # pragma: no cover
            raise ImportError(
                'Please install the `starlette` package to use the `streaming_response()` method, '
                'you can use the `ui` optional group — `pip install "pydantic-ai-slim[ui]"`'
            ) from e

        return StreamingResponse(
            self.encode_stream(stream),
            headers=self.response_headers,
            media_type=self.content_type,
        )

    async def transform_stream(  # noqa: C901
        self, stream: AsyncIterator[NativeEvent], on_complete: OnCompleteFunc[EventT] | None = None
    ) -> AsyncIterator[EventT]:
        """Transform a stream of Pydantic AI events into protocol-specific events.

        This method dispatches to specific hooks and `handle_*` methods that subclasses can override:
        - [`before_stream()`][pydantic_ai.ui.UIEventStream.before_stream]
        - [`after_stream()`][pydantic_ai.ui.UIEventStream.after_stream]
        - [`on_error()`][pydantic_ai.ui.UIEventStream.on_error]
        - [`before_request()`][pydantic_ai.ui.UIEventStream.before_request]
        - [`after_request()`][pydantic_ai.ui.UIEventStream.after_request]
        - [`before_response()`][pydantic_ai.ui.UIEventStream.before_response]
        - [`after_response()`][pydantic_ai.ui.UIEventStream.after_response]
        - [`handle_event()`][pydantic_ai.ui.UIEventStream.handle_event]

        Args:
            stream: The stream of Pydantic AI events to transform.
            on_complete: Optional callback function called when the agent run completes successfully.
                The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
        """
        async for e in self.before_stream():
            yield e

        try:
            async for event in stream:
                if isinstance(event, PartStartEvent):
                    async for e in self._turn_to('response'):
                        yield e
                elif isinstance(event, ToolCallEvent):
                    tool_call_id = event.part.tool_call_id
                    kind: Literal['function', 'output'] = (
                        'output' if isinstance(event, OutputToolCallEvent) else 'function'
                    )
                    self._pending_tool_calls[tool_call_id] = _PendingToolCall(kind, event.part.tool_name)
                    if kind == 'output':
                        # The output tool call is now tracked in `_pending_tool_calls`,
                        # so the `FinalResultEvent` backup used by the error path is no longer needed.
                        self._final_result_event = None
                    async for e in self._turn_to('request'):
                        yield e
                elif isinstance(event, AgentRunResultEvent):
                    result = cast(AgentRunResult[OutputDataT], event.result)
                    self._result = result

                    async for e in self._turn_to(None):
                        yield e

                    if on_complete is not None:
                        if inspect.isasyncgenfunction(on_complete):
                            async for e in on_complete(result):
                                yield e
                        elif _utils.is_async_callable(on_complete):
                            await on_complete(result)
                        else:
                            await _utils.run_in_executor(on_complete, result)
                elif isinstance(event, FinalResultEvent):
                    self._final_result_event = event

                elif isinstance(event, ToolResultEvent):
                    tool_call_id = event.part.tool_call_id
                    self._pending_tool_calls.pop(tool_call_id, None)

                async for e in self.handle_event(event):
                    yield e
        except Exception as exc:  # `exc` to avoid shadowing by `async for e in` below
            # Close any pending tool calls before emitting the error,
            # so the UI doesn't show them as still running.

            # Pending output-tool call (stored via FinalResultEvent if the call event hasn't fired yet)
            if (
                self._final_result_event
                and (tool_call_id := self._final_result_event.tool_call_id)
                and (tool_name := self._final_result_event.tool_name)
            ):
                self._final_result_event = None
                self._pending_tool_calls[tool_call_id] = _PendingToolCall('output', tool_name)

            # Pending tool calls
            for tool_call_id, (kind, tool_name) in self._pending_tool_calls.items():
                async for e in self._turn_to('request'):
                    yield e
                error_part = ToolReturnPart(
                    tool_call_id=tool_call_id,
                    tool_name=tool_name,
                    content='Tool execution was interrupted by an error.',
                    outcome='failed',
                )
                if kind == 'output':
                    async for e in self.handle_output_tool_result(OutputToolResultEvent(error_part)):
                        yield e
                else:
                    async for e in self.handle_function_tool_result(FunctionToolResultEvent(error_part)):
                        yield e
            self._pending_tool_calls.clear()

            async for e in self.on_error(exc):
                yield e
        finally:
            async for e in self._turn_to(None):
                yield e

            async for e in self.after_stream():
                yield e

    async def _turn_to(self, to_turn: Literal['request', 'response'] | None) -> AsyncIterator[EventT]:
        """Fire hooks when turning from request to response or vice versa."""
        if to_turn == self._turn:
            return

        if self._turn == 'request':
            async for e in self.after_request():
                yield e
        elif self._turn == 'response':
            async for e in self.after_response():
                yield e

        self._turn = to_turn

        if to_turn == 'request':
            async for e in self.before_request():
                yield e
        elif to_turn == 'response':
            async for e in self.before_response():
                yield e

    async def handle_event(self, event: NativeEvent) -> AsyncIterator[EventT]:  # noqa: C901
        """Transform a Pydantic AI event into one or more protocol-specific events.

        This method dispatches to specific `handle_*` methods based on event type:

        - [`PartStartEvent`][pydantic_ai.messages.PartStartEvent] -> [`handle_part_start()`][pydantic_ai.ui.UIEventStream.handle_part_start]
        - [`PartDeltaEvent`][pydantic_ai.messages.PartDeltaEvent] -> `handle_part_delta`
        - [`PartEndEvent`][pydantic_ai.messages.PartEndEvent] -> `handle_part_end`
        - [`FinalResultEvent`][pydantic_ai.messages.FinalResultEvent] -> `handle_final_result`
        - [`FunctionToolCallEvent`][pydantic_ai.messages.FunctionToolCallEvent] -> `handle_function_tool_call`
        - [`FunctionToolResultEvent`][pydantic_ai.messages.FunctionToolResultEvent] -> `handle_function_tool_result`
        - [`OutputToolCallEvent`][pydantic_ai.messages.OutputToolCallEvent] -> `handle_output_tool_call`
        - [`OutputToolResultEvent`][pydantic_ai.messages.OutputToolResultEvent] -> `handle_output_tool_result`
        - [`AgentRunResultEvent`][pydantic_ai.run.AgentRunResultEvent] -> `handle_run_result`

        Subclasses are encouraged to override the individual `handle_*` methods rather than this one.
        If you need specific behavior for all events, make sure you call the super method.
        """
        match event:
            case PartStartEvent():
                async for e in self.handle_part_start(event):
                    yield e
            case PartDeltaEvent():
                async for e in self.handle_part_delta(event):
                    yield e
            case PartEndEvent():
                async for e in self.handle_part_end(event):
                    yield e
            case FinalResultEvent():
                async for e in self.handle_final_result(event):
                    yield e
            case FunctionToolCallEvent():
                async for e in self.handle_function_tool_call(event):
                    yield e
            case FunctionToolResultEvent():
                async for e in self.handle_function_tool_result(event):
                    yield e
            case OutputToolCallEvent():
                async for e in self.handle_output_tool_call(event):
                    yield e
            case OutputToolResultEvent():
                async for e in self.handle_output_tool_result(event):
                    yield e
            case AgentRunResultEvent():
                async for e in self.handle_run_result(event):
                    yield e
            case _:
                pass

    async def handle_part_start(self, event: PartStartEvent) -> AsyncIterator[EventT]:
        """Handle a `PartStartEvent`.

        This method dispatches to specific `handle_*` methods based on part type:

        - [`TextPart`][pydantic_ai.messages.TextPart] -> [`handle_text_start()`][pydantic_ai.ui.UIEventStream.handle_text_start]
        - [`ThinkingPart`][pydantic_ai.messages.ThinkingPart] -> [`handle_thinking_start()`][pydantic_ai.ui.UIEventStream.handle_thinking_start]
        - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] -> [`handle_tool_call_start()`][pydantic_ai.ui.UIEventStream.handle_tool_call_start]
        - [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] -> [`handle_builtin_tool_call_start()`][pydantic_ai.ui.UIEventStream.handle_builtin_tool_call_start]
        - [`NativeToolReturnPart`][pydantic_ai.messages.NativeToolReturnPart] -> [`handle_builtin_tool_return()`][pydantic_ai.ui.UIEventStream.handle_builtin_tool_return]
        - [`FilePart`][pydantic_ai.messages.FilePart] -> [`handle_file()`][pydantic_ai.ui.UIEventStream.handle_file]
        - [`CompactionPart`][pydantic_ai.messages.CompactionPart] -> [`handle_compaction()`][pydantic_ai.ui.UIEventStream.handle_compaction]

        Subclasses are encouraged to override the individual `handle_*` methods rather than this one.
        If you need specific behavior for all part start events, make sure you call the super method.

        Args:
            event: The part start event.
        """
        part = event.part
        previous_part_kind = event.previous_part_kind
        match part:
            case TextPart():
                async for e in self.handle_text_start(part, follows_text=previous_part_kind == 'text'):
                    yield e
            case ThinkingPart():
                async for e in self.handle_thinking_start(part, follows_thinking=previous_part_kind == 'thinking'):
                    yield e
            case ToolCallPart():
                async for e in self.handle_tool_call_start(part):
                    yield e
            case NativeToolCallPart():
                async for e in self.handle_builtin_tool_call_start(part):
                    yield e
            case NativeToolReturnPart():
                async for e in self.handle_builtin_tool_return(part):
                    yield e
            case FilePart():
                async for e in self.handle_file(part):
                    yield e
            case CompactionPart():  # pragma: no cover
                async for e in self.handle_compaction(part):
                    yield e

    async def handle_part_delta(self, event: PartDeltaEvent) -> AsyncIterator[EventT]:
        """Handle a PartDeltaEvent.

        This method dispatches to specific `handle_*_delta` methods based on part delta type:

        - [`TextPartDelta`][pydantic_ai.messages.TextPartDelta] -> [`handle_text_delta()`][pydantic_ai.ui.UIEventStream.handle_text_delta]
        - [`ThinkingPartDelta`][pydantic_ai.messages.ThinkingPartDelta] -> [`handle_thinking_delta()`][pydantic_ai.ui.UIEventStream.handle_thinking_delta]
        - [`ToolCallPartDelta`][pydantic_ai.messages.ToolCallPartDelta] -> [`handle_tool_call_delta()`][pydantic_ai.ui.UIEventStream.handle_tool_call_delta]

        Subclasses are encouraged to override the individual `handle_*_delta` methods rather than this one.
        If you need specific behavior for all part delta events, make sure you call the super method.

        Args:
            event: The PartDeltaEvent.
        """
        delta = event.delta
        match delta:
            case TextPartDelta():
                async for e in self.handle_text_delta(delta):
                    yield e
            case ThinkingPartDelta():
                async for e in self.handle_thinking_delta(delta):
                    yield e
            case ToolCallPartDelta():  # pragma: no branch
                async for e in self.handle_tool_call_delta(delta):
                    yield e

    async def handle_part_end(self, event: PartEndEvent) -> AsyncIterator[EventT]:
        """Handle a `PartEndEvent`.

        This method dispatches to specific `handle_*_end` methods based on part type:

        - [`TextPart`][pydantic_ai.messages.TextPart] -> [`handle_text_end()`][pydantic_ai.ui.UIEventStream.handle_text_end]
        - [`ThinkingPart`][pydantic_ai.messages.ThinkingPart] -> [`handle_thinking_end()`][pydantic_ai.ui.UIEventStream.handle_thinking_end]
        - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] -> [`handle_tool_call_end()`][pydantic_ai.ui.UIEventStream.handle_tool_call_end]
        - [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] -> [`handle_builtin_tool_call_end()`][pydantic_ai.ui.UIEventStream.handle_builtin_tool_call_end]

        Subclasses are encouraged to override the individual `handle_*_end` methods rather than this one.
        If you need specific behavior for all part end events, make sure you call the super method.

        Args:
            event: The part end event.
        """
        part = event.part
        next_part_kind = event.next_part_kind
        match part:
            case TextPart():
                async for e in self.handle_text_end(part, followed_by_text=next_part_kind == 'text'):
                    yield e
            case ThinkingPart():
                async for e in self.handle_thinking_end(part, followed_by_thinking=next_part_kind == 'thinking'):
                    yield e
            case ToolCallPart():
                async for e in self.handle_tool_call_end(part):
                    yield e
            case NativeToolCallPart():
                async for e in self.handle_builtin_tool_call_end(part):
                    yield e
            case NativeToolReturnPart() | FilePart() | CompactionPart():  # pragma: no cover
                # These don't have deltas, so they don't need to be ended.
                pass

    async def before_stream(self) -> AsyncIterator[EventT]:
        """Yield events before agent streaming starts.

        This hook is called before any agent events are processed.
        Override this to inject custom events at the start of the stream.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def after_stream(self) -> AsyncIterator[EventT]:
        """Yield events after agent streaming completes.

        This hook is called after all agent events have been processed.
        Override this to inject custom events at the end of the stream.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def on_error(self, error: Exception) -> AsyncIterator[EventT]:
        """Handle errors that occur during streaming.

        Args:
            error: The error that occurred during streaming.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def before_request(self) -> AsyncIterator[EventT]:
        """Yield events before a model request is processed.

        Override this to inject custom events at the start of the request.
        """
        return  # pragma: lax no cover
        yield  # Make this an async generator

    async def after_request(self) -> AsyncIterator[EventT]:
        """Yield events after a model request is processed.

        Override this to inject custom events at the end of the request.
        """
        return  # pragma: lax no cover
        yield  # Make this an async generator

    async def before_response(self) -> AsyncIterator[EventT]:
        """Yield events before a model response is processed.

        Override this to inject custom events at the start of the response.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def after_response(self) -> AsyncIterator[EventT]:
        """Yield events after a model response is processed.

        Override this to inject custom events at the end of the response.
        """
        return  # pragma: lax no cover
        yield  # Make this an async generator

    async def handle_text_start(self, part: TextPart, follows_text: bool = False) -> AsyncIterator[EventT]:
        """Handle the start of a `TextPart`.

        Args:
            part: The text part.
            follows_text: Whether the part is directly preceded by another text part. In this case, you may want to yield a "text-delta" event instead of a "text-start" event.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_text_delta(self, delta: TextPartDelta) -> AsyncIterator[EventT]:
        """Handle a `TextPartDelta`.

        Args:
            delta: The text part delta.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_text_end(self, part: TextPart, followed_by_text: bool = False) -> AsyncIterator[EventT]:
        """Handle the end of a `TextPart`.

        Args:
            part: The text part.
            followed_by_text: Whether the part is directly followed by another text part. In this case, you may not want to yield a "text-end" event yet.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_thinking_start(self, part: ThinkingPart, follows_thinking: bool = False) -> AsyncIterator[EventT]:
        """Handle the start of a `ThinkingPart`.

        Args:
            part: The thinking part.
            follows_thinking: Whether the part is directly preceded by another thinking part. In this case, you may want to yield a "thinking-delta" event instead of a "thinking-start" event.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_thinking_delta(self, delta: ThinkingPartDelta) -> AsyncIterator[EventT]:
        """Handle a `ThinkingPartDelta`.

        Args:
            delta: The thinking part delta.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_thinking_end(
        self, part: ThinkingPart, followed_by_thinking: bool = False
    ) -> AsyncIterator[EventT]:
        """Handle the end of a `ThinkingPart`.

        Args:
            part: The thinking part.
            followed_by_thinking: Whether the part is directly followed by another thinking part. In this case, you may not want to yield a "thinking-end" event yet.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_tool_call_start(self, part: ToolCallPart) -> AsyncIterator[EventT]:
        """Handle the start of a `ToolCallPart`.

        Args:
            part: The tool call part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_tool_call_delta(self, delta: ToolCallPartDelta) -> AsyncIterator[EventT]:
        """Handle a `ToolCallPartDelta`.

        Args:
            delta: The tool call part delta.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_tool_call_end(self, part: ToolCallPart) -> AsyncIterator[EventT]:
        """Handle the end of a `ToolCallPart`.

        Args:
            part: The tool call part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_builtin_tool_call_start(self, part: NativeToolCallPart) -> AsyncIterator[EventT]:
        """Handle a `NativeToolCallPart` at start.

        Args:
            part: The builtin tool call part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_builtin_tool_call_end(self, part: NativeToolCallPart) -> AsyncIterator[EventT]:
        """Handle the end of a `NativeToolCallPart`.

        Args:
            part: The builtin tool call part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_builtin_tool_return(self, part: NativeToolReturnPart) -> AsyncIterator[EventT]:
        """Handle a `NativeToolReturnPart`.

        Args:
            part: The builtin tool return part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_file(self, part: FilePart) -> AsyncIterator[EventT]:
        """Handle a `FilePart`.

        Args:
            part: The file part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_compaction(self, part: CompactionPart) -> AsyncIterator[EventT]:
        """Handle a `CompactionPart`.

        Args:
            part: The compaction part.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_final_result(self, event: FinalResultEvent) -> AsyncIterator[EventT]:
        """Handle a `FinalResultEvent`.

        Args:
            event: The final result event.
        """
        return
        yield  # Make this an async generator

    async def handle_function_tool_call(self, event: FunctionToolCallEvent) -> AsyncIterator[EventT]:
        """Handle a `FunctionToolCallEvent`.

        Args:
            event: The function tool call event.
        """
        return
        yield  # Make this an async generator

    async def handle_function_tool_result(self, event: FunctionToolResultEvent) -> AsyncIterator[EventT]:
        """Handle a `FunctionToolResultEvent`.

        Args:
            event: The function tool result event.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_output_tool_call(self, event: OutputToolCallEvent) -> AsyncIterator[EventT]:
        """Handle an `OutputToolCallEvent` (the model's "submit final answer" call).

        Args:
            event: The output tool call event.
        """
        return
        yield  # Make this an async generator

    async def handle_output_tool_result(self, event: OutputToolResultEvent) -> AsyncIterator[EventT]:
        """Handle an `OutputToolResultEvent` (the result of an output tool call).

        Args:
            event: The output tool result event.
        """
        return  # pragma: no cover
        yield  # Make this an async generator

    async def handle_run_result(self, event: AgentRunResultEvent) -> AsyncIterator[EventT]:
        """Handle an `AgentRunResultEvent`.

        Args:
            event: The agent run result event.
        """
        return
        yield  # Make this an async generator

accept `class-attribute` `instance-attribute`

accept: str | None = None

The Accept header value of the request, used to determine how to encode the protocol-specific events for the streaming response.

message_id `class-attribute` `instance-attribute`

message_id: str = field(
    default_factory=lambda: str(uuid4())
)

The message ID to use for the next event.

new_message_id

new_message_id() -> str

Generate and store a new message ID.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

def new_message_id(self) -> str:
    """Generate and store a new message ID."""
    self.message_id = str(uuid4())
    return self.message_id

response_headers `property`

response_headers: Mapping[str, str] | None

Response headers to return to the frontend.

content_type `property`

content_type: str

Get the content type for the event stream, compatible with the Accept header value.

By default, this returns the Server-Sent Events content type (text/event-stream). If a subclass supports other types as well, it should consider self.accept in encode_event() and return the resulting content type.

encode_event `abstractmethod`

encode_event(event: EventT) -> str

Encode a protocol-specific event as a string.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

@abstractmethod
def encode_event(self, event: EventT) -> str:
    """Encode a protocol-specific event as a string."""
    raise NotImplementedError

encode_stream `async`

encode_stream(
    stream: AsyncIterator[EventT],
) -> AsyncIterator[str]

Encode a stream of protocol-specific events as strings according to the Accept header value.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def encode_stream(self, stream: AsyncIterator[EventT]) -> AsyncIterator[str]:
    """Encode a stream of protocol-specific events as strings according to the `Accept` header value."""
    async for event in stream:
        yield self.encode_event(event)

streaming_response

streaming_response(
    stream: AsyncIterator[EventT],
) -> StreamingResponse

Generate a streaming response from a stream of protocol-specific events.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

def streaming_response(self, stream: AsyncIterator[EventT]) -> StreamingResponse:
    """Generate a streaming response from a stream of protocol-specific events."""
    try:
        from starlette.responses import StreamingResponse
    except ImportError as e:  # pragma: no cover
        raise ImportError(
            'Please install the `starlette` package to use the `streaming_response()` method, '
            'you can use the `ui` optional group — `pip install "pydantic-ai-slim[ui]"`'
        ) from e

    return StreamingResponse(
        self.encode_stream(stream),
        headers=self.response_headers,
        media_type=self.content_type,
    )

transform_stream `async`

transform_stream(
    stream: AsyncIterator[NativeEvent],
    on_complete: OnCompleteFunc[EventT] | None = None,
) -> AsyncIterator[EventT]

Transform a stream of Pydantic AI events into protocol-specific events.

This method dispatches to specific hooks and handle_* methods that subclasses can override: - before_stream() - after_stream() - on_error() - before_request() - after_request() - before_response() - after_response() - handle_event()

Parameters:

Name	Type	Description	Default
`stream`	`AsyncIterator[NativeEvent]`	The stream of Pydantic AI events to transform.	required
`on_complete`	`OnCompleteFunc[EventT] \| None`	Optional callback function called when the agent run completes successfully. The callback receives the completed `AgentRunResult` and can optionally yield additional protocol-specific events.	`None`

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def transform_stream(  # noqa: C901
    self, stream: AsyncIterator[NativeEvent], on_complete: OnCompleteFunc[EventT] | None = None
) -> AsyncIterator[EventT]:
    """Transform a stream of Pydantic AI events into protocol-specific events.

    This method dispatches to specific hooks and `handle_*` methods that subclasses can override:
    - [`before_stream()`][pydantic_ai.ui.UIEventStream.before_stream]
    - [`after_stream()`][pydantic_ai.ui.UIEventStream.after_stream]
    - [`on_error()`][pydantic_ai.ui.UIEventStream.on_error]
    - [`before_request()`][pydantic_ai.ui.UIEventStream.before_request]
    - [`after_request()`][pydantic_ai.ui.UIEventStream.after_request]
    - [`before_response()`][pydantic_ai.ui.UIEventStream.before_response]
    - [`after_response()`][pydantic_ai.ui.UIEventStream.after_response]
    - [`handle_event()`][pydantic_ai.ui.UIEventStream.handle_event]

    Args:
        stream: The stream of Pydantic AI events to transform.
        on_complete: Optional callback function called when the agent run completes successfully.
            The callback receives the completed [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] and can optionally yield additional protocol-specific events.
    """
    async for e in self.before_stream():
        yield e

    try:
        async for event in stream:
            if isinstance(event, PartStartEvent):
                async for e in self._turn_to('response'):
                    yield e
            elif isinstance(event, ToolCallEvent):
                tool_call_id = event.part.tool_call_id
                kind: Literal['function', 'output'] = (
                    'output' if isinstance(event, OutputToolCallEvent) else 'function'
                )
                self._pending_tool_calls[tool_call_id] = _PendingToolCall(kind, event.part.tool_name)
                if kind == 'output':
                    # The output tool call is now tracked in `_pending_tool_calls`,
                    # so the `FinalResultEvent` backup used by the error path is no longer needed.
                    self._final_result_event = None
                async for e in self._turn_to('request'):
                    yield e
            elif isinstance(event, AgentRunResultEvent):
                result = cast(AgentRunResult[OutputDataT], event.result)
                self._result = result

                async for e in self._turn_to(None):
                    yield e

                if on_complete is not None:
                    if inspect.isasyncgenfunction(on_complete):
                        async for e in on_complete(result):
                            yield e
                    elif _utils.is_async_callable(on_complete):
                        await on_complete(result)
                    else:
                        await _utils.run_in_executor(on_complete, result)
            elif isinstance(event, FinalResultEvent):
                self._final_result_event = event

            elif isinstance(event, ToolResultEvent):
                tool_call_id = event.part.tool_call_id
                self._pending_tool_calls.pop(tool_call_id, None)

            async for e in self.handle_event(event):
                yield e
    except Exception as exc:  # `exc` to avoid shadowing by `async for e in` below
        # Close any pending tool calls before emitting the error,
        # so the UI doesn't show them as still running.

        # Pending output-tool call (stored via FinalResultEvent if the call event hasn't fired yet)
        if (
            self._final_result_event
            and (tool_call_id := self._final_result_event.tool_call_id)
            and (tool_name := self._final_result_event.tool_name)
        ):
            self._final_result_event = None
            self._pending_tool_calls[tool_call_id] = _PendingToolCall('output', tool_name)

        # Pending tool calls
        for tool_call_id, (kind, tool_name) in self._pending_tool_calls.items():
            async for e in self._turn_to('request'):
                yield e
            error_part = ToolReturnPart(
                tool_call_id=tool_call_id,
                tool_name=tool_name,
                content='Tool execution was interrupted by an error.',
                outcome='failed',
            )
            if kind == 'output':
                async for e in self.handle_output_tool_result(OutputToolResultEvent(error_part)):
                    yield e
            else:
                async for e in self.handle_function_tool_result(FunctionToolResultEvent(error_part)):
                    yield e
        self._pending_tool_calls.clear()

        async for e in self.on_error(exc):
            yield e
    finally:
        async for e in self._turn_to(None):
            yield e

        async for e in self.after_stream():
            yield e

handle_event `async`

handle_event(event: NativeEvent) -> AsyncIterator[EventT]

Transform a Pydantic AI event into one or more protocol-specific events.

This method dispatches to specific handle_* methods based on event type:

PartStartEvent -> handle_part_start()
PartDeltaEvent -> handle_part_delta
PartEndEvent -> handle_part_end
FinalResultEvent -> handle_final_result
FunctionToolCallEvent -> handle_function_tool_call
FunctionToolResultEvent -> handle_function_tool_result
OutputToolCallEvent -> handle_output_tool_call
OutputToolResultEvent -> handle_output_tool_result
AgentRunResultEvent -> handle_run_result

Subclasses are encouraged to override the individual handle_* methods rather than this one. If you need specific behavior for all events, make sure you call the super method.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_event(self, event: NativeEvent) -> AsyncIterator[EventT]:  # noqa: C901
    """Transform a Pydantic AI event into one or more protocol-specific events.

    This method dispatches to specific `handle_*` methods based on event type:

    - [`PartStartEvent`][pydantic_ai.messages.PartStartEvent] -> [`handle_part_start()`][pydantic_ai.ui.UIEventStream.handle_part_start]
    - [`PartDeltaEvent`][pydantic_ai.messages.PartDeltaEvent] -> `handle_part_delta`
    - [`PartEndEvent`][pydantic_ai.messages.PartEndEvent] -> `handle_part_end`
    - [`FinalResultEvent`][pydantic_ai.messages.FinalResultEvent] -> `handle_final_result`
    - [`FunctionToolCallEvent`][pydantic_ai.messages.FunctionToolCallEvent] -> `handle_function_tool_call`
    - [`FunctionToolResultEvent`][pydantic_ai.messages.FunctionToolResultEvent] -> `handle_function_tool_result`
    - [`OutputToolCallEvent`][pydantic_ai.messages.OutputToolCallEvent] -> `handle_output_tool_call`
    - [`OutputToolResultEvent`][pydantic_ai.messages.OutputToolResultEvent] -> `handle_output_tool_result`
    - [`AgentRunResultEvent`][pydantic_ai.run.AgentRunResultEvent] -> `handle_run_result`

    Subclasses are encouraged to override the individual `handle_*` methods rather than this one.
    If you need specific behavior for all events, make sure you call the super method.
    """
    match event:
        case PartStartEvent():
            async for e in self.handle_part_start(event):
                yield e
        case PartDeltaEvent():
            async for e in self.handle_part_delta(event):
                yield e
        case PartEndEvent():
            async for e in self.handle_part_end(event):
                yield e
        case FinalResultEvent():
            async for e in self.handle_final_result(event):
                yield e
        case FunctionToolCallEvent():
            async for e in self.handle_function_tool_call(event):
                yield e
        case FunctionToolResultEvent():
            async for e in self.handle_function_tool_result(event):
                yield e
        case OutputToolCallEvent():
            async for e in self.handle_output_tool_call(event):
                yield e
        case OutputToolResultEvent():
            async for e in self.handle_output_tool_result(event):
                yield e
        case AgentRunResultEvent():
            async for e in self.handle_run_result(event):
                yield e
        case _:
            pass

handle_part_start `async`

handle_part_start(
    event: PartStartEvent,
) -> AsyncIterator[EventT]

Handle a PartStartEvent.

This method dispatches to specific handle_* methods based on part type:

TextPart -> handle_text_start()
ThinkingPart -> handle_thinking_start()
ToolCallPart -> handle_tool_call_start()
NativeToolCallPart -> handle_builtin_tool_call_start()
NativeToolReturnPart -> handle_builtin_tool_return()
FilePart -> handle_file()
CompactionPart -> handle_compaction()

Subclasses are encouraged to override the individual handle_* methods rather than this one. If you need specific behavior for all part start events, make sure you call the super method.

Parameters:

Name	Type	Description	Default
`event`	`PartStartEvent`	The part start event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_part_start(self, event: PartStartEvent) -> AsyncIterator[EventT]:
    """Handle a `PartStartEvent`.

    This method dispatches to specific `handle_*` methods based on part type:

    - [`TextPart`][pydantic_ai.messages.TextPart] -> [`handle_text_start()`][pydantic_ai.ui.UIEventStream.handle_text_start]
    - [`ThinkingPart`][pydantic_ai.messages.ThinkingPart] -> [`handle_thinking_start()`][pydantic_ai.ui.UIEventStream.handle_thinking_start]
    - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] -> [`handle_tool_call_start()`][pydantic_ai.ui.UIEventStream.handle_tool_call_start]
    - [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] -> [`handle_builtin_tool_call_start()`][pydantic_ai.ui.UIEventStream.handle_builtin_tool_call_start]
    - [`NativeToolReturnPart`][pydantic_ai.messages.NativeToolReturnPart] -> [`handle_builtin_tool_return()`][pydantic_ai.ui.UIEventStream.handle_builtin_tool_return]
    - [`FilePart`][pydantic_ai.messages.FilePart] -> [`handle_file()`][pydantic_ai.ui.UIEventStream.handle_file]
    - [`CompactionPart`][pydantic_ai.messages.CompactionPart] -> [`handle_compaction()`][pydantic_ai.ui.UIEventStream.handle_compaction]

    Subclasses are encouraged to override the individual `handle_*` methods rather than this one.
    If you need specific behavior for all part start events, make sure you call the super method.

    Args:
        event: The part start event.
    """
    part = event.part
    previous_part_kind = event.previous_part_kind
    match part:
        case TextPart():
            async for e in self.handle_text_start(part, follows_text=previous_part_kind == 'text'):
                yield e
        case ThinkingPart():
            async for e in self.handle_thinking_start(part, follows_thinking=previous_part_kind == 'thinking'):
                yield e
        case ToolCallPart():
            async for e in self.handle_tool_call_start(part):
                yield e
        case NativeToolCallPart():
            async for e in self.handle_builtin_tool_call_start(part):
                yield e
        case NativeToolReturnPart():
            async for e in self.handle_builtin_tool_return(part):
                yield e
        case FilePart():
            async for e in self.handle_file(part):
                yield e
        case CompactionPart():  # pragma: no cover
            async for e in self.handle_compaction(part):
                yield e

handle_part_delta `async`

handle_part_delta(
    event: PartDeltaEvent,
) -> AsyncIterator[EventT]

Handle a PartDeltaEvent.

This method dispatches to specific handle_*_delta methods based on part delta type:

Subclasses are encouraged to override the individual handle_*_delta methods rather than this one. If you need specific behavior for all part delta events, make sure you call the super method.

Parameters:

Name	Type	Description	Default
`event`	`PartDeltaEvent`	The PartDeltaEvent.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_part_delta(self, event: PartDeltaEvent) -> AsyncIterator[EventT]:
    """Handle a PartDeltaEvent.

    This method dispatches to specific `handle_*_delta` methods based on part delta type:

    - [`TextPartDelta`][pydantic_ai.messages.TextPartDelta] -> [`handle_text_delta()`][pydantic_ai.ui.UIEventStream.handle_text_delta]
    - [`ThinkingPartDelta`][pydantic_ai.messages.ThinkingPartDelta] -> [`handle_thinking_delta()`][pydantic_ai.ui.UIEventStream.handle_thinking_delta]
    - [`ToolCallPartDelta`][pydantic_ai.messages.ToolCallPartDelta] -> [`handle_tool_call_delta()`][pydantic_ai.ui.UIEventStream.handle_tool_call_delta]

    Subclasses are encouraged to override the individual `handle_*_delta` methods rather than this one.
    If you need specific behavior for all part delta events, make sure you call the super method.

    Args:
        event: The PartDeltaEvent.
    """
    delta = event.delta
    match delta:
        case TextPartDelta():
            async for e in self.handle_text_delta(delta):
                yield e
        case ThinkingPartDelta():
            async for e in self.handle_thinking_delta(delta):
                yield e
        case ToolCallPartDelta():  # pragma: no branch
            async for e in self.handle_tool_call_delta(delta):
                yield e

handle_part_end `async`

handle_part_end(
    event: PartEndEvent,
) -> AsyncIterator[EventT]

Handle a PartEndEvent.

This method dispatches to specific handle_*_end methods based on part type:

TextPart -> handle_text_end()
ThinkingPart -> handle_thinking_end()
ToolCallPart -> handle_tool_call_end()
NativeToolCallPart -> handle_builtin_tool_call_end()

Subclasses are encouraged to override the individual handle_*_end methods rather than this one. If you need specific behavior for all part end events, make sure you call the super method.

Parameters:

Name	Type	Description	Default
`event`	`PartEndEvent`	The part end event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_part_end(self, event: PartEndEvent) -> AsyncIterator[EventT]:
    """Handle a `PartEndEvent`.

    This method dispatches to specific `handle_*_end` methods based on part type:

    - [`TextPart`][pydantic_ai.messages.TextPart] -> [`handle_text_end()`][pydantic_ai.ui.UIEventStream.handle_text_end]
    - [`ThinkingPart`][pydantic_ai.messages.ThinkingPart] -> [`handle_thinking_end()`][pydantic_ai.ui.UIEventStream.handle_thinking_end]
    - [`ToolCallPart`][pydantic_ai.messages.ToolCallPart] -> [`handle_tool_call_end()`][pydantic_ai.ui.UIEventStream.handle_tool_call_end]
    - [`NativeToolCallPart`][pydantic_ai.messages.NativeToolCallPart] -> [`handle_builtin_tool_call_end()`][pydantic_ai.ui.UIEventStream.handle_builtin_tool_call_end]

    Subclasses are encouraged to override the individual `handle_*_end` methods rather than this one.
    If you need specific behavior for all part end events, make sure you call the super method.

    Args:
        event: The part end event.
    """
    part = event.part
    next_part_kind = event.next_part_kind
    match part:
        case TextPart():
            async for e in self.handle_text_end(part, followed_by_text=next_part_kind == 'text'):
                yield e
        case ThinkingPart():
            async for e in self.handle_thinking_end(part, followed_by_thinking=next_part_kind == 'thinking'):
                yield e
        case ToolCallPart():
            async for e in self.handle_tool_call_end(part):
                yield e
        case NativeToolCallPart():
            async for e in self.handle_builtin_tool_call_end(part):
                yield e
        case NativeToolReturnPart() | FilePart() | CompactionPart():  # pragma: no cover
            # These don't have deltas, so they don't need to be ended.
            pass

before_stream `async`

before_stream() -> AsyncIterator[EventT]

Yield events before agent streaming starts.

This hook is called before any agent events are processed. Override this to inject custom events at the start of the stream.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def before_stream(self) -> AsyncIterator[EventT]:
    """Yield events before agent streaming starts.

    This hook is called before any agent events are processed.
    Override this to inject custom events at the start of the stream.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

after_stream `async`

after_stream() -> AsyncIterator[EventT]

Yield events after agent streaming completes.

This hook is called after all agent events have been processed. Override this to inject custom events at the end of the stream.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def after_stream(self) -> AsyncIterator[EventT]:
    """Yield events after agent streaming completes.

    This hook is called after all agent events have been processed.
    Override this to inject custom events at the end of the stream.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

on_error `async`

on_error(error: Exception) -> AsyncIterator[EventT]

Handle errors that occur during streaming.

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The error that occurred during streaming.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def on_error(self, error: Exception) -> AsyncIterator[EventT]:
    """Handle errors that occur during streaming.

    Args:
        error: The error that occurred during streaming.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

before_request `async`

before_request() -> AsyncIterator[EventT]

Yield events before a model request is processed.

Override this to inject custom events at the start of the request.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def before_request(self) -> AsyncIterator[EventT]:
    """Yield events before a model request is processed.

    Override this to inject custom events at the start of the request.
    """
    return  # pragma: lax no cover
    yield  # Make this an async generator

after_request `async`

after_request() -> AsyncIterator[EventT]

Yield events after a model request is processed.

Override this to inject custom events at the end of the request.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def after_request(self) -> AsyncIterator[EventT]:
    """Yield events after a model request is processed.

    Override this to inject custom events at the end of the request.
    """
    return  # pragma: lax no cover
    yield  # Make this an async generator

before_response `async`

before_response() -> AsyncIterator[EventT]

Yield events before a model response is processed.

Override this to inject custom events at the start of the response.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def before_response(self) -> AsyncIterator[EventT]:
    """Yield events before a model response is processed.

    Override this to inject custom events at the start of the response.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

after_response `async`

after_response() -> AsyncIterator[EventT]

Yield events after a model response is processed.

Override this to inject custom events at the end of the response.

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def after_response(self) -> AsyncIterator[EventT]:
    """Yield events after a model response is processed.

    Override this to inject custom events at the end of the response.
    """
    return  # pragma: lax no cover
    yield  # Make this an async generator

handle_text_start `async`

handle_text_start(
    part: TextPart, follows_text: bool = False
) -> AsyncIterator[EventT]

Handle the start of a TextPart.

Parameters:

Name	Type	Description	Default
`part`	`TextPart`	The text part.	required
`follows_text`	`bool`	Whether the part is directly preceded by another text part. In this case, you may want to yield a "text-delta" event instead of a "text-start" event.	`False`

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_text_start(self, part: TextPart, follows_text: bool = False) -> AsyncIterator[EventT]:
    """Handle the start of a `TextPart`.

    Args:
        part: The text part.
        follows_text: Whether the part is directly preceded by another text part. In this case, you may want to yield a "text-delta" event instead of a "text-start" event.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_text_delta `async`

handle_text_delta(
    delta: TextPartDelta,
) -> AsyncIterator[EventT]

Handle a TextPartDelta.

Parameters:

Name	Type	Description	Default
`delta`	`TextPartDelta`	The text part delta.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_text_delta(self, delta: TextPartDelta) -> AsyncIterator[EventT]:
    """Handle a `TextPartDelta`.

    Args:
        delta: The text part delta.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_text_end `async`

handle_text_end(
    part: TextPart, followed_by_text: bool = False
) -> AsyncIterator[EventT]

Handle the end of a TextPart.

Parameters:

Name	Type	Description	Default
`part`	`TextPart`	The text part.	required
`followed_by_text`	`bool`	Whether the part is directly followed by another text part. In this case, you may not want to yield a "text-end" event yet.	`False`

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_text_end(self, part: TextPart, followed_by_text: bool = False) -> AsyncIterator[EventT]:
    """Handle the end of a `TextPart`.

    Args:
        part: The text part.
        followed_by_text: Whether the part is directly followed by another text part. In this case, you may not want to yield a "text-end" event yet.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_thinking_start `async`

handle_thinking_start(
    part: ThinkingPart, follows_thinking: bool = False
) -> AsyncIterator[EventT]

Handle the start of a ThinkingPart.

Parameters:

Name	Type	Description	Default
`part`	`ThinkingPart`	The thinking part.	required
`follows_thinking`	`bool`	Whether the part is directly preceded by another thinking part. In this case, you may want to yield a "thinking-delta" event instead of a "thinking-start" event.	`False`

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_thinking_start(self, part: ThinkingPart, follows_thinking: bool = False) -> AsyncIterator[EventT]:
    """Handle the start of a `ThinkingPart`.

    Args:
        part: The thinking part.
        follows_thinking: Whether the part is directly preceded by another thinking part. In this case, you may want to yield a "thinking-delta" event instead of a "thinking-start" event.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_thinking_delta `async`

handle_thinking_delta(
    delta: ThinkingPartDelta,
) -> AsyncIterator[EventT]

Handle a ThinkingPartDelta.

Parameters:

Name	Type	Description	Default
`delta`	`ThinkingPartDelta`	The thinking part delta.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_thinking_delta(self, delta: ThinkingPartDelta) -> AsyncIterator[EventT]:
    """Handle a `ThinkingPartDelta`.

    Args:
        delta: The thinking part delta.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_thinking_end `async`

handle_thinking_end(
    part: ThinkingPart, followed_by_thinking: bool = False
) -> AsyncIterator[EventT]

Handle the end of a ThinkingPart.

Parameters:

Name	Type	Description	Default
`part`	`ThinkingPart`	The thinking part.	required
`followed_by_thinking`	`bool`	Whether the part is directly followed by another thinking part. In this case, you may not want to yield a "thinking-end" event yet.	`False`

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_thinking_end(
    self, part: ThinkingPart, followed_by_thinking: bool = False
) -> AsyncIterator[EventT]:
    """Handle the end of a `ThinkingPart`.

    Args:
        part: The thinking part.
        followed_by_thinking: Whether the part is directly followed by another thinking part. In this case, you may not want to yield a "thinking-end" event yet.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_tool_call_start `async`

handle_tool_call_start(
    part: ToolCallPart,
) -> AsyncIterator[EventT]

Handle the start of a ToolCallPart.

Parameters:

Name	Type	Description	Default
`part`	`ToolCallPart`	The tool call part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_tool_call_start(self, part: ToolCallPart) -> AsyncIterator[EventT]:
    """Handle the start of a `ToolCallPart`.

    Args:
        part: The tool call part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_tool_call_delta `async`

handle_tool_call_delta(
    delta: ToolCallPartDelta,
) -> AsyncIterator[EventT]

Handle a ToolCallPartDelta.

Parameters:

Name	Type	Description	Default
`delta`	`ToolCallPartDelta`	The tool call part delta.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_tool_call_delta(self, delta: ToolCallPartDelta) -> AsyncIterator[EventT]:
    """Handle a `ToolCallPartDelta`.

    Args:
        delta: The tool call part delta.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_tool_call_end `async`

handle_tool_call_end(
    part: ToolCallPart,
) -> AsyncIterator[EventT]

Handle the end of a ToolCallPart.

Parameters:

Name	Type	Description	Default
`part`	`ToolCallPart`	The tool call part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_tool_call_end(self, part: ToolCallPart) -> AsyncIterator[EventT]:
    """Handle the end of a `ToolCallPart`.

    Args:
        part: The tool call part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_builtin_tool_call_start `async`

handle_builtin_tool_call_start(
    part: NativeToolCallPart,
) -> AsyncIterator[EventT]

Handle a NativeToolCallPart at start.

Parameters:

Name	Type	Description	Default
`part`	`NativeToolCallPart`	The builtin tool call part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_builtin_tool_call_start(self, part: NativeToolCallPart) -> AsyncIterator[EventT]:
    """Handle a `NativeToolCallPart` at start.

    Args:
        part: The builtin tool call part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_builtin_tool_call_end `async`

handle_builtin_tool_call_end(
    part: NativeToolCallPart,
) -> AsyncIterator[EventT]

Handle the end of a NativeToolCallPart.

Parameters:

Name	Type	Description	Default
`part`	`NativeToolCallPart`	The builtin tool call part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_builtin_tool_call_end(self, part: NativeToolCallPart) -> AsyncIterator[EventT]:
    """Handle the end of a `NativeToolCallPart`.

    Args:
        part: The builtin tool call part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_builtin_tool_return `async`

handle_builtin_tool_return(
    part: NativeToolReturnPart,
) -> AsyncIterator[EventT]

Handle a NativeToolReturnPart.

Parameters:

Name	Type	Description	Default
`part`	`NativeToolReturnPart`	The builtin tool return part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_builtin_tool_return(self, part: NativeToolReturnPart) -> AsyncIterator[EventT]:
    """Handle a `NativeToolReturnPart`.

    Args:
        part: The builtin tool return part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_file `async`

handle_file(part: FilePart) -> AsyncIterator[EventT]

Handle a FilePart.

Parameters:

Name	Type	Description	Default
`part`	`FilePart`	The file part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_file(self, part: FilePart) -> AsyncIterator[EventT]:
    """Handle a `FilePart`.

    Args:
        part: The file part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_compaction `async`

handle_compaction(
    part: CompactionPart,
) -> AsyncIterator[EventT]

Handle a CompactionPart.

Parameters:

Name	Type	Description	Default
`part`	`CompactionPart`	The compaction part.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_compaction(self, part: CompactionPart) -> AsyncIterator[EventT]:
    """Handle a `CompactionPart`.

    Args:
        part: The compaction part.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_final_result `async`

handle_final_result(
    event: FinalResultEvent,
) -> AsyncIterator[EventT]

Handle a FinalResultEvent.

Parameters:

Name	Type	Description	Default
`event`	`FinalResultEvent`	The final result event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_final_result(self, event: FinalResultEvent) -> AsyncIterator[EventT]:
    """Handle a `FinalResultEvent`.

    Args:
        event: The final result event.
    """
    return
    yield  # Make this an async generator

handle_function_tool_call `async`

handle_function_tool_call(
    event: FunctionToolCallEvent,
) -> AsyncIterator[EventT]

Handle a FunctionToolCallEvent.

Parameters:

Name	Type	Description	Default
`event`	`FunctionToolCallEvent`	The function tool call event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_function_tool_call(self, event: FunctionToolCallEvent) -> AsyncIterator[EventT]:
    """Handle a `FunctionToolCallEvent`.

    Args:
        event: The function tool call event.
    """
    return
    yield  # Make this an async generator

handle_function_tool_result `async`

handle_function_tool_result(
    event: FunctionToolResultEvent,
) -> AsyncIterator[EventT]

Handle a FunctionToolResultEvent.

Parameters:

Name	Type	Description	Default
`event`	`FunctionToolResultEvent`	The function tool result event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_function_tool_result(self, event: FunctionToolResultEvent) -> AsyncIterator[EventT]:
    """Handle a `FunctionToolResultEvent`.

    Args:
        event: The function tool result event.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_output_tool_call `async`

handle_output_tool_call(
    event: OutputToolCallEvent,
) -> AsyncIterator[EventT]

Handle an OutputToolCallEvent (the model's "submit final answer" call).

Parameters:

Name	Type	Description	Default
`event`	`OutputToolCallEvent`	The output tool call event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_output_tool_call(self, event: OutputToolCallEvent) -> AsyncIterator[EventT]:
    """Handle an `OutputToolCallEvent` (the model's "submit final answer" call).

    Args:
        event: The output tool call event.
    """
    return
    yield  # Make this an async generator

handle_output_tool_result `async`

handle_output_tool_result(
    event: OutputToolResultEvent,
) -> AsyncIterator[EventT]

Handle an OutputToolResultEvent (the result of an output tool call).

Parameters:

Name	Type	Description	Default
`event`	`OutputToolResultEvent`	The output tool result event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_output_tool_result(self, event: OutputToolResultEvent) -> AsyncIterator[EventT]:
    """Handle an `OutputToolResultEvent` (the result of an output tool call).

    Args:
        event: The output tool result event.
    """
    return  # pragma: no cover
    yield  # Make this an async generator

handle_run_result `async`

handle_run_result(
    event: AgentRunResultEvent,
) -> AsyncIterator[EventT]

Handle an AgentRunResultEvent.

Parameters:

Name	Type	Description	Default
`event`	`AgentRunResultEvent`	The agent run result event.	required

Source code in pydantic_ai_slim/pydantic_ai/ui/_event_stream.py

async def handle_run_result(self, event: AgentRunResultEvent) -> AsyncIterator[EventT]:
    """Handle an `AgentRunResultEvent`.

    Args:
        event: The agent run result event.
    """
    return
    yield  # Make this an async generator

MessagesBuilder `dataclass`

Helper class to build Pydantic AI messages from request/response parts.

Source code in pydantic_ai_slim/pydantic_ai/ui/_messages_builder.py

@dataclass
class MessagesBuilder:
    """Helper class to build Pydantic AI messages from request/response parts."""

    messages: list[ModelMessage] = field(default_factory=list[ModelMessage])

    def add(self, part: ModelRequestPart | ModelResponsePart) -> None:
        """Add a new part, creating a new request or response message if necessary."""
        last_message = self.messages[-1] if self.messages else None
        if isinstance(part, get_union_args(ModelRequestPart)):
            part = cast(ModelRequestPart, part)
            if isinstance(last_message, ModelRequest):
                last_message.parts = [*last_message.parts, part]
            else:
                self.messages.append(ModelRequest(parts=[part]))
        else:
            part = cast(ModelResponsePart, part)
            if isinstance(last_message, ModelResponse):
                last_message.parts = [*last_message.parts, part]
            else:
                self.messages.append(ModelResponse(parts=[part]))

add

add(part: ModelRequestPart | ModelResponsePart) -> None

Add a new part, creating a new request or response message if necessary.

Source code in pydantic_ai_slim/pydantic_ai/ui/_messages_builder.py

def add(self, part: ModelRequestPart | ModelResponsePart) -> None:
    """Add a new part, creating a new request or response message if necessary."""
    last_message = self.messages[-1] if self.messages else None
    if isinstance(part, get_union_args(ModelRequestPart)):
        part = cast(ModelRequestPart, part)
        if isinstance(last_message, ModelRequest):
            last_message.parts = [*last_message.parts, part]
        else:
            self.messages.append(ModelRequest(parts=[part]))
    else:
        part = cast(ModelResponsePart, part)
        if isinstance(last_message, ModelResponse):
            last_message.parts = [*last_message.parts, part]
        else:
            self.messages.append(ModelResponse(parts=[part]))

pydantic_ai.ui

StateDeps dataclass

StateHandler

state property writable

UIAdapter dataclass

agent instance-attribute

run_input instance-attribute

accept class-attribute instance-attribute

manage_system_prompt class-attribute instance-attribute

allowed_file_url_schemes class-attribute instance-attribute

from_request async classmethod

build_run_input abstractmethod classmethod

load_messages abstractmethod classmethod

dump_messages classmethod

build_event_stream abstractmethod

messages abstractmethod cached property

toolset cached property

state cached property

deferred_tool_results cached property

conversation_id cached property

sanitize_messages

transform_stream

encode_stream

streaming_response

run_stream_native

run_stream

dispatch_request async classmethod

SSE_CONTENT_TYPE module-attribute

NativeEvent module-attribute

OnCompleteFunc module-attribute

UIEventStream dataclass

accept class-attribute instance-attribute

message_id class-attribute instance-attribute

new_message_id

response_headers property

content_type property

encode_event abstractmethod

encode_stream async

streaming_response

transform_stream async

handle_event async

handle_part_start async

handle_part_delta async

handle_part_end async

before_stream async

after_stream async

on_error async

before_request async

after_request async

before_response async

after_response async

handle_text_start async

handle_text_delta async

handle_text_end async

handle_thinking_start async

handle_thinking_delta async

handle_thinking_end async

handle_tool_call_start async

handle_tool_call_delta async

handle_tool_call_end async

handle_builtin_tool_call_start async

handle_builtin_tool_call_end async

handle_builtin_tool_return async

handle_file async

handle_compaction async

handle_final_result async

handle_function_tool_call async

handle_function_tool_result async

handle_output_tool_call async

handle_output_tool_result async

handle_run_result async

MessagesBuilder dataclass

add

`pydantic_ai.ui`

StateDeps `dataclass`

state `property` `writable`

UIAdapter `dataclass`

agent `instance-attribute`

run_input `instance-attribute`

accept `class-attribute` `instance-attribute`

manage_system_prompt `class-attribute` `instance-attribute`

allowed_file_url_schemes `class-attribute` `instance-attribute`

from_request `async` `classmethod`

build_run_input `abstractmethod` `classmethod`

load_messages `abstractmethod` `classmethod`

dump_messages `classmethod`

build_event_stream `abstractmethod`

messages `abstractmethod` `cached` `property`

toolset `cached` `property`

state `cached` `property`

deferred_tool_results `cached` `property`

conversation_id `cached` `property`

dispatch_request `async` `classmethod`

SSE_CONTENT_TYPE `module-attribute`

NativeEvent `module-attribute`

OnCompleteFunc `module-attribute`

UIEventStream `dataclass`

accept `class-attribute` `instance-attribute`

message_id `class-attribute` `instance-attribute`

response_headers `property`

content_type `property`

encode_event `abstractmethod`

encode_stream `async`

transform_stream `async`

handle_event `async`

handle_part_start `async`

handle_part_delta `async`

handle_part_end `async`

before_stream `async`

after_stream `async`

on_error `async`

before_request `async`

after_request `async`

before_response `async`

after_response `async`

handle_text_start `async`

handle_text_delta `async`

handle_text_end `async`

handle_thinking_start `async`

handle_thinking_delta `async`

handle_thinking_end `async`

handle_tool_call_start `async`

handle_tool_call_delta `async`

handle_tool_call_end `async`

handle_builtin_tool_call_start `async`

handle_builtin_tool_call_end `async`

handle_builtin_tool_return `async`

handle_file `async`

handle_compaction `async`

handle_final_result `async`

handle_function_tool_call `async`

handle_function_tool_result `async`

handle_output_tool_call `async`

handle_output_tool_result `async`

handle_run_result `async`

MessagesBuilder `dataclass`