[FEAT] Support JSON Schema in Responses

pyrit/prompt_target/openai/openai_response_target.py

@@ -303,6 +303,29 @@ async def _construct_request_body(self, conversation: MutableSequence[Message],
         """
         input_items = await self._build_input_for_multi_modal_async(conversation)
 
+        text_format = None
+        if is_json_response:
+            if conversation[-1].message_pieces[0].prompt_metadata.get("json_schema"):
+                json_schema_str = str(conversation[-1].message_pieces[0].prompt_metadata["json_schema"])
+                try:
+                    json_schema = json.loads(json_schema_str)
+                except json.JSONDecodeError as e:
+                    raise PyritException(
+                        message=f"Failed to parse provided JSON schema for response_format as JSON.\n"
+                        f"Schema: {json_schema_str}\nFull error: {e}"
+                    )
+                text_format = {
+                    "format": {
+                        "type": "json_schema",
+                        "name": "CustomSchema",
+                        "schema": json_schema,
+                        "strict": True,
+                    }
+                }
+            else:
+                logger.info("Falling back to json_object; not recommended for new models")
+                text_format = {"format": {"type": "json_object"}}
+
         body_parameters = {
             "model": self._model_name,
             "max_output_tokens": self._max_output_tokens,
@@ -311,7 +334,7 @@ async def _construct_request_body(self, conversation: MutableSequence[Message],
             "stream": False,
             "input": input_items,
             # Correct JSON response format per Responses API
-            "response_format": {"type": "json_object"} if is_json_response else None,
+            "text": text_format,
         }
 
         if self._extra_body_parameters:

tests/integration/targets/test_openai_responses_gpt5.py

@@ -2,9 +2,11 @@
 # Licensed under the MIT license.
 
 
+import json
 import os
 import uuid
 
+import jsonschema
 import pytest
 
 from pyrit.models import MessagePiece
@@ -17,6 +19,7 @@ async def test_openai_responses_gpt5(sqlite_instance):
         "endpoint": os.getenv("AZURE_OPENAI_GPT5_RESPONSES_ENDPOINT"),
         "model_name": os.getenv("AZURE_OPENAI_GPT5_MODEL"),
         "api_key": os.getenv("AZURE_OPENAI_GPT5_RESPONSES_KEY"),
+        # "use_entra_auth": True,
     }
 
     target = OpenAIResponseTarget(**args)
@@ -46,3 +49,92 @@ async def test_openai_responses_gpt5(sqlite_instance):
     assert result.message_pieces[1].role == "assistant"
     # Hope that the model manages to give the correct answer somewhere (GPT-5 really should)
     assert "Paris" in result.message_pieces[1].converted_value
+
+
+@pytest.mark.asyncio
+async def test_openai_responses_gpt5_json_schema(sqlite_instance):
+    args = {
+        "endpoint": os.getenv("AZURE_OPENAI_GPT5_RESPONSES_ENDPOINT"),
+        "model_name": os.getenv("AZURE_OPENAI_GPT5_MODEL"),
+        "api_key": os.getenv("AZURE_OPENAI_GPT5_RESPONSES_KEY"),
+        # "use_entra_auth": True,
+    }
+
+    target = OpenAIResponseTarget(**args)
+
+    conv_id = str(uuid.uuid4())
+
+    developer_piece = MessagePiece(
+        role="developer",
+        original_value="You are an expert in the lore of cats.",
+        original_value_data_type="text",
+        conversation_id=conv_id,
+        attack_identifier={"id": str(uuid.uuid4())},
+    )
+    sqlite_instance.add_message_to_memory(request=developer_piece.to_message())
+
+    cat_schema = {
+        "type": "object",
+        "properties": {
+            "name": {"type": "string", "minLength": 12},
+            "age": {"type": "integer", "minimum": 0, "maximum": 20},
+            "colour": {
+                "type": "array",
+                "items": {"type": "integer", "minimum": 0, "maximum": 255},
+                "minItems": 3,
+                "maxItems": 3,
+            },
+        },
+        "required": ["name", "age", "colour"],
+        "additionalProperties": False,
+    }
+
+    user_piece = MessagePiece(
+        role="user",
+        original_value="Create a JSON object that describes a mystical cat with the following properties: name, age, colour.",
+        original_value_data_type="text",
+        conversation_id=conv_id,
+        prompt_metadata={"response_format": "json", "json_schema": json.dumps(cat_schema)},
+    )
+
+    response = await target.send_prompt_async(prompt_request=user_piece.to_message())
+
+    response_content = response.get_value(1)
+    response_json = json.loads(response_content)
+    jsonschema.validate(instance=response_json, schema=cat_schema)
+
+
+@pytest.mark.asyncio
+async def test_openai_responses_gpt5_json_object(sqlite_instance):
+    args = {
+        "endpoint": os.getenv("AZURE_OPENAI_GPT5_RESPONSES_ENDPOINT"),
+        "model_name": os.getenv("AZURE_OPENAI_GPT5_MODEL"),
+        "api_key": os.getenv("AZURE_OPENAI_GPT5_RESPONSES_KEY"),
+        # "use_entra_auth": True,
+    }
+
+    target = OpenAIResponseTarget(**args)
+
+    conv_id = str(uuid.uuid4())
+
+    developer_piece = MessagePiece(
+        role="developer",
+        original_value="You are an expert in the lore of cats.",
+        original_value_data_type="text",
+        conversation_id=conv_id,
+        attack_identifier={"id": str(uuid.uuid4())},
+    )
+    sqlite_instance.add_message_to_memory(request=developer_piece.to_message())
+    user_piece = MessagePiece(
+        role="user",
+        original_value="Create a JSON object that describes a mystical cat with the following properties: name, age, colour.",
+        original_value_data_type="text",
+        conversation_id=conv_id,
+        prompt_metadata={"response_format": "json"},
+    )
+    response = await target.send_prompt_async(prompt_request=user_piece.to_message())
+
+    response_content = response.get_value(1)
+    response_json = json.loads(response_content)
+    assert response_json is not None
+    # Can't assert more, since the failure could be due to a bad generation by the model