Feat: Content Filtering Exception Handling

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -603,6 +603,12 @@ async def _run_stream() -> AsyncIterator[_messages.HandleResponseEvent]:  # noqa
                             f'Model token limit ({max_tokens or "provider default"}) exceeded before any response was generated. Increase the `max_tokens` model setting, or simplify the prompt to result in a shorter response that will fit within the limit.'
                         )
 
+                    # Check for content filter on empty response
+                    if self.model_response.finish_reason == 'content_filter':
+                        raise exceptions.ContentFilterError(
+                            f'Content filter triggered for model {self.model_response.model_name}'
+                        )
+
                     # we got an empty response.
                     # this sometimes happens with anthropic (and perhaps other models)
                     # when the model has already returned text along side tool calls

pydantic_ai_slim/pydantic_ai/exceptions.py

@@ -24,6 +24,7 @@
     'UsageLimitExceeded',
     'ModelAPIError',
     'ModelHTTPError',
+    'ContentFilterError',
     'IncompleteToolCall',
     'FallbackExceptionGroup',
 )
@@ -152,6 +153,10 @@ def __str__(self) -> str:
             return self.message
 
 
+class ContentFilterError(UnexpectedModelBehavior):
+    """Raised when content filtering is triggered by the model provider."""
+
+
 class ModelAPIError(AgentRunError):
     """Raised when a model provider API request fails."""
 

pydantic_ai_slim/pydantic_ai/models/google.py

@@ -503,11 +503,7 @@ def _process_response(self, response: GenerateContentResponse) -> ModelResponse:
             finish_reason = _FINISH_REASON_MAP.get(raw_finish_reason)
 
         if candidate.content is None or candidate.content.parts is None:
-            if finish_reason == 'content_filter' and raw_finish_reason:
-                raise UnexpectedModelBehavior(
-                    f'Content filter {raw_finish_reason.value!r} triggered', response.model_dump_json()
-                )
-            parts = []  # pragma: no cover
+            parts = []
         else:
             parts = candidate.content.parts or []
 
@@ -707,12 +703,7 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                 yield self._parts_manager.handle_part(vendor_part_id=uuid4(), part=web_fetch_return)
 
             if candidate.content is None or candidate.content.parts is None:
-                if self.finish_reason == 'content_filter' and raw_finish_reason:  # pragma: no cover
-                    raise UnexpectedModelBehavior(
-                        f'Content filter {raw_finish_reason.value!r} triggered', chunk.model_dump_json()
-                    )
-                else:  # pragma: no cover
-                    continue
+                continue
 
             parts = candidate.content.parts
             if not parts:

pydantic_ai_slim/pydantic_ai/models/openai.py

@@ -176,6 +176,20 @@ def _resolve_openai_image_generation_size(
     return mapped_size
 
 
+def _check_azure_content_filter(e: APIStatusError) -> bool:
+    """Check if the error is an Azure content filter error."""
+    if e.status_code == 400:
+        body_any: Any = e.body
+
+        if isinstance(body_any, dict):
+            body_dict = cast(dict[str, Any], body_any)
+
+            if (error := body_dict.get('error')) and isinstance(error, dict):
+                error_dict = cast(dict[str, Any], error)
+                return error_dict.get('code') == 'content_filter'
+    return False
+
+
 class OpenAIChatModelSettings(ModelSettings, total=False):
     """Settings used for an OpenAI model request."""
 
@@ -584,6 +598,20 @@ async def _completions_create(
                 extra_body=model_settings.get('extra_body'),
             )
         except APIStatusError as e:
+            if _check_azure_content_filter(e):
+                return chat.ChatCompletion(
+                    id='content_filter',
+                    choices=[
+                        chat.chat_completion.Choice(
+                            finish_reason='content_filter',
+                            index=0,
+                            message=chat.ChatCompletionMessage(content=None, role='assistant'),
+                        )
+                    ],
+                    created=0,
+                    model=self.model_name,
+                    object='chat.completion',
+                )
             if (status_code := e.status_code) >= 400:
                 raise ModelHTTPError(status_code=status_code, model_name=self.model_name, body=e.body) from e
             raise  # pragma: lax no cover
@@ -631,6 +659,7 @@ def _process_response(self, response: chat.ChatCompletion | str) -> ModelRespons
             raise UnexpectedModelBehavior(f'Invalid response from {self.system} chat completions endpoint: {e}') from e
 
         choice = response.choices[0]
+
         items: list[ModelResponsePart] = []
 
         if thinking_parts := self._process_thinking(choice.message):
@@ -1431,6 +1460,19 @@ async def _responses_create(  # noqa: C901
                 extra_body=model_settings.get('extra_body'),
             )
         except APIStatusError as e:
+            if _check_azure_content_filter(e):
+                return responses.Response(
+                    id='content_filter',
+                    model=self.model_name,
+                    created_at=0,
+                    object='response',
+                    status='incomplete',
+                    incomplete_details={'reason': 'content_filter'},  # type: ignore
+                    output=[],
+                    parallel_tool_calls=False,
+                    tool_choice='auto',
+                    tools=[],
+                )
             if (status_code := e.status_code) >= 400:
                 raise ModelHTTPError(status_code=status_code, model_name=self.model_name, body=e.body) from e
             raise  # pragma: lax no cover
@@ -2089,6 +2131,7 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                 raw_finish_reason = (
                     details.reason if (details := chunk.response.incomplete_details) else chunk.response.status
                 )
+
                 if raw_finish_reason:  # pragma: no branch
                     self.provider_details = {'finish_reason': raw_finish_reason}
                     self.finish_reason = _RESPONSES_FINISH_REASON_MAP.get(raw_finish_reason)

tests/models/test_google.py

@@ -54,7 +54,13 @@
     WebFetchTool,
     WebSearchTool,
 )
-from pydantic_ai.exceptions import ModelAPIError, ModelHTTPError, ModelRetry, UnexpectedModelBehavior, UserError
+from pydantic_ai.exceptions import (
+    ContentFilterError,
+    ModelAPIError,
+    ModelHTTPError,
+    ModelRetry,
+    UserError,
+)
 from pydantic_ai.messages import (
     BuiltinToolCallEvent,  # pyright: ignore[reportDeprecated]
     BuiltinToolResultEvent,  # pyright: ignore[reportDeprecated]
@@ -994,7 +1000,10 @@ async def test_google_model_safety_settings(allow_model_requests: None, google_p
     )
     agent = Agent(m, instructions='You hate the world!', model_settings=settings)
 
-    with pytest.raises(UnexpectedModelBehavior, match="Content filter 'SAFETY' triggered"):
+    with pytest.raises(
+        ContentFilterError,
+        match='Content filter triggered for model gemini-1.5-flash',
+    ):
         await agent.run('Tell me a joke about a Brazilians.')
 
 
@@ -4610,3 +4619,57 @@ def get_country() -> str:
             ),
         ]
     )
+
+
+async def test_google_stream_empty_chunk(
+    allow_model_requests: None, google_provider: GoogleProvider, mocker: MockerFixture
+):
+    """Test that empty chunks in the stream are ignored (coverage for continue)."""
+    model_name = 'gemini-2.5-flash'
+    model = GoogleModel(model_name, provider=google_provider)
+
+    # Chunk with NO content
+    empty_candidate = mocker.Mock(finish_reason=None, content=None)
+    empty_candidate.grounding_metadata = None
+    empty_candidate.url_context_metadata = None
+
+    chunk_empty = mocker.Mock(
+        candidates=[empty_candidate], model_version=model_name, usage_metadata=None, create_time=datetime.datetime.now()
+    )
+    chunk_empty.model_dump_json.return_value = '{}'
+
+    # Chunk WITH content (valid)
+    part_mock = mocker.Mock(
+        text='Hello',
+        thought=False,
+        function_call=None,
+        inline_data=None,
+        executable_code=None,
+        code_execution_result=None,
+    )
+    part_mock.thought_signature = None
+
+    valid_candidate = mocker.Mock(
+        finish_reason=GoogleFinishReason.STOP,
+        content=mocker.Mock(parts=[part_mock]),
+        grounding_metadata=None,
+        url_context_metadata=None,
+    )
+
+    chunk_valid = mocker.Mock(
+        candidates=[valid_candidate], model_version=model_name, usage_metadata=None, create_time=datetime.datetime.now()
+    )
+    chunk_valid.model_dump_json.return_value = '{"content": "Hello"}'
+
+    async def stream_iterator():
+        yield chunk_empty
+        yield chunk_valid
+
+    mocker.patch.object(model.client.aio.models, 'generate_content_stream', return_value=stream_iterator())
+
+    agent = Agent(model=model)
+
+    async with agent.run_stream('hello') as result:
+        output = await result.get_output()
+
+    assert output == 'Hello'

tests/models/test_model_function.py

@@ -22,6 +22,7 @@
     ToolReturnPart,
     UserPromptPart,
 )
+from pydantic_ai.exceptions import ContentFilterError
 from pydantic_ai.models.function import AgentInfo, DeltaToolCall, DeltaToolCalls, FunctionModel
 from pydantic_ai.models.test import TestModel
 from pydantic_ai.result import RunUsage
@@ -538,3 +539,43 @@ async def test_return_empty():
     with pytest.raises(ValueError, match='Stream function must return at least one item'):
         async with agent.run_stream(''):
             pass
+
+
+async def test_central_content_filter_handling():
+    """
+    Test that the agent graph correctly raises ContentFilterError
+    when a model returns finish_reason='content_filter' AND empty content.
+    """
+
+    async def filtered_response(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        # Simulate a model response that was blocked completely
+        return ModelResponse(
+            parts=[],  # Empty parts triggers the exception
+            model_name='test-model',
+            finish_reason='content_filter',
+        )
+
+    model = FunctionModel(function=filtered_response, model_name='test-model')
+    agent = Agent(model)
+
+    with pytest.raises(ContentFilterError, match='Content filter triggered for model test-model'):
+        await agent.run('Trigger filter')
+
+
+async def test_central_content_filter_with_partial_content():
+    """
+    Test that the agent graph returns partial content (does not raise exception)
+    even if finish_reason='content_filter', provided parts are not empty.
+    """
+
+    async def filtered_response(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        return ModelResponse(
+            parts=[TextPart('Partially generated content...')], model_name='test-model', finish_reason='content_filter'
+        )
+
+    model = FunctionModel(function=filtered_response, model_name='test-model')
+    agent = Agent(model)
+
+    # Should NOT raise ContentFilterError
+    result = await agent.run('Trigger filter')
+    assert result.output == 'Partially generated content...'

tests/models/test_openai.py

@@ -38,6 +38,7 @@
 )
 from pydantic_ai._json_schema import InlineDefsJsonSchemaTransformer
 from pydantic_ai.builtin_tools import WebSearchTool
+from pydantic_ai.exceptions import ContentFilterError
 from pydantic_ai.models import ModelRequestParameters
 from pydantic_ai.output import NativeOutput, PromptedOutput, TextOutput, ToolOutput
 from pydantic_ai.profiles.openai import OpenAIModelProfile, openai_model_profile
@@ -3325,3 +3326,104 @@ async def test_openai_reasoning_in_thinking_tags(allow_model_requests: None):
 """,
         }
     )
+
+
+def test_azure_prompt_filter_error(allow_model_requests: None) -> None:
+    mock_client = MockOpenAI.create_mock(
+        APIStatusError(
+            'content filter',
+            response=httpx.Response(status_code=400, request=httpx.Request('POST', 'https://example.com/v1')),
+            body={'error': {'code': 'content_filter', 'message': 'The content was filtered.'}},
+        )
+    )
+
+    m = OpenAIChatModel('gpt-5-mini', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m)
+
+    with pytest.raises(ContentFilterError, match=r'Content filter triggered for model gpt-5-mini'):
+        agent.run_sync('bad prompt')
+
+
+def test_responses_azure_prompt_filter_error(allow_model_requests: None) -> None:
+    mock_client = MockOpenAIResponses.create_mock(
+        APIStatusError(
+            'content filter',
+            response=httpx.Response(status_code=400, request=httpx.Request('POST', 'https://example.com/v1')),
+            body={'error': {'code': 'content_filter', 'message': 'The content was filtered.'}},
+        )
+    )
+    m = OpenAIResponsesModel('gpt-5-mini', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m)
+
+    with pytest.raises(ContentFilterError, match=r'Content filter triggered for model gpt-5-mini'):
+        agent.run_sync('bad prompt')
+
+
+async def test_openai_response_filter_error_sync(allow_model_requests: None):
+    """Test that ContentFilterError is raised when response is empty and finish_reason is content_filter."""
+    c = completion_message(
+        ChatCompletionMessage(content=None, role='assistant'),
+    )
+    c.choices[0].finish_reason = 'content_filter'
+    c.model = 'gpt-5-mini'
+
+    mock_client = MockOpenAI.create_mock(c)
+    m = OpenAIChatModel('gpt-5-mini', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m)
+
+    with pytest.raises(ContentFilterError, match=r'Content filter triggered for model gpt-5-mini'):
+        await agent.run('hello')
+
+
+async def test_openai_response_filter_with_partial_content(allow_model_requests: None):
+    """Test that NO exception is raised if content is returned, even if finish_reason is content_filter."""
+    c = completion_message(
+        ChatCompletionMessage(content='Partial content', role='assistant'),
+    )
+    c.choices[0].finish_reason = 'content_filter'
+
+    mock_client = MockOpenAI.create_mock(c)
+    m = OpenAIChatModel('gpt-5-mini', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m)
+
+    # Should NOT raise ContentFilterError
+    result = await agent.run('hello')
+    assert result.output == 'Partial content'
+
+
+def test_openai_400_non_content_filter(allow_model_requests: None) -> None:
+    """Test a 400 error that is NOT a content filter (different code)."""
+    mock_client = MockOpenAI.create_mock(
+        APIStatusError(
+            'Bad Request',
+            response=httpx.Response(status_code=400, request=httpx.Request('POST', 'https://api.openai.com/v1')),
+            body={'error': {'code': 'invalid_parameter', 'message': 'Invalid param.'}},
+        )
+    )
+    m = OpenAIChatModel('gpt-5-mini', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m)
+
+    with pytest.raises(ModelHTTPError) as exc_info:
+        agent.run_sync('hello')
+
+    # Should be ModelHTTPError, NOT ContentFilterError
+    assert not isinstance(exc_info.value, ContentFilterError)
+    assert exc_info.value.status_code == 400
+
+
+def test_openai_400_non_dict_body(allow_model_requests: None) -> None:
+    """Test a 400 error where the body is not a dictionary."""
+    mock_client = MockOpenAI.create_mock(
+        APIStatusError(
+            'Bad Request',
+            response=httpx.Response(status_code=400, request=httpx.Request('POST', 'https://api.openai.com/v1')),
+            body='Raw string body',
+        )
+    )
+    m = OpenAIChatModel('gpt-5-mini', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m)
+
+    with pytest.raises(ModelHTTPError) as exc_info:
+        agent.run_sync('hello')
+
+    assert exc_info.value.status_code == 400