Raise UserError when tool system_prompt returns a None

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -190,6 +190,7 @@ async def _reevaluate_dynamic_prompts(
                             # Look up the runner by its ref
                             if runner := self.system_prompt_dynamic_functions.get(part.dynamic_ref):
                                 updated_part_content = await runner.run(run_context)
+
                                 msg.parts[i] = _messages.SystemPromptPart(
                                     updated_part_content, dynamic_ref=part.dynamic_ref
                                 )
@@ -199,6 +200,10 @@ async def _sys_parts(self, run_context: RunContext[DepsT]) -> list[_messages.Mod
         messages: list[_messages.ModelRequestPart] = [_messages.SystemPromptPart(p) for p in self.system_prompts]
         for sys_prompt_runner in self.system_prompt_functions:
             prompt = await sys_prompt_runner.run(run_context)
+            if prompt is None:  # type: ignore[comparison-overlap]
+                raise exceptions.UserError(f'system prompt function {sys_prompt_runner.function} returned None')
+            if not isinstance(prompt, str):
+                prompt = str(prompt)
             if sys_prompt_runner.dynamic:
                 messages.append(_messages.SystemPromptPart(prompt, dynamic_ref=sys_prompt_runner.function.__qualname__))
             else:

tests/test_agent.py

@@ -1353,6 +1353,47 @@ def test_double_capture_run_messages() -> None:
     )
 
 
+@pytest.mark.parametrize(
+    'return_value,valid',
+    [
+        ('', True),
+        (None, False),
+        (5, True),  ## TODO - decide if this is valid
+    ],
+)
+@pytest.mark.parametrize('dynamic', [True, False])
+def test_system_prompt_egde_cases(dynamic: bool, return_value: Any, valid: bool):
+    """Tests edge cases for system prompt.
+    i.e: SystemPromptPart(
+            content="A",       <--- Remains the same when `message_history` is passed.
+        part_kind='system-prompt')
+    """
+    agent = Agent('test', system_prompt='repeat what user says')
+
+    @agent.system_prompt(dynamic=dynamic)
+    async def func() -> str:
+        return return_value
+
+    if valid:
+        res = agent.run_sync('Hello')
+        model_request = res.all_messages()[0]
+        system_messages = [m for m in model_request.parts if isinstance(m, SystemPromptPart)]
+        last_system_message = system_messages[-1]
+        assert last_system_message.content == str(return_value)
+    else:
+        with pytest.raises(UserError, match=f'returned {return_value}'):
+            agent.run_sync('Hello')
+
+    if dynamic:
+        if return_value is None:
+            return_value = ''
+            res = agent.run_sync('Hello')
+        else:
+            return_value = None
+            with pytest.raises(UserError, match=f'returned {return_value}'):
+                agent.run_sync('Hello')
+
+
 def test_dynamic_false_no_reevaluate():
     """When dynamic is false (default), the system prompt is not reevaluated
     i.e: SystemPromptPart(