Add VercelAIAdapter.dump_messages to convert Pydantic AI messages to Vercel AI messages

pydantic_ai_slim/pydantic_ai/ui/_adapter.py

@@ -143,6 +143,11 @@ def load_messages(cls, messages: Sequence[MessageT]) -> list[ModelMessage]:
         """Transform protocol-specific messages into Pydantic AI messages."""
         raise NotImplementedError
 
+    @classmethod
+    def dump_messages(cls, messages: Sequence[ModelMessage]) -> list[MessageT]:
+        """Transform Pydantic AI messages into protocol-specific messages."""
+        raise NotImplementedError
+
     @abstractmethod
     def build_event_stream(self) -> UIEventStream[RunInputT, EventT, AgentDepsT, OutputDataT]:
         """Build a protocol-specific event stream transformer."""

pydantic_ai_slim/pydantic_ai/ui/vercel_ai/_adapter.py

@@ -14,7 +14,6 @@
 
 from ...messages import (
     AudioUrl,
-    BaseToolCallPart,
     BinaryContent,
     BuiltinToolCallPart,
     BuiltinToolReturnPart,
@@ -246,10 +245,12 @@ def _dump_request_message(msg: ModelRequest) -> tuple[list[UIMessagePart], list[
                 # Tool returns are merged into the tool call in the assistant message
                 pass
             elif isinstance(part, RetryPromptPart):
-                # RetryPromptPart always has a tool_call_id (generated if not provided).
-                # These are handled when processing ToolCallPart in ModelResponse,
-                # where they become DynamicToolOutputErrorPart via the tool_errors dict.
-                pass
+                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)
 
@@ -258,8 +259,7 @@ def _dump_request_message(msg: ModelRequest) -> tuple[list[UIMessagePart], list[
     @staticmethod
     def _dump_response_message(  # noqa: C901
         msg: ModelResponse,
-        tool_returns: dict[str, ToolReturnPart],
-        tool_errors: dict[str, RetryPromptPart],
+        tool_results: dict[str, ToolReturnPart | RetryPromptPart],
     ) -> list[UIMessagePart]:
         """Convert a ModelResponse into a UIMessage."""
         ui_parts: list[UIMessagePart] = []
@@ -298,71 +298,69 @@ def _dump_response_message(  # noqa: C901
                         media_type=part.content.media_type,
                     )
                 )
-            elif isinstance(part, BaseToolCallPart):
-                if isinstance(part, BuiltinToolCallPart):
-                    call_provider_metadata = (
-                        {'pydantic_ai': {'provider_name': part.provider_name}} if part.provider_name else None
-                    )
+            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,
-                            )
+                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:  # pragma: no cover
-                        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,
-                            )
+                    )
+                else:  # pragma: no cover
+                    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,
                         )
-                else:
-                    tool_return = tool_returns.get(part.tool_call_id)
-                    tool_error = tool_errors.get(part.tool_call_id)
-
-                    if isinstance(tool_return, ToolReturnPart):
-                        content = tool_return.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(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 tool_error:
-                        error_text = tool_error.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',
-                            )
+                    )
+                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:
+                    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)
 
@@ -381,16 +379,15 @@ def dump_messages(
         Returns:
             A list of UIMessage objects in Vercel AI format
         """
-        tool_returns: dict[str, ToolReturnPart] = {}
-        tool_errors: dict[str, RetryPromptPart] = {}
+        tool_results: dict[str, ToolReturnPart | RetryPromptPart] = {}
 
         for msg in messages:
             if isinstance(msg, ModelRequest):
                 for part in msg.parts:
                     if isinstance(part, ToolReturnPart):
-                        tool_returns[part.tool_call_id] = part
-                    elif isinstance(part, RetryPromptPart) and part.tool_call_id:
-                        tool_errors[part.tool_call_id] = part
+                        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] = []
 
@@ -406,9 +403,7 @@ def dump_messages(
             elif isinstance(  # pragma: no branch
                 msg, ModelResponse
             ):
-                ui_parts: list[UIMessagePart] = cls._dump_response_message(
-                    msg, tool_returns=tool_returns, tool_errors=tool_errors
-                )
+                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:

tests/test_vercel_ai.py

@@ -2292,6 +2292,61 @@ async def test_adapter_dump_messages_with_retry():
     )
 
 
+async def test_adapter_dump_messages_with_retry_no_tool_name():
+    """Test dumping messages with retry prompts without tool_name (e.g., output validation errors)."""
+    messages = [
+        ModelRequest(parts=[UserPromptPart(content='Give me a number')]),
+        ModelResponse(parts=[TextPart(content='Not a valid number')]),
+        ModelRequest(
+            parts=[
+                RetryPromptPart(
+                    content='Output validation failed: expected integer',
+                    # No tool_name - this is an output validation error, not a tool error
+                )
+            ]
+        ),
+    ]
+
+    ui_messages = VercelAIAdapter.dump_messages(messages)
+
+    ui_message_dicts = [msg.model_dump() for msg in ui_messages]
+
+    assert ui_message_dicts == snapshot(
+        [
+            {
+                'id': IsStr(),
+                'role': 'user',
+                'metadata': None,
+                'parts': [{'type': 'text', 'text': 'Give me a number', 'state': 'done', 'provider_metadata': None}],
+            },
+            {
+                'id': IsStr(),
+                'role': 'assistant',
+                'metadata': None,
+                'parts': [{'type': 'text', 'text': 'Not a valid number', 'state': 'done', 'provider_metadata': None}],
+            },
+            {
+                'id': IsStr(),
+                'role': 'user',
+                'metadata': None,
+                'parts': [
+                    {
+                        'type': 'text',
+                        'text': """\
+Validation feedback:
+Output validation failed: expected integer
+
+Fix the errors and try again.\
+""",
+                        'state': 'done',
+                        'provider_metadata': None,
+                    }
+                ],
+            },
+        ]
+    )
+
+
 async def test_adapter_dump_messages_consecutive_text():
     """Test that consecutive text parts are concatenated correctly."""
     messages = [