`pydantic_ai.ui.vercel_ai`

Vercel AI protocol adapter for Pydantic AI agents.

This module provides classes for integrating Pydantic AI agents with the Vercel AI protocol, enabling streaming event-based communication for interactive AI applications.

Converted to Python from: https://github.com/vercel/ai/blob/ai%405.0.34/packages/ai/src/ui/ui-messages.ts

VercelAIAdapter `dataclass`

Bases: UIAdapter[RequestData, UIMessage, BaseChunk, AgentDepsT, OutputDataT]

UI adapter for the Vercel AI protocol.

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

@dataclass
class VercelAIAdapter(UIAdapter[RequestData, UIMessage, BaseChunk, AgentDepsT, OutputDataT]):
    """UI adapter for the Vercel AI protocol."""

    _: KW_ONLY
    sdk_version: Literal[5, 6] = 5
    """Vercel AI SDK version to target. Default is 5 for backwards compatibility.

    Setting `sdk_version=6` enables tool approval streaming for human-in-the-loop workflows.
    """

    @classmethod
    def build_run_input(cls, body: bytes) -> RequestData:
        """Build a Vercel AI run input object from the request body."""
        return request_data_ta.validate_json(body)

    @classmethod
    async def from_request(
        cls,
        request: Request,
        *,
        agent: AbstractAgent[AgentDepsT, OutputDataT],
        sdk_version: Literal[5, 6] = 5,
        **kwargs: Any,
    ) -> VercelAIAdapter[AgentDepsT, OutputDataT]:
        """Extends [`from_request`][pydantic_ai.ui.UIAdapter.from_request] with the `sdk_version` parameter."""
        return await super().from_request(request, agent=agent, sdk_version=sdk_version, **kwargs)

    @classmethod
    async def dispatch_request(
        cls,
        request: Request,
        *,
        agent: AbstractAgent[DispatchDepsT, DispatchOutputDataT],
        sdk_version: Literal[5, 6] = 5,
        message_history: Sequence[ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        model: Model | KnownModelName | str | None = None,
        instructions: Instructions[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,
        builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
        on_complete: OnCompleteFunc[BaseChunk] | None = None,
        **kwargs: Any,
    ) -> Response:
        """Extends [`dispatch_request`][pydantic_ai.ui.UIAdapter.dispatch_request] with the `sdk_version` parameter."""
        return await super().dispatch_request(
            request,
            agent=agent,
            sdk_version=sdk_version,
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
            model=model,
            instructions=instructions,
            deps=deps,
            output_type=output_type,
            model_settings=model_settings,
            usage_limits=usage_limits,
            usage=usage,
            metadata=metadata,
            infer_name=infer_name,
            toolsets=toolsets,
            builtin_tools=builtin_tools,
            on_complete=on_complete,
            **kwargs,
        )

    def build_event_stream(self) -> UIEventStream[RequestData, BaseChunk, AgentDepsT, OutputDataT]:
        """Build a Vercel AI event stream transformer."""
        return VercelAIEventStream(self.run_input, accept=self.accept, sdk_version=self.sdk_version)

    @cached_property
    def deferred_tool_results(self) -> DeferredToolResults | None:
        """Extract deferred tool results from Vercel AI messages with approval responses."""
        if self.sdk_version < 6:
            return None
        approvals: dict[str, bool | DeferredToolApprovalResult] = {}
        for tool_call_id, approval in iter_tool_approval_responses(self.run_input.messages):
            if approval.approved:
                approvals[tool_call_id] = True
            elif approval.reason:
                approvals[tool_call_id] = ToolDenied(message=approval.reason)
            else:
                approvals[tool_call_id] = False
        return DeferredToolResults(approvals=approvals) if approvals else None

    @cached_property
    def messages(self) -> list[ModelMessage]:
        """Pydantic AI messages from the Vercel AI run input."""
        return self.load_messages(self.run_input.messages)

    @classmethod
    def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  # noqa: C901
        """Transform Vercel AI messages into Pydantic AI messages."""
        builder = MessagesBuilder()

        for msg in messages:
            if msg.role == 'system':
                for part in msg.parts:
                    if isinstance(part, TextUIPart):
                        builder.add(SystemPromptPart(content=part.text))
                    else:  # pragma: no cover
                        raise ValueError(f'Unsupported system message part type: {type(part)}')
            elif msg.role == 'user':
                user_prompt_content: str | list[UserContent] = []
                for part in msg.parts:
                    if isinstance(part, TextUIPart):
                        user_prompt_content.append(part.text)
                    elif isinstance(part, FileUIPart):
                        try:
                            file = BinaryContent.from_data_uri(part.url)
                        except ValueError:
                            media_type_prefix = part.media_type.split('/', 1)[0]
                            match media_type_prefix:
                                case 'image':
                                    file = ImageUrl(url=part.url, media_type=part.media_type)
                                case 'video':
                                    file = VideoUrl(url=part.url, media_type=part.media_type)
                                case 'audio':
                                    file = AudioUrl(url=part.url, media_type=part.media_type)
                                case _:
                                    file = DocumentUrl(url=part.url, media_type=part.media_type)
                        user_prompt_content.append(file)
                    else:  # pragma: no cover
                        raise ValueError(f'Unsupported user message part type: {type(part)}')

                if user_prompt_content:  # pragma: no branch
                    if len(user_prompt_content) == 1 and isinstance(user_prompt_content[0], str):
                        user_prompt_content = user_prompt_content[0]
                    builder.add(UserPromptPart(content=user_prompt_content))

            elif msg.role == 'assistant':
                for part in msg.parts:
                    if isinstance(part, TextUIPart):
                        provider_meta = load_provider_metadata(part.provider_metadata)
                        builder.add(
                            TextPart(
                                content=part.text,
                                id=provider_meta.get('id'),
                                provider_name=provider_meta.get('provider_name'),
                                provider_details=provider_meta.get('provider_details'),
                            )
                        )
                    elif isinstance(part, ReasoningUIPart):
                        provider_meta = load_provider_metadata(part.provider_metadata)
                        builder.add(
                            ThinkingPart(
                                content=part.text,
                                id=provider_meta.get('id'),
                                signature=provider_meta.get('signature'),
                                provider_name=provider_meta.get('provider_name'),
                                provider_details=provider_meta.get('provider_details'),
                            )
                        )
                    elif isinstance(part, FileUIPart):
                        try:
                            file = BinaryContent.from_data_uri(part.url)
                        except ValueError as e:  # pragma: no cover
                            # We don't yet handle non-data-URI file URLs returned by assistants, as no Pydantic AI models do this.
                            raise ValueError(
                                'Vercel AI integration can currently only handle assistant file parts with data URIs.'
                            ) from e
                        provider_meta = load_provider_metadata(part.provider_metadata)
                        builder.add(
                            FilePart(
                                content=file,
                                id=provider_meta.get('id'),
                                provider_name=provider_meta.get('provider_name'),
                                provider_details=provider_meta.get('provider_details'),
                            )
                        )
                    elif isinstance(part, ToolUIPart | DynamicToolUIPart):
                        if isinstance(part, DynamicToolUIPart):
                            tool_name = part.tool_name
                            builtin_tool = False
                        else:
                            tool_name = part.type.removeprefix('tool-')
                            builtin_tool = part.provider_executed

                        tool_call_id = part.tool_call_id

                        args: str | dict[str, Any] | None = part.input

                        if isinstance(args, str):
                            try:
                                parsed = json.loads(args)
                                if isinstance(parsed, dict):
                                    args = cast(dict[str, Any], parsed)
                            except json.JSONDecodeError:
                                pass
                        elif isinstance(args, dict) or args is None:
                            pass
                        else:
                            assert_never(args)

                        provider_meta = load_provider_metadata(part.call_provider_metadata)
                        part_id = provider_meta.get('id')
                        provider_name = provider_meta.get('provider_name')
                        provider_details = provider_meta.get('provider_details')

                        if builtin_tool:
                            # For builtin tools, we need to create 2 parts (BuiltinToolCall & BuiltinToolReturn) for a single Vercel ToolOutput
                            # The call and return metadata are combined in the output part.
                            # So we extract and return them to the respective parts
                            call_meta = return_meta = {}
                            has_tool_output = isinstance(part, (ToolOutputAvailablePart, ToolOutputErrorPart))

                            if has_tool_output:
                                call_meta, return_meta = cls._load_builtin_tool_meta(provider_meta)

                            builder.add(
                                BuiltinToolCallPart(
                                    tool_name=tool_name,
                                    tool_call_id=tool_call_id,
                                    args=args,
                                    id=call_meta.get('id') or part_id,
                                    provider_name=call_meta.get('provider_name') or provider_name,
                                    provider_details=call_meta.get('provider_details') or provider_details,
                                )
                            )

                            if has_tool_output:
                                output: Any | None = None
                                if isinstance(part, ToolOutputAvailablePart):
                                    output = part.output
                                elif isinstance(part, ToolOutputErrorPart):  # pragma: no branch
                                    output = {'error_text': part.error_text, 'is_error': True}
                                builder.add(
                                    BuiltinToolReturnPart(
                                        tool_name=tool_name,
                                        tool_call_id=tool_call_id,
                                        content=output,
                                        provider_name=return_meta.get('provider_name') or provider_name,
                                        provider_details=return_meta.get('provider_details') or provider_details,
                                    )
                                )
                        else:
                            builder.add(
                                ToolCallPart(
                                    tool_name=tool_name,
                                    tool_call_id=tool_call_id,
                                    args=args,
                                    id=part_id,
                                    provider_name=provider_name,
                                    provider_details=provider_details,
                                )
                            )

                            if part.state == 'output-available':
                                builder.add(
                                    ToolReturnPart(tool_name=tool_name, tool_call_id=tool_call_id, content=part.output)
                                )
                            elif part.state == 'output-error':
                                builder.add(
                                    RetryPromptPart(
                                        tool_name=tool_name, tool_call_id=tool_call_id, content=part.error_text
                                    )
                                )
                    elif isinstance(part, DataUIPart):  # pragma: no cover
                        # Contains custom data that shouldn't be sent to the model
                        pass
                    elif isinstance(part, SourceUrlUIPart):  # pragma: no cover
                        # TODO: Once we support citations: https://github.com/pydantic/pydantic-ai/issues/3126
                        pass
                    elif isinstance(part, SourceDocumentUIPart):  # pragma: no cover
                        # TODO: Once we support citations: https://github.com/pydantic/pydantic-ai/issues/3126
                        pass
                    elif isinstance(part, StepStartUIPart):  # pragma: no cover
                        # Nothing to do here
                        pass
                    else:
                        assert_never(part)
            else:
                assert_never(msg.role)

        return builder.messages

    @staticmethod
    def _dump_builtin_tool_meta(
        call_provider_metadata: ProviderMetadata | None, return_provider_metadata: ProviderMetadata | None
    ) -> ProviderMetadata | None:
        """Use special keys (call_meta and return_meta) to dump combined provider metadata."""
        return dump_provider_metadata(call_meta=call_provider_metadata, return_meta=return_provider_metadata)

    @staticmethod
    def _load_builtin_tool_meta(
        provider_metadata: ProviderMetadata,
    ) -> tuple[dict[str, Any], dict[str, Any]]:
        """Use special keys (call_meta and return_meta) to load combined provider metadata."""
        return provider_metadata.get('call_meta') or {}, provider_metadata.get('return_meta') or {}

    @staticmethod
    def _dump_request_message(msg: ModelRequest) -> tuple[list[UIMessagePart], list[UIMessagePart]]:
        """Convert a ModelRequest into a UIMessage."""
        system_ui_parts: list[UIMessagePart] = []
        user_ui_parts: list[UIMessagePart] = []

        for part in msg.parts:
            if isinstance(part, SystemPromptPart):
                system_ui_parts.append(TextUIPart(text=part.content, state='done'))
            elif isinstance(part, UserPromptPart):
                user_ui_parts.extend(_convert_user_prompt_part(part))
            elif isinstance(part, ToolReturnPart):
                # Tool returns are merged into the tool call in the assistant message
                pass
            elif isinstance(part, RetryPromptPart):
                if part.tool_name:
                    # Tool-related retries are handled when processing ToolCallPart in ModelResponse
                    pass
                else:
                    # Non-tool retries (e.g., output validation errors) become user text
                    user_ui_parts.append(TextUIPart(text=part.model_response(), state='done'))
            else:
                assert_never(part)

        return system_ui_parts, user_ui_parts

    @classmethod
    def _dump_response_message(
        cls, msg: ModelResponse, tool_results: dict[str, ToolReturnPart | RetryPromptPart]
    ) -> list[UIMessagePart]:
        """Convert a ModelResponse into a UIMessage."""
        ui_parts: list[UIMessagePart] = []

        # For builtin tools, returns can be in the same ModelResponse as calls
        local_builtin_returns: dict[str, BuiltinToolReturnPart] = {
            part.tool_call_id: part for part in msg.parts if isinstance(part, BuiltinToolReturnPart)
        }

        for part in msg.parts:
            if isinstance(part, BuiltinToolReturnPart):
                continue
            elif isinstance(part, TextPart):
                # Combine consecutive text parts
                if ui_parts and isinstance(ui_parts[-1], TextUIPart):
                    ui_parts[-1].text += part.content
                else:
                    provider_metadata = dump_provider_metadata(
                        id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
                    )
                    ui_parts.append(TextUIPart(text=part.content, state='done', provider_metadata=provider_metadata))
            elif isinstance(part, ThinkingPart):
                provider_metadata = dump_provider_metadata(
                    id=part.id,
                    signature=part.signature,
                    provider_name=part.provider_name,
                    provider_details=part.provider_details,
                )
                ui_parts.append(ReasoningUIPart(text=part.content, state='done', provider_metadata=provider_metadata))
            elif isinstance(part, FilePart):
                ui_parts.append(
                    FileUIPart(
                        url=part.content.data_uri,
                        media_type=part.content.media_type,
                        provider_metadata=dump_provider_metadata(
                            id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
                        ),
                    )
                )
            elif isinstance(part, BuiltinToolCallPart):
                tool_name = f'tool-{part.tool_name}'
                if builtin_return := local_builtin_returns.get(part.tool_call_id):
                    # Builtin tool calls are represented by two parts in pydantic_ai:
                    #   1. BuiltinToolCallPart (the tool request) -> part
                    #   2. BuiltinToolReturnPart (the tool's output) -> builtin_return
                    # The Vercel AI SDK only has a single ToolOutputPart (ToolOutputAvailablePart or ToolOutputErrorPart).
                    # So, we need to combine the metadata so that when we later convert back from Vercel AI to pydantic_ai,
                    # we can properly reconstruct both the call and return parts with their respective metadata.
                    # Note: This extra metadata handling is only needed for built-in tools, since normal tool returns
                    # (ToolReturnPart) do not include provider metadata.

                    call_meta = dump_provider_metadata(
                        wrapper_key=None,
                        id=part.id,
                        provider_name=part.provider_name,
                        provider_details=part.provider_details,
                    )
                    return_meta = dump_provider_metadata(
                        wrapper_key=None,
                        provider_name=builtin_return.provider_name,
                        provider_details=builtin_return.provider_details,
                    )
                    combined_provider_meta = cls._dump_builtin_tool_meta(call_meta, return_meta)

                    response_object = builtin_return.model_response_object()
                    # These `is_error`/`error_text` fields are only present when the BuiltinToolReturnPart
                    # was parsed from an incoming VercelAI request. We can't detect errors for other sources
                    # until BuiltinToolReturnPart has standardized error fields (see https://github.com/pydantic/pydantic-ai/issues/3561).3
                    if response_object.get('is_error') is True and (
                        (error_text := response_object.get('error_text')) is not None
                    ):
                        ui_parts.append(
                            ToolOutputErrorPart(
                                type=tool_name,
                                tool_call_id=part.tool_call_id,
                                input=part.args_as_json_str(),
                                error_text=error_text,
                                state='output-error',
                                provider_executed=True,
                                call_provider_metadata=combined_provider_meta,
                            )
                        )
                    else:
                        content = builtin_return.model_response_str()
                        ui_parts.append(
                            ToolOutputAvailablePart(
                                type=tool_name,
                                tool_call_id=part.tool_call_id,
                                input=part.args_as_json_str(),
                                output=content,
                                state='output-available',
                                provider_executed=True,
                                call_provider_metadata=combined_provider_meta,
                            )
                        )
                else:
                    call_provider_metadata = dump_provider_metadata(
                        id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
                    )
                    ui_parts.append(
                        ToolInputAvailablePart(
                            type=tool_name,
                            tool_call_id=part.tool_call_id,
                            input=part.args_as_json_str(),
                            state='input-available',
                            provider_executed=True,
                            call_provider_metadata=call_provider_metadata,
                        )
                    )
            elif isinstance(part, ToolCallPart):
                tool_result = tool_results.get(part.tool_call_id)
                call_provider_metadata = dump_provider_metadata(
                    id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
                )

                if isinstance(tool_result, ToolReturnPart):
                    content = tool_result.model_response_str()
                    ui_parts.append(
                        DynamicToolOutputAvailablePart(
                            tool_name=part.tool_name,
                            tool_call_id=part.tool_call_id,
                            input=part.args_as_json_str(),
                            output=content,
                            state='output-available',
                            call_provider_metadata=call_provider_metadata,
                        )
                    )
                    # Check for Vercel AI chunks returned by tool calls via metadata.
                    ui_parts.extend(_extract_metadata_ui_parts(tool_result))
                elif isinstance(tool_result, RetryPromptPart):
                    error_text = tool_result.model_response()
                    ui_parts.append(
                        DynamicToolOutputErrorPart(
                            tool_name=part.tool_name,
                            tool_call_id=part.tool_call_id,
                            input=part.args_as_json_str(),
                            error_text=error_text,
                            state='output-error',
                            call_provider_metadata=call_provider_metadata,
                        )
                    )
                else:
                    ui_parts.append(
                        DynamicToolInputAvailablePart(
                            tool_name=part.tool_name,
                            tool_call_id=part.tool_call_id,
                            input=part.args_as_json_str(),
                            state='input-available',
                            call_provider_metadata=call_provider_metadata,
                        )
                    )
            else:
                assert_never(part)

        return ui_parts

    @classmethod
    def dump_messages(
        cls,
        messages: Sequence[ModelMessage],
    ) -> list[UIMessage]:
        """Transform Pydantic AI messages into Vercel AI messages.

        Args:
            messages: A sequence of ModelMessage objects to convert

        Returns:
            A list of UIMessage objects in Vercel AI format
        """
        tool_results: dict[str, ToolReturnPart | RetryPromptPart] = {}

        for msg in messages:
            if isinstance(msg, ModelRequest):
                for part in msg.parts:
                    if isinstance(part, ToolReturnPart):
                        tool_results[part.tool_call_id] = part
                    elif isinstance(part, RetryPromptPart) and part.tool_name:
                        tool_results[part.tool_call_id] = part

        result: list[UIMessage] = []

        for msg in messages:
            if isinstance(msg, ModelRequest):
                system_ui_parts, user_ui_parts = cls._dump_request_message(msg)
                if system_ui_parts:
                    result.append(UIMessage(id=str(uuid.uuid4()), role='system', parts=system_ui_parts))

                if user_ui_parts:
                    result.append(UIMessage(id=str(uuid.uuid4()), role='user', parts=user_ui_parts))

            elif isinstance(  # pragma: no branch
                msg, ModelResponse
            ):
                ui_parts: list[UIMessagePart] = cls._dump_response_message(msg, tool_results)
                if ui_parts:  # pragma: no branch
                    result.append(UIMessage(id=str(uuid.uuid4()), role='assistant', parts=ui_parts))
            else:
                assert_never(msg)

        return result

sdk_version `class-attribute` `instance-attribute`

sdk_version: Literal[5, 6] = 5

Vercel AI SDK version to target. Default is 5 for backwards compatibility.

Setting sdk_version=6 enables tool approval streaming for human-in-the-loop workflows.

build_run_input `classmethod`

build_run_input(body: bytes) -> RequestData

Build a Vercel AI run input object from the request body.

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

@classmethod
def build_run_input(cls, body: bytes) -> RequestData:
    """Build a Vercel AI run input object from the request body."""
    return request_data_ta.validate_json(body)

from_request `async` `classmethod`

from_request(
    request: Request,
    *,
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    sdk_version: Literal[5, 6] = 5,
    **kwargs: Any
) -> VercelAIAdapter[AgentDepsT, OutputDataT]

Extends from_request with the sdk_version parameter.

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

@classmethod
async def from_request(
    cls,
    request: Request,
    *,
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    sdk_version: Literal[5, 6] = 5,
    **kwargs: Any,
) -> VercelAIAdapter[AgentDepsT, OutputDataT]:
    """Extends [`from_request`][pydantic_ai.ui.UIAdapter.from_request] with the `sdk_version` parameter."""
    return await super().from_request(request, agent=agent, sdk_version=sdk_version, **kwargs)

dispatch_request `async` `classmethod`

dispatch_request(
    request: Request,
    *,
    agent: AbstractAgent[
        DispatchDepsT, DispatchOutputDataT
    ],
    sdk_version: Literal[5, 6] = 5,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: (
        DeferredToolResults | None
    ) = None,
    model: Model | KnownModelName | str | None = None,
    instructions: Instructions[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,
    builtin_tools: (
        Sequence[AbstractBuiltinTool] | None
    ) = None,
    on_complete: OnCompleteFunc[BaseChunk] | None = None,
    **kwargs: Any
) -> Response

Extends dispatch_request with the sdk_version parameter.

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

@classmethod
async def dispatch_request(
    cls,
    request: Request,
    *,
    agent: AbstractAgent[DispatchDepsT, DispatchOutputDataT],
    sdk_version: Literal[5, 6] = 5,
    message_history: Sequence[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    model: Model | KnownModelName | str | None = None,
    instructions: Instructions[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,
    builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
    on_complete: OnCompleteFunc[BaseChunk] | None = None,
    **kwargs: Any,
) -> Response:
    """Extends [`dispatch_request`][pydantic_ai.ui.UIAdapter.dispatch_request] with the `sdk_version` parameter."""
    return await super().dispatch_request(
        request,
        agent=agent,
        sdk_version=sdk_version,
        message_history=message_history,
        deferred_tool_results=deferred_tool_results,
        model=model,
        instructions=instructions,
        deps=deps,
        output_type=output_type,
        model_settings=model_settings,
        usage_limits=usage_limits,
        usage=usage,
        metadata=metadata,
        infer_name=infer_name,
        toolsets=toolsets,
        builtin_tools=builtin_tools,
        on_complete=on_complete,
        **kwargs,
    )

build_event_stream

build_event_stream() -> (
    UIEventStream[
        RequestData, BaseChunk, AgentDepsT, OutputDataT
    ]
)

Build a Vercel AI event stream transformer.

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

def build_event_stream(self) -> UIEventStream[RequestData, BaseChunk, AgentDepsT, OutputDataT]:
    """Build a Vercel AI event stream transformer."""
    return VercelAIEventStream(self.run_input, accept=self.accept, sdk_version=self.sdk_version)

deferred_tool_results `cached` `property`

deferred_tool_results: DeferredToolResults | None

Extract deferred tool results from Vercel AI messages with approval responses.

messages `cached` `property`

messages: list[ModelMessage]

Pydantic AI messages from the Vercel AI run input.

load_messages `classmethod`

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

Transform Vercel AI messages into Pydantic AI messages.

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

@classmethod
def load_messages(cls, messages: Sequence[UIMessage]) -> list[ModelMessage]:  # noqa: C901
    """Transform Vercel AI messages into Pydantic AI messages."""
    builder = MessagesBuilder()

    for msg in messages:
        if msg.role == 'system':
            for part in msg.parts:
                if isinstance(part, TextUIPart):
                    builder.add(SystemPromptPart(content=part.text))
                else:  # pragma: no cover
                    raise ValueError(f'Unsupported system message part type: {type(part)}')
        elif msg.role == 'user':
            user_prompt_content: str | list[UserContent] = []
            for part in msg.parts:
                if isinstance(part, TextUIPart):
                    user_prompt_content.append(part.text)
                elif isinstance(part, FileUIPart):
                    try:
                        file = BinaryContent.from_data_uri(part.url)
                    except ValueError:
                        media_type_prefix = part.media_type.split('/', 1)[0]
                        match media_type_prefix:
                            case 'image':
                                file = ImageUrl(url=part.url, media_type=part.media_type)
                            case 'video':
                                file = VideoUrl(url=part.url, media_type=part.media_type)
                            case 'audio':
                                file = AudioUrl(url=part.url, media_type=part.media_type)
                            case _:
                                file = DocumentUrl(url=part.url, media_type=part.media_type)
                    user_prompt_content.append(file)
                else:  # pragma: no cover
                    raise ValueError(f'Unsupported user message part type: {type(part)}')

            if user_prompt_content:  # pragma: no branch
                if len(user_prompt_content) == 1 and isinstance(user_prompt_content[0], str):
                    user_prompt_content = user_prompt_content[0]
                builder.add(UserPromptPart(content=user_prompt_content))

        elif msg.role == 'assistant':
            for part in msg.parts:
                if isinstance(part, TextUIPart):
                    provider_meta = load_provider_metadata(part.provider_metadata)
                    builder.add(
                        TextPart(
                            content=part.text,
                            id=provider_meta.get('id'),
                            provider_name=provider_meta.get('provider_name'),
                            provider_details=provider_meta.get('provider_details'),
                        )
                    )
                elif isinstance(part, ReasoningUIPart):
                    provider_meta = load_provider_metadata(part.provider_metadata)
                    builder.add(
                        ThinkingPart(
                            content=part.text,
                            id=provider_meta.get('id'),
                            signature=provider_meta.get('signature'),
                            provider_name=provider_meta.get('provider_name'),
                            provider_details=provider_meta.get('provider_details'),
                        )
                    )
                elif isinstance(part, FileUIPart):
                    try:
                        file = BinaryContent.from_data_uri(part.url)
                    except ValueError as e:  # pragma: no cover
                        # We don't yet handle non-data-URI file URLs returned by assistants, as no Pydantic AI models do this.
                        raise ValueError(
                            'Vercel AI integration can currently only handle assistant file parts with data URIs.'
                        ) from e
                    provider_meta = load_provider_metadata(part.provider_metadata)
                    builder.add(
                        FilePart(
                            content=file,
                            id=provider_meta.get('id'),
                            provider_name=provider_meta.get('provider_name'),
                            provider_details=provider_meta.get('provider_details'),
                        )
                    )
                elif isinstance(part, ToolUIPart | DynamicToolUIPart):
                    if isinstance(part, DynamicToolUIPart):
                        tool_name = part.tool_name
                        builtin_tool = False
                    else:
                        tool_name = part.type.removeprefix('tool-')
                        builtin_tool = part.provider_executed

                    tool_call_id = part.tool_call_id

                    args: str | dict[str, Any] | None = part.input

                    if isinstance(args, str):
                        try:
                            parsed = json.loads(args)
                            if isinstance(parsed, dict):
                                args = cast(dict[str, Any], parsed)
                        except json.JSONDecodeError:
                            pass
                    elif isinstance(args, dict) or args is None:
                        pass
                    else:
                        assert_never(args)

                    provider_meta = load_provider_metadata(part.call_provider_metadata)
                    part_id = provider_meta.get('id')
                    provider_name = provider_meta.get('provider_name')
                    provider_details = provider_meta.get('provider_details')

                    if builtin_tool:
                        # For builtin tools, we need to create 2 parts (BuiltinToolCall & BuiltinToolReturn) for a single Vercel ToolOutput
                        # The call and return metadata are combined in the output part.
                        # So we extract and return them to the respective parts
                        call_meta = return_meta = {}
                        has_tool_output = isinstance(part, (ToolOutputAvailablePart, ToolOutputErrorPart))

                        if has_tool_output:
                            call_meta, return_meta = cls._load_builtin_tool_meta(provider_meta)

                        builder.add(
                            BuiltinToolCallPart(
                                tool_name=tool_name,
                                tool_call_id=tool_call_id,
                                args=args,
                                id=call_meta.get('id') or part_id,
                                provider_name=call_meta.get('provider_name') or provider_name,
                                provider_details=call_meta.get('provider_details') or provider_details,
                            )
                        )

                        if has_tool_output:
                            output: Any | None = None
                            if isinstance(part, ToolOutputAvailablePart):
                                output = part.output
                            elif isinstance(part, ToolOutputErrorPart):  # pragma: no branch
                                output = {'error_text': part.error_text, 'is_error': True}
                            builder.add(
                                BuiltinToolReturnPart(
                                    tool_name=tool_name,
                                    tool_call_id=tool_call_id,
                                    content=output,
                                    provider_name=return_meta.get('provider_name') or provider_name,
                                    provider_details=return_meta.get('provider_details') or provider_details,
                                )
                            )
                    else:
                        builder.add(
                            ToolCallPart(
                                tool_name=tool_name,
                                tool_call_id=tool_call_id,
                                args=args,
                                id=part_id,
                                provider_name=provider_name,
                                provider_details=provider_details,
                            )
                        )

                        if part.state == 'output-available':
                            builder.add(
                                ToolReturnPart(tool_name=tool_name, tool_call_id=tool_call_id, content=part.output)
                            )
                        elif part.state == 'output-error':
                            builder.add(
                                RetryPromptPart(
                                    tool_name=tool_name, tool_call_id=tool_call_id, content=part.error_text
                                )
                            )
                elif isinstance(part, DataUIPart):  # pragma: no cover
                    # Contains custom data that shouldn't be sent to the model
                    pass
                elif isinstance(part, SourceUrlUIPart):  # pragma: no cover
                    # TODO: Once we support citations: https://github.com/pydantic/pydantic-ai/issues/3126
                    pass
                elif isinstance(part, SourceDocumentUIPart):  # pragma: no cover
                    # TODO: Once we support citations: https://github.com/pydantic/pydantic-ai/issues/3126
                    pass
                elif isinstance(part, StepStartUIPart):  # pragma: no cover
                    # Nothing to do here
                    pass
                else:
                    assert_never(part)
        else:
            assert_never(msg.role)

    return builder.messages

dump_messages `classmethod`

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

Transform Pydantic AI messages into Vercel AI messages.

Parameters:

Name	Type	Description	Default
`messages`	`Sequence[ModelMessage]`	A sequence of ModelMessage objects to convert	required

Returns:

Type	Description
`list[UIMessage]`	A list of UIMessage objects in Vercel AI format

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

@classmethod
def dump_messages(
    cls,
    messages: Sequence[ModelMessage],
) -> list[UIMessage]:
    """Transform Pydantic AI messages into Vercel AI messages.

    Args:
        messages: A sequence of ModelMessage objects to convert

    Returns:
        A list of UIMessage objects in Vercel AI format
    """
    tool_results: dict[str, ToolReturnPart | RetryPromptPart] = {}

    for msg in messages:
        if isinstance(msg, ModelRequest):
            for part in msg.parts:
                if isinstance(part, ToolReturnPart):
                    tool_results[part.tool_call_id] = part
                elif isinstance(part, RetryPromptPart) and part.tool_name:
                    tool_results[part.tool_call_id] = part

    result: list[UIMessage] = []

    for msg in messages:
        if isinstance(msg, ModelRequest):
            system_ui_parts, user_ui_parts = cls._dump_request_message(msg)
            if system_ui_parts:
                result.append(UIMessage(id=str(uuid.uuid4()), role='system', parts=system_ui_parts))

            if user_ui_parts:
                result.append(UIMessage(id=str(uuid.uuid4()), role='user', parts=user_ui_parts))

        elif isinstance(  # pragma: no branch
            msg, ModelResponse
        ):
            ui_parts: list[UIMessagePart] = cls._dump_response_message(msg, tool_results)
            if ui_parts:  # pragma: no branch
                result.append(UIMessage(id=str(uuid.uuid4()), role='assistant', parts=ui_parts))
        else:
            assert_never(msg)

    return result

VercelAIEventStream `dataclass`

Bases: UIEventStream[RequestData, BaseChunk, AgentDepsT, OutputDataT]

UI event stream transformer for the Vercel AI protocol.

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

@dataclass
class VercelAIEventStream(UIEventStream[RequestData, BaseChunk, AgentDepsT, OutputDataT]):
    """UI event stream transformer for the Vercel AI protocol."""

    _: KW_ONLY
    sdk_version: Literal[5, 6] = 5
    """Vercel AI SDK version to target. Setting to 6 enables tool approval streaming."""

    _step_started: bool = False
    _finish_reason: FinishReason = None

    @cached_property
    def _denied_tool_ids(self) -> set[str]:
        """Get the set of tool_call_ids that were denied by the user."""
        return {
            tool_call_id
            for tool_call_id, approval in iter_tool_approval_responses(self.run_input.messages)
            if not approval.approved
        }

    @property
    def response_headers(self) -> Mapping[str, str] | None:
        return VERCEL_AI_DSP_HEADERS

    def encode_event(self, event: BaseChunk) -> str:
        return f'data: {event.encode(self.sdk_version)}\n\n'

    async def before_stream(self) -> AsyncIterator[BaseChunk]:
        yield StartChunk()

    async def before_response(self) -> AsyncIterator[BaseChunk]:
        if self._step_started:
            yield FinishStepChunk()

        self._step_started = True
        yield StartStepChunk()

    async def after_stream(self) -> AsyncIterator[BaseChunk]:
        yield FinishStepChunk()

        yield FinishChunk(finish_reason=self._finish_reason)
        yield DoneChunk()

    async def handle_run_result(self, event: AgentRunResultEvent) -> AsyncIterator[BaseChunk]:
        pydantic_reason = event.result.response.finish_reason
        if pydantic_reason:
            self._finish_reason = _FINISH_REASON_MAP.get(pydantic_reason, 'other')

        # Emit tool approval requests for deferred approvals (only when sdk_version >= 6)
        output = event.result.output
        if self.sdk_version >= 6 and isinstance(output, DeferredToolRequests):
            for tool_call in output.approvals:
                yield ToolApprovalRequestChunk(
                    approval_id=str(uuid4()),
                    tool_call_id=tool_call.tool_call_id,
                )
            return
        return
        yield

    async def on_error(self, error: Exception) -> AsyncIterator[BaseChunk]:
        self._finish_reason = 'error'
        yield ErrorChunk(error_text=str(error))

    async def handle_text_start(self, part: TextPart, follows_text: bool = False) -> AsyncIterator[BaseChunk]:
        provider_metadata = dump_provider_metadata(
            id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
        )
        if follows_text:
            message_id = self.message_id
        else:
            message_id = self.new_message_id()
            yield TextStartChunk(id=message_id, provider_metadata=provider_metadata)

        if part.content:
            yield TextDeltaChunk(id=message_id, delta=part.content, provider_metadata=provider_metadata)

    async def handle_text_delta(self, delta: TextPartDelta) -> AsyncIterator[BaseChunk]:
        if delta.content_delta:  # pragma: no branch
            provider_metadata = dump_provider_metadata(
                provider_name=delta.provider_name, provider_details=delta.provider_details
            )
            yield TextDeltaChunk(id=self.message_id, delta=delta.content_delta, provider_metadata=provider_metadata)

    async def handle_text_end(self, part: TextPart, followed_by_text: bool = False) -> AsyncIterator[BaseChunk]:
        if not followed_by_text:
            provider_metadata = dump_provider_metadata(
                id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
            )
            yield TextEndChunk(id=self.message_id, provider_metadata=provider_metadata)

    async def handle_thinking_start(
        self, part: ThinkingPart, follows_thinking: bool = False
    ) -> AsyncIterator[BaseChunk]:
        message_id = self.new_message_id()
        provider_metadata = dump_provider_metadata(
            id=part.id,
            signature=part.signature,
            provider_name=part.provider_name,
            provider_details=part.provider_details,
        )
        yield ReasoningStartChunk(id=message_id, provider_metadata=provider_metadata)
        if part.content:
            yield ReasoningDeltaChunk(id=message_id, delta=part.content, provider_metadata=provider_metadata)

    async def handle_thinking_delta(self, delta: ThinkingPartDelta) -> AsyncIterator[BaseChunk]:
        if delta.content_delta:  # pragma: no branch
            provider_metadata = dump_provider_metadata(
                provider_name=delta.provider_name,
                signature=delta.signature_delta,
                provider_details=delta.provider_details,
            )
            yield ReasoningDeltaChunk(
                id=self.message_id, delta=delta.content_delta, provider_metadata=provider_metadata
            )

    async def handle_thinking_end(
        self, part: ThinkingPart, followed_by_thinking: bool = False
    ) -> AsyncIterator[BaseChunk]:
        provider_metadata = dump_provider_metadata(
            id=part.id,
            signature=part.signature,
            provider_name=part.provider_name,
            provider_details=part.provider_details,
        )
        yield ReasoningEndChunk(id=self.message_id, provider_metadata=provider_metadata)

    def handle_tool_call_start(self, part: ToolCallPart | BuiltinToolCallPart) -> AsyncIterator[BaseChunk]:
        return self._handle_tool_call_start(part)

    def handle_builtin_tool_call_start(self, part: BuiltinToolCallPart) -> AsyncIterator[BaseChunk]:
        return self._handle_tool_call_start(part, provider_executed=True)

    async def _handle_tool_call_start(
        self,
        part: ToolCallPart | BuiltinToolCallPart,
        tool_call_id: str | None = None,
        provider_executed: bool | None = None,
    ) -> AsyncIterator[BaseChunk]:
        tool_call_id = tool_call_id or part.tool_call_id
        yield ToolInputStartChunk(
            tool_call_id=tool_call_id,
            tool_name=part.tool_name,
            provider_executed=provider_executed,
            provider_metadata=dump_provider_metadata(
                id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
            ),
        )
        if part.args:
            yield ToolInputDeltaChunk(tool_call_id=tool_call_id, input_text_delta=part.args_as_json_str())

    async def handle_tool_call_delta(self, delta: ToolCallPartDelta) -> AsyncIterator[BaseChunk]:
        tool_call_id = delta.tool_call_id or ''
        assert tool_call_id, '`ToolCallPartDelta.tool_call_id` must be set'
        yield ToolInputDeltaChunk(
            tool_call_id=tool_call_id,
            input_text_delta=delta.args_delta if isinstance(delta.args_delta, str) else _json_dumps(delta.args_delta),
        )

    async def handle_tool_call_end(self, part: ToolCallPart) -> AsyncIterator[BaseChunk]:
        yield ToolInputAvailableChunk(
            tool_call_id=part.tool_call_id,
            tool_name=part.tool_name,
            input=part.args_as_dict(),
            provider_metadata=dump_provider_metadata(
                id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
            ),
        )

    async def handle_builtin_tool_call_end(self, part: BuiltinToolCallPart) -> AsyncIterator[BaseChunk]:
        yield ToolInputAvailableChunk(
            tool_call_id=part.tool_call_id,
            tool_name=part.tool_name,
            input=part.args_as_dict(),
            provider_executed=True,
            provider_metadata=dump_provider_metadata(
                id=part.id, provider_name=part.provider_name, provider_details=part.provider_details
            ),
        )

    async def handle_builtin_tool_return(self, part: BuiltinToolReturnPart) -> AsyncIterator[BaseChunk]:
        yield ToolOutputAvailableChunk(
            tool_call_id=part.tool_call_id,
            output=self._tool_return_output(part),
            provider_executed=True,
        )

    async def handle_file(self, part: FilePart) -> AsyncIterator[BaseChunk]:
        file = part.content
        yield FileChunk(url=file.data_uri, media_type=file.media_type)

    async def handle_function_tool_result(self, event: FunctionToolResultEvent) -> AsyncIterator[BaseChunk]:
        part = event.result
        tool_call_id = part.tool_call_id

        # Check if this tool was denied by the user (only when sdk_version >= 6)
        if self.sdk_version >= 6 and tool_call_id in self._denied_tool_ids:
            yield ToolOutputDeniedChunk(tool_call_id=tool_call_id)
        elif isinstance(part, RetryPromptPart):
            yield ToolOutputErrorChunk(tool_call_id=tool_call_id, error_text=part.model_response())
        else:
            yield ToolOutputAvailableChunk(tool_call_id=tool_call_id, output=self._tool_return_output(part))

        # ToolOutputAvailableChunk/ToolOutputErrorChunk.output may hold user parts
        # (e.g. text, images) that Vercel AI does not currently have chunk types for.

        # Check for data-carrying Vercel AI chunks returned by tool calls via metadata.
        # Only data-carrying chunks (DataChunk, SourceUrlChunk, etc.) are yielded;
        # protocol-control chunks are filtered out by iter_metadata_chunks.
        if isinstance(part, ToolReturnPart):
            for chunk in iter_metadata_chunks(part):
                yield chunk

    def _tool_return_output(self, part: BaseToolReturnPart) -> Any:
        output = part.model_response_object()
        # Unwrap the return value from the output dictionary if it exists
        return output.get('return_value', output)

sdk_version `class-attribute` `instance-attribute`

sdk_version: Literal[5, 6] = 5

Vercel AI SDK version to target. Setting to 6 enables tool approval streaming.

Vercel AI request types (UI messages).

Converted to Python from: https://github.com/vercel/ai/blob/ai%406.0.57/packages/ai/src/ui/ui-messages.ts

Tool approval types (ToolApprovalRequested, ToolApprovalResponded) require AI SDK v6 or later.

ProviderMetadata `module-attribute`

ProviderMetadata = dict[str, dict[str, JSONValue]]

Provider metadata.

BaseUIPart

Bases: CamelBaseModel, ABC

Abstract base class for all UI parts.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class BaseUIPart(CamelBaseModel, ABC):
    """Abstract base class for all UI parts."""

TextUIPart

Bases: BaseUIPart

A text part of a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class TextUIPart(BaseUIPart):
    """A text part of a message."""

    type: Literal['text'] = 'text'

    text: str
    """The text content."""

    state: Literal['streaming', 'done'] | None = None
    """The state of the text part."""

    provider_metadata: ProviderMetadata | None = None
    """The provider metadata."""

text `instance-attribute`

text: str

The text content.

state `class-attribute` `instance-attribute`

state: Literal['streaming', 'done'] | None = None

The state of the text part.

provider_metadata `class-attribute` `instance-attribute`

provider_metadata: ProviderMetadata | None = None

The provider metadata.

ReasoningUIPart

Bases: BaseUIPart

A reasoning part of a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ReasoningUIPart(BaseUIPart):
    """A reasoning part of a message."""

    type: Literal['reasoning'] = 'reasoning'

    text: str
    """The reasoning text."""

    state: Literal['streaming', 'done'] | None = None
    """The state of the reasoning part."""

    provider_metadata: ProviderMetadata | None = None
    """The provider metadata."""

text `instance-attribute`

text: str

The reasoning text.

state `class-attribute` `instance-attribute`

state: Literal['streaming', 'done'] | None = None

The state of the reasoning part.

provider_metadata `class-attribute` `instance-attribute`

provider_metadata: ProviderMetadata | None = None

The provider metadata.

SourceUrlUIPart

Bases: BaseUIPart

A source part of a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class SourceUrlUIPart(BaseUIPart):
    """A source part of a message."""

    type: Literal['source-url'] = 'source-url'
    source_id: str
    url: str
    title: str | None = None
    provider_metadata: ProviderMetadata | None = None

SourceDocumentUIPart

Bases: BaseUIPart

A document source part of a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class SourceDocumentUIPart(BaseUIPart):
    """A document source part of a message."""

    type: Literal['source-document'] = 'source-document'
    source_id: str
    media_type: str
    title: str
    filename: str | None = None
    provider_metadata: ProviderMetadata | None = None

FileUIPart

Bases: BaseUIPart

A file part of a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class FileUIPart(BaseUIPart):
    """A file part of a message."""

    type: Literal['file'] = 'file'

    media_type: str
    """
    IANA media type of the file.
    @see https://www.iana.org/assignments/media-types/media-types.xhtml
    """

    filename: str | None = None
    """Optional filename of the file."""

    url: str
    """
    The URL of the file.
    It can either be a URL to a hosted file or a [Data URL](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/Data_URLs).
    """

    provider_metadata: ProviderMetadata | None = None
    """The provider metadata."""

media_type `instance-attribute`

media_type: str

IANA media type of the file. @see https://www.iana.org/assignments/media-types/media-types.xhtml

filename `class-attribute` `instance-attribute`

filename: str | None = None

Optional filename of the file.

url `instance-attribute`

url: str

The URL of the file. It can either be a URL to a hosted file or a Data URL.

provider_metadata `class-attribute` `instance-attribute`

provider_metadata: ProviderMetadata | None = None

The provider metadata.

StepStartUIPart

Bases: BaseUIPart

A step boundary part of a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class StepStartUIPart(BaseUIPart):
    """A step boundary part of a message."""

    type: Literal['step-start'] = 'step-start'

DataUIPart

Bases: BaseUIPart

Data part with dynamic type based on data name.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class DataUIPart(BaseUIPart):
    """Data part with dynamic type based on data name."""

    type: Annotated[str, Field(pattern=r'^data-')]
    id: str | None = None
    data: Any

ToolApprovalRequested

Bases: CamelBaseModel

Tool approval in requested state (awaiting user response).

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ToolApprovalRequested(CamelBaseModel):
    """Tool approval in requested state (awaiting user response)."""

    id: str
    """The approval request ID."""

id `instance-attribute`

id: str

The approval request ID.

ToolApprovalResponded

Bases: CamelBaseModel

Tool approval in responded state (user has approved or denied).

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ToolApprovalResponded(CamelBaseModel):
    """Tool approval in responded state (user has approved or denied)."""

    id: str
    """The approval request ID."""

    approved: bool
    """Whether the user approved the tool call."""

    reason: str | None = None
    """Optional reason for the approval or denial."""

id `instance-attribute`

id: str

The approval request ID.

approved `instance-attribute`

approved: bool

Whether the user approved the tool call.

reason `class-attribute` `instance-attribute`

reason: str | None = None

Optional reason for the approval or denial.

ToolApproval `module-attribute`

ToolApproval = ToolApprovalRequested | ToolApprovalResponded

Union of tool approval states.

ToolInputStreamingPart

Bases: BaseUIPart

Tool part in input-streaming state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ToolInputStreamingPart(BaseUIPart):
    """Tool part in input-streaming state."""

    type: Annotated[str, Field(pattern=r'^tool-')]
    tool_call_id: str
    state: Literal['input-streaming'] = 'input-streaming'
    input: Any | None = None
    provider_executed: bool | None = None
    call_provider_metadata: ProviderMetadata | None = None
    approval: ToolApproval | None = None

ToolInputAvailablePart

Bases: BaseUIPart

Tool part in input-available state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ToolInputAvailablePart(BaseUIPart):
    """Tool part in input-available state."""

    type: Annotated[str, Field(pattern=r'^tool-')]
    tool_call_id: str
    state: Literal['input-available'] = 'input-available'
    input: Any | None = None
    provider_executed: bool | None = None
    call_provider_metadata: ProviderMetadata | None = None
    approval: ToolApproval | None = None

ToolOutputAvailablePart

Bases: BaseUIPart

Tool part in output-available state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ToolOutputAvailablePart(BaseUIPart):
    """Tool part in output-available state."""

    type: Annotated[str, Field(pattern=r'^tool-')]
    tool_call_id: str
    state: Literal['output-available'] = 'output-available'
    input: Any | None = None
    output: Any | None = None
    provider_executed: bool | None = None
    call_provider_metadata: ProviderMetadata | None = None
    preliminary: bool | None = None
    approval: ToolApproval | None = None

ToolOutputErrorPart

Bases: BaseUIPart

Tool part in output-error state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class ToolOutputErrorPart(BaseUIPart):
    """Tool part in output-error state."""

    type: Annotated[str, Field(pattern=r'^tool-')]
    tool_call_id: str
    state: Literal['output-error'] = 'output-error'
    input: Any | None = None
    raw_input: Any | None = None
    error_text: str
    provider_executed: bool | None = None
    call_provider_metadata: ProviderMetadata | None = None
    approval: ToolApproval | None = None

ToolUIPart `module-attribute`

ToolUIPart = (
    ToolInputStreamingPart
    | ToolInputAvailablePart
    | ToolOutputAvailablePart
    | ToolOutputErrorPart
)

Union of all tool part types.

DynamicToolInputStreamingPart

Bases: BaseUIPart

Dynamic tool part in input-streaming state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class DynamicToolInputStreamingPart(BaseUIPart):
    """Dynamic tool part in input-streaming state."""

    type: Literal['dynamic-tool'] = 'dynamic-tool'
    tool_name: str
    tool_call_id: str
    state: Literal['input-streaming'] = 'input-streaming'
    input: Any | None = None
    call_provider_metadata: ProviderMetadata | None = None
    approval: ToolApproval | None = None

DynamicToolInputAvailablePart

Bases: BaseUIPart

Dynamic tool part in input-available state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class DynamicToolInputAvailablePart(BaseUIPart):
    """Dynamic tool part in input-available state."""

    type: Literal['dynamic-tool'] = 'dynamic-tool'
    tool_name: str
    tool_call_id: str
    state: Literal['input-available'] = 'input-available'
    input: Any
    call_provider_metadata: ProviderMetadata | None = None
    approval: ToolApproval | None = None

DynamicToolOutputAvailablePart

Bases: BaseUIPart

Dynamic tool part in output-available state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class DynamicToolOutputAvailablePart(BaseUIPart):
    """Dynamic tool part in output-available state."""

    type: Literal['dynamic-tool'] = 'dynamic-tool'
    tool_name: str
    tool_call_id: str
    state: Literal['output-available'] = 'output-available'
    input: Any
    output: Any
    call_provider_metadata: ProviderMetadata | None = None
    preliminary: bool | None = None
    approval: ToolApproval | None = None

DynamicToolOutputErrorPart

Bases: BaseUIPart

Dynamic tool part in output-error state.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class DynamicToolOutputErrorPart(BaseUIPart):
    """Dynamic tool part in output-error state."""

    type: Literal['dynamic-tool'] = 'dynamic-tool'
    tool_name: str
    tool_call_id: str
    state: Literal['output-error'] = 'output-error'
    input: Any
    error_text: str
    call_provider_metadata: ProviderMetadata | None = None
    approval: ToolApproval | None = None

DynamicToolUIPart `module-attribute`

DynamicToolUIPart = (
    DynamicToolInputStreamingPart
    | DynamicToolInputAvailablePart
    | DynamicToolOutputAvailablePart
    | DynamicToolOutputErrorPart
)

Union of all dynamic tool part types.

UIMessagePart `module-attribute`

UIMessagePart = (
    TextUIPart
    | ReasoningUIPart
    | ToolUIPart
    | DynamicToolUIPart
    | SourceUrlUIPart
    | SourceDocumentUIPart
    | FileUIPart
    | DataUIPart
    | StepStartUIPart
)

Union of all message part types.

UIMessage

Bases: CamelBaseModel

A message as displayed in the UI by Vercel AI Elements.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class UIMessage(CamelBaseModel):
    """A message as displayed in the UI by Vercel AI Elements."""

    id: str
    """A unique identifier for the message."""

    role: Literal['system', 'user', 'assistant']
    """The role of the message."""

    metadata: Any | None = None
    """The metadata of the message."""

    parts: list[UIMessagePart]
    """
    The parts of the message. Use this for rendering the message in the UI.
    System messages should be avoided (set the system prompt on the server instead).
    They can have text parts.
    User messages can have text parts and file parts.
    Assistant messages can have text, reasoning, tool invocation, and file parts.
    """

id `instance-attribute`

id: str

A unique identifier for the message.

role `instance-attribute`

role: Literal['system', 'user', 'assistant']

The role of the message.

metadata `class-attribute` `instance-attribute`

metadata: Any | None = None

The metadata of the message.

parts `instance-attribute`

parts: list[UIMessagePart]

The parts of the message. Use this for rendering the message in the UI. System messages should be avoided (set the system prompt on the server instead). They can have text parts. User messages can have text parts and file parts. Assistant messages can have text, reasoning, tool invocation, and file parts.

SubmitMessage

Bases: CamelBaseModel

Submit message request.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class SubmitMessage(CamelBaseModel, extra='allow'):
    """Submit message request."""

    trigger: Literal['submit-message'] = 'submit-message'
    id: str
    messages: list[UIMessage]

RegenerateMessage

Bases: CamelBaseModel

Ask the agent to regenerate a message.

Source code in pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py

class RegenerateMessage(CamelBaseModel, extra='allow'):
    """Ask the agent to regenerate a message."""

    trigger: Literal['regenerate-message']
    id: str
    messages: list[UIMessage]
    message_id: str

RequestData `module-attribute`

RequestData = Annotated[
    SubmitMessage | RegenerateMessage,
    Discriminator("trigger"),
]

Union of all request data types.

Vercel AI response types (SSE chunks).

Converted to Python from: https://github.com/vercel/ai/blob/ai%406.0.57/packages/ai/src/ui-message-stream/ui-message-chunks.ts

Tool approval types (ToolApprovalRequestChunk, ToolOutputDeniedChunk) require AI SDK UI v6 or later.

ProviderMetadata `module-attribute`

ProviderMetadata = dict[str, dict[str, JSONValue]]

Provider metadata.

FinishReason `module-attribute`

FinishReason = (
    Literal[
        "stop",
        "length",
        "content-filter",
        "tool-calls",
        "error",
        "other",
    ]
    | None
)

Reason why the model finished generating.