Removing from_text and from_tool_call utilities that complicate snapshot testing (#744)

Author: sydney-runkle

docs/api/models/function.md

@@ -11,7 +11,7 @@ Here's a minimal example:
 
 ```py {title="function_model_usage.py" call_name="test_my_agent" noqa="I001"}
 from pydantic_ai import Agent
-from pydantic_ai.messages import ModelMessage, ModelResponse
+from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
 from pydantic_ai.models.function import FunctionModel, AgentInfo
 
 my_agent = Agent('openai:gpt-4o')
@@ -41,7 +41,7 @@ async def model_function(
         function_tools=[], allow_text_result=True, result_tools=[], model_settings=None
     )
     """
-    return ModelResponse.from_text('hello world')
+    return ModelResponse(parts=[TextPart('hello world')])
 
 
 async def test_my_agent():

docs/testing-evals.md

@@ -202,6 +202,7 @@ from pydantic_ai import models
 from pydantic_ai.messages import (
     ModelMessage,
     ModelResponse,
+    TextPart,
     ToolCallPart,
 )
 from pydantic_ai.models.function import AgentInfo, FunctionModel
@@ -229,7 +230,7 @@ def call_weather_forecast(  # (1)!
         # second call, return the forecast
         msg = messages[-1].parts[0]
         assert msg.part_kind == 'tool-return'
-        return ModelResponse.from_text(f'The forecast is: {msg.content}')
+        return ModelResponse(parts=[TextPart(f'The forecast is: {msg.content}')])
 
 
 async def test_forecast_future():

docs/tools.md

@@ -243,7 +243,7 @@ To demonstrate a tool's schema, here we use [`FunctionModel`][pydantic_ai.models
 
 ```python {title="tool_schema.py"}
 from pydantic_ai import Agent
-from pydantic_ai.messages import ModelMessage, ModelResponse
+from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
 from pydantic_ai.models.function import AgentInfo, FunctionModel
 
 agent = Agent()
@@ -283,7 +283,7 @@ def print_schema(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse
         'additionalProperties': False,
     }
     """
-    return ModelResponse.from_text(content='foobar')
+    return ModelResponse(parts=[TextPart('foobar')])
 
 
 agent.run_sync('hello', model=FunctionModel(print_schema))

examples/pydantic_ai_examples/chat_app.py

@@ -128,7 +128,7 @@ async def stream_messages():
             async for text in result.stream(debounce_by=0.01):
                 # text here is a `str` and the frontend wants
                 # JSON encoded ModelResponse, so we create one
-                m = ModelResponse.from_text(content=text, timestamp=result.timestamp())
+                m = ModelResponse(parts=[TextPart(text)], timestamp=result.timestamp())
                 yield json.dumps(to_chat_message(m)).encode('utf-8') + b'\n'
 
         # add new messages (e.g. the user prompt and the agent response in this case) to the database

pydantic_ai_slim/pydantic_ai/messages.py

@@ -264,16 +264,6 @@ class ModelResponse:
     kind: Literal['response'] = 'response'
     """Message type identifier, this is available on all parts as a discriminator."""
 
-    @classmethod
-    def from_text(cls, content: str, model_name: str | None = None, timestamp: datetime | None = None) -> Self:
-        """Create a `ModelResponse` containing a single `TextPart`."""
-        return cls([TextPart(content=content)], model_name=model_name, timestamp=timestamp or _now_utc())
-
-    @classmethod
-    def from_tool_call(cls, tool_call: ToolCallPart, model_name: str | None = None) -> Self:
-        """Create a `ModelResponse` containing a single `ToolCallPart`."""
-        return cls([tool_call], model_name=model_name)
-
 
 ModelMessage = Annotated[Union[ModelRequest, ModelResponse], pydantic.Discriminator('kind')]
 """Any message sent to or returned by a model."""

pydantic_ai_slim/pydantic_ai/models/test.py

@@ -191,11 +191,13 @@ def _request(self, messages: list[ModelMessage], model_settings: ModelSettings |
                             if isinstance(part, ToolReturnPart):
                                 output[part.tool_name] = part.content
                 if output:
-                    return ModelResponse.from_text(pydantic_core.to_json(output).decode(), model_name=self.model_name)
+                    return ModelResponse(
+                        parts=[TextPart(pydantic_core.to_json(output).decode())], model_name=self.model_name
+                    )
                 else:
-                    return ModelResponse.from_text('success (no tool calls)', model_name=self.model_name)
+                    return ModelResponse(parts=[TextPart('success (no tool calls)')], model_name=self.model_name)
             else:
-                return ModelResponse.from_text(response_text.value, model_name=self.model_name)
+                return ModelResponse(parts=[TextPart(response_text.value)], model_name=self.model_name)
         else:
             assert self.result_tools, 'No result tools provided'
             custom_result_args = self.result.right

pydantic_ai_slim/pydantic_ai/result.py

@@ -270,7 +270,10 @@ async def _stream_text_deltas() -> AsyncIterator[str]:
 
                 lf_span.set_attribute('combined_text', combined_validated_text)
                 await self._marked_completed(
-                    _messages.ModelResponse.from_text(combined_validated_text, self._stream_response.model_name())
+                    _messages.ModelResponse(
+                        parts=[_messages.TextPart(combined_validated_text)],
+                        model_name=self._stream_response.model_name(),
+                    )
                 )
 
     async def stream_structured(

tests/models/test_cohere.py

@@ -14,6 +14,7 @@
     ModelResponse,
     RetryPromptPart,
     SystemPromptPart,
+    TextPart,
     ToolCallPart,
     ToolReturnPart,
     UserPromptPart,
@@ -101,12 +102,12 @@ async def test_request_simple_success(allow_model_requests: None):
     assert result.all_messages() == snapshot(
         [
             ModelRequest(parts=[UserPromptPart(content='hello', timestamp=IsNow(tz=timezone.utc))]),
-            ModelResponse.from_text(
-                content='world', model_name='command-r7b-12-2024', timestamp=IsNow(tz=timezone.utc)
+            ModelResponse(
+                parts=[TextPart('world')], model_name='command-r7b-12-2024', timestamp=IsNow(tz=timezone.utc)
             ),
             ModelRequest(parts=[UserPromptPart(content='hello', timestamp=IsNow(tz=timezone.utc))]),
-            ModelResponse.from_text(
-                content='world', model_name='command-r7b-12-2024', timestamp=IsNow(tz=timezone.utc)
+            ModelResponse(
+                parts=[TextPart('world')], model_name='command-r7b-12-2024', timestamp=IsNow(tz=timezone.utc)
             ),
         ]
     )
@@ -294,8 +295,8 @@ async def get_location(loc_name: str) -> str:
                     )
                 ]
             ),
-            ModelResponse.from_text(
-                content='final response', model_name='command-r7b-12-2024', timestamp=IsNow(tz=timezone.utc)
+            ModelResponse(
+                parts=[TextPart('final response')], model_name='command-r7b-12-2024', timestamp=IsNow(tz=timezone.utc)
             ),
         ]
     )

tests/models/test_gemini.py

@@ -435,7 +435,7 @@ def example_usage() -> _GeminiUsageMetaData:
 
 
 async def test_text_success(get_gemini_client: GetGeminiClient):
-    response = gemini_response(_content_model_response(ModelResponse.from_text('Hello world')))
+    response = gemini_response(_content_model_response(ModelResponse(parts=[TextPart('Hello world')])))
     gemini_client = get_gemini_client(response)
     m = GeminiModel('gemini-1.5-flash', http_client=gemini_client)
     agent = Agent(m)
@@ -525,7 +525,7 @@ async def test_request_tool_call(get_gemini_client: GetGeminiClient):
                 )
             )
         ),
-        gemini_response(_content_model_response(ModelResponse.from_text('final response'))),
+        gemini_response(_content_model_response(ModelResponse(parts=[TextPart('final response')]))),
     ]
     gemini_client = get_gemini_client(responses)
     m = GeminiModel('gemini-1.5-flash', http_client=gemini_client)
@@ -621,8 +621,8 @@ def handler(_: httpx.Request):
 
 async def test_stream_text(get_gemini_client: GetGeminiClient):
     responses = [
-        gemini_response(_content_model_response(ModelResponse.from_text('Hello '))),
-        gemini_response(_content_model_response(ModelResponse.from_text('world'))),
+        gemini_response(_content_model_response(ModelResponse(parts=[TextPart('Hello ')]))),
+        gemini_response(_content_model_response(ModelResponse(parts=[TextPart('world')]))),
     ]
     json_data = _gemini_streamed_response_ta.dump_json(responses, by_alias=True)
     stream = AsyncByteStreamList([json_data[:100], json_data[100:200], json_data[200:]])
@@ -763,7 +763,7 @@ async def bar(y: str) -> str:
 
 async def test_stream_text_heterogeneous(get_gemini_client: GetGeminiClient):
     responses = [
-        gemini_response(_content_model_response(ModelResponse.from_text('Hello '))),
+        gemini_response(_content_model_response(ModelResponse(parts=[TextPart('Hello ')]))),
         gemini_response(
             _GeminiContent(
                 role='model',

tests/models/test_model_function.py

@@ -31,7 +31,7 @@
 
 
 def hello(_messages: list[ModelMessage], _agent_info: AgentInfo) -> ModelResponse:
-    return ModelResponse.from_text('hello world')  # pragma: no cover
+    return ModelResponse(parts=[TextPart('hello world')])  # pragma: no cover
 
 
 async def stream_hello(_messages: list[ModelMessage], _agent_info: AgentInfo) -> AsyncIterator[str]:
@@ -55,7 +55,7 @@ async def return_last(messages: list[ModelMessage], _: AgentInfo) -> ModelRespon
     response = asdict(last)
     response.pop('timestamp', None)
     response['message_count'] = len(messages)
-    return ModelResponse.from_text(' '.join(f'{k}={v!r}' for k, v in response.items()))
+    return ModelResponse(parts=[TextPart(' '.join(f'{k}={v!r}' for k, v in response.items()))])
 
 
 def test_simple(set_event_loop: None):
@@ -117,7 +117,7 @@ async def weather_model(messages: list[ModelMessage], info: AgentInfo) -> ModelR
                     break
 
             assert location_name is not None
-            return ModelResponse.from_text(f'{last.content} in {location_name}')
+            return ModelResponse(parts=[TextPart(f'{last.content} in {location_name}')])
 
     raise ValueError(f'Unexpected message: {last}')
 
@@ -200,7 +200,7 @@ async def call_function_model(messages: list[ModelMessage], _: AgentInfo) -> Mod
                 ]
             )
     elif isinstance(last, ToolReturnPart):
-        return ModelResponse.from_text(pydantic_core.to_json(last).decode())
+        return ModelResponse(parts=[TextPart(pydantic_core.to_json(last).decode())])
 
     raise ValueError(f'Unexpected message: {last}')
 
@@ -236,7 +236,7 @@ async def call_tool(messages: list[ModelMessage], info: AgentInfo) -> ModelRespo
         tool_name = info.function_tools[0].name
         return ModelResponse(parts=[ToolCallPart.from_raw_args(tool_name, '{}')])
     else:
-        return ModelResponse.from_text('final response')
+        return ModelResponse(parts=[TextPart('final response')])
 
 
 def test_deps_none(set_event_loop: None):
@@ -318,8 +318,12 @@ def spam() -> str:
 
 def test_register_all(set_event_loop: None):
     async def f(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
-        return ModelResponse.from_text(
-            f'messages={len(messages)} allow_text_result={info.allow_text_result} tools={len(info.function_tools)}'
+        return ModelResponse(
+            parts=[
+                TextPart(
+                    f'messages={len(messages)} allow_text_result={info.allow_text_result} tools={len(info.function_tools)}'
+                )
+            ]
         )
 
     result = agent_all.run_sync('Hello', model=FunctionModel(f))
@@ -373,7 +377,7 @@ async def try_again(msgs_: list[ModelMessage], _agent_info: AgentInfo) -> ModelR
         nonlocal call_count
         call_count += 1
 
-        return ModelResponse.from_text(str(call_count))
+        return ModelResponse(parts=[TextPart(str(call_count))])
 
     agent = Agent(FunctionModel(try_again))