`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.""" @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) 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) @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): builder.add(TextPart(content=part.text)) elif isinstance(part, ReasoningUIPart): pydantic_ai_meta = (part.provider_metadata or {}).get('pydantic_ai', {}) builder.add( ThinkingPart( content=part.text, id=pydantic_ai_meta.get('id'), signature=pydantic_ai_meta.get('signature'), provider_name=pydantic_ai_meta.get('provider_name'), provider_details=pydantic_ai_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 builder.add(FilePart(content=file)) 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) if builtin_tool: call_part = BuiltinToolCallPart(tool_name=tool_name, tool_call_id=tool_call_id, args=args) builder.add(call_part) if isinstance(part, ToolOutputAvailablePart | ToolOutputErrorPart): if part.state == 'output-available': output = part.output else: output = {'error_text': part.error_text, 'is_error': True} provider_name = ( (part.call_provider_metadata or {}).get('pydantic_ai', {}).get('provider_name') ) call_part.provider_name = provider_name builder.add( BuiltinToolReturnPart( tool_name=tool_name, tool_call_id=tool_call_id, content=output, provider_name=provider_name, ) ) else: builder.add(ToolCallPart(tool_name=tool_name, tool_call_id=tool_call_id, args=args)) 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_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 @staticmethod def _dump_response_message( # noqa: C901 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: ui_parts.append(TextUIPart(text=part.content, state='done')) elif isinstance(part, ThinkingPart): thinking_metadata: dict[str, Any] = {} if part.id is not None: thinking_metadata['id'] = part.id if part.signature is not None: thinking_metadata['signature'] = part.signature if part.provider_name is not None: thinking_metadata['provider_name'] = part.provider_name if part.provider_details is not None: thinking_metadata['provider_details'] = part.provider_details provider_metadata = {'pydantic_ai': thinking_metadata} if thinking_metadata else None 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, ) ) elif isinstance(part, BuiltinToolCallPart): call_provider_metadata = ( {'pydantic_ai': {'provider_name': part.provider_name}} if part.provider_name else None ) if builtin_return := local_builtin_returns.get(part.tool_call_id): content = builtin_return.model_response_str() ui_parts.append( ToolOutputAvailablePart( type=f'tool-{part.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=call_provider_metadata, ) ) else: ui_parts.append( ToolInputAvailablePart( type=f'tool-{part.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) 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', ) ) 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', ) ) 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', ) ) 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 

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) 

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) 

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): builder.add(TextPart(content=part.text)) elif isinstance(part, ReasoningUIPart): pydantic_ai_meta = (part.provider_metadata or {}).get('pydantic_ai', {}) builder.add( ThinkingPart( content=part.text, id=pydantic_ai_meta.get('id'), signature=pydantic_ai_meta.get('signature'), provider_name=pydantic_ai_meta.get('provider_name'), provider_details=pydantic_ai_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 builder.add(FilePart(content=file)) 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) if builtin_tool: call_part = BuiltinToolCallPart(tool_name=tool_name, tool_call_id=tool_call_id, args=args) builder.add(call_part) if isinstance(part, ToolOutputAvailablePart | ToolOutputErrorPart): if part.state == 'output-available': output = part.output else: output = {'error_text': part.error_text, 'is_error': True} provider_name = ( (part.call_provider_metadata or {}).get('pydantic_ai', {}).get('provider_name') ) call_part.provider_name = provider_name builder.add( BuiltinToolReturnPart( tool_name=tool_name, tool_call_id=tool_call_id, content=output, provider_name=provider_name, ) ) else: builder.add(ToolCallPart(tool_name=tool_name, tool_call_id=tool_call_id, args=args)) 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.""" _step_started: bool = False @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()}\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() yield DoneChunk() async def on_error(self, error: Exception) -> AsyncIterator[BaseChunk]: yield ErrorChunk(error_text=str(error)) async def handle_text_start(self, part: TextPart, follows_text: bool = False) -> AsyncIterator[BaseChunk]: if follows_text: message_id = self.message_id else: message_id = self.new_message_id() yield TextStartChunk(id=message_id) if part.content: yield TextDeltaChunk(id=message_id, delta=part.content) async def handle_text_delta(self, delta: TextPartDelta) -> AsyncIterator[BaseChunk]: if delta.content_delta: # pragma: no branch yield TextDeltaChunk(id=self.message_id, delta=delta.content_delta) async def handle_text_end(self, part: TextPart, followed_by_text: bool = False) -> AsyncIterator[BaseChunk]: if not followed_by_text: yield TextEndChunk(id=self.message_id) async def handle_thinking_start( self, part: ThinkingPart, follows_thinking: bool = False ) -> AsyncIterator[BaseChunk]: message_id = self.new_message_id() yield ReasoningStartChunk(id=message_id) if part.content: yield ReasoningDeltaChunk(id=message_id, delta=part.content) async def handle_thinking_delta(self, delta: ThinkingPartDelta) -> AsyncIterator[BaseChunk]: if delta.content_delta: # pragma: no branch yield ReasoningDeltaChunk(id=self.message_id, delta=delta.content_delta) async def handle_thinking_end( self, part: ThinkingPart, followed_by_thinking: bool = False ) -> AsyncIterator[BaseChunk]: yield ReasoningEndChunk(id=self.message_id) 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, ) 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() ) 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={'pydantic_ai': {'provider_name': part.provider_name}}, ) 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 if isinstance(part, RetryPromptPart): yield ToolOutputErrorChunk(tool_call_id=part.tool_call_id, error_text=part.model_response()) else: yield ToolOutputAvailableChunk(tool_call_id=part.tool_call_id, output=self._tool_return_output(part)) # ToolCallResultEvent.content may hold user parts (e.g. text, images) that Vercel AI does not currently have events for 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) 

Vercel AI request types (UI messages).

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

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 

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 

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 

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 

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 

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 

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 

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 

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 

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%405.0.59/packages/ai/src/ui-message-stream/ui-message-chunks.ts

ProviderMetadata `module-attribute`

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

Provider metadata.