Type ModelRequest.parts and ModelResponse.parts as Sequence (#2798)

Author: moritzwilksch

pydantic_ai_slim/pydantic_ai/_a2a.py

@@ -272,7 +272,7 @@ def _response_parts_from_a2a(self, parts: list[Part]) -> list[ModelResponsePart]
                 assert_never(part)
         return model_parts
 
-    def _response_parts_to_a2a(self, parts: list[ModelResponsePart]) -> list[Part]:
+    def _response_parts_to_a2a(self, parts: Sequence[ModelResponsePart]) -> list[Part]:
         """Convert pydantic-ai ModelResponsePart objects to A2A Part objects.
 
         This handles the conversion from pydantic-ai's internal response parts to

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -301,16 +301,21 @@ async def _reevaluate_dynamic_prompts(
         if self.system_prompt_dynamic_functions:
             for msg in messages:
                 if isinstance(msg, _messages.ModelRequest):
-                    for i, part in enumerate(msg.parts):
+                    reevaluated_message_parts: list[_messages.ModelRequestPart] = []
+                    for part in msg.parts:
                         if isinstance(part, _messages.SystemPromptPart) and part.dynamic_ref:
                             # Look up the runner by its ref
                             if runner := self.system_prompt_dynamic_functions.get(  # pragma: lax no cover
                                 part.dynamic_ref
                             ):
                                 updated_part_content = await runner.run(run_context)
-                                msg.parts[i] = _messages.SystemPromptPart(
-                                    updated_part_content, dynamic_ref=part.dynamic_ref
-                                )
+                                part = _messages.SystemPromptPart(updated_part_content, dynamic_ref=part.dynamic_ref)
+
+                        reevaluated_message_parts.append(part)
+
+                    # Replace message parts with reevaluated ones to prevent mutating parts list
+                    if reevaluated_message_parts != msg.parts:
+                        msg.parts = reevaluated_message_parts
 
     async def _sys_parts(self, run_context: RunContext[DepsT]) -> list[_messages.ModelRequestPart]:
         """Build the initial messages for the conversation."""

pydantic_ai_slim/pydantic_ai/messages.py

@@ -833,7 +833,7 @@ def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_me
 class ModelRequest:
     """A request generated by Pydantic AI and sent to a model, e.g. a message from the Pydantic AI app to the model."""
 
-    parts: list[ModelRequestPart]
+    parts: Sequence[ModelRequestPart]
     """The parts of the user message."""
 
     _: KW_ONLY
@@ -988,7 +988,7 @@ class BuiltinToolCallPart(BaseToolCallPart):
 class ModelResponse:
     """A response from a model, e.g. a message from the model to the Pydantic AI app."""
 
-    parts: list[ModelResponsePart]
+    parts: Sequence[ModelResponsePart]
     """The parts of the model message."""
 
     _: KW_ONLY

tests/test_agent.py

@@ -2993,6 +2993,27 @@ async def func():
     )
 
 
+def test_dynamic_system_prompt_no_changes():
+    """Test coverage for _reevaluate_dynamic_prompts branch where no parts are changed
+    and the messages loop continues after replacement of parts.
+    """
+    agent = Agent('test')
+
+    @agent.system_prompt(dynamic=True)
+    async def dynamic_func() -> str:
+        return 'Dynamic'
+
+    result1 = agent.run_sync('Hello')
+
+    # Create ModelRequest with non-dynamic SystemPromptPart (no dynamic_ref)
+    manual_request = ModelRequest(parts=[SystemPromptPart(content='Static'), UserPromptPart(content='Manual')])
+
+    # Mix dynamic and non-dynamic messages to trigger branch coverage
+    result2 = agent.run_sync('Second call', message_history=result1.all_messages() + [manual_request])
+
+    assert result2.output == 'success (no tool calls)'
+
+
 def test_capture_run_messages_tool_agent() -> None:
     agent_outer = Agent('test')
     agent_inner = Agent(TestModel(custom_output_text='inner agent result'))