Python: Fix runtime response format for responses client (#2440)

* Fix runtime response format for responses client

* Handle run time schema for Azure AI Client.

python/packages/azure-ai/agent_framework_azure_ai/_client.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import sys
-from collections.abc import MutableSequence
+from collections.abc import Mapping, MutableSequence
 from typing import Any, ClassVar, TypeVar
 
 from agent_framework import (
@@ -14,7 +14,7 @@ from agent_framework import (
     use_chat_middleware,
     use_function_invocation,
 )
-from agent_framework.exceptions import ServiceInitializationError
+from agent_framework.exceptions import ServiceInitializationError, ServiceInvalidRequestError
 from agent_framework.observability import use_observability
 from agent_framework.openai._responses_client import OpenAIBaseResponsesClient
 from azure.ai.projects.aio import AIProjectClient
@@ -22,7 +22,9 @@ from azure.ai.projects.models import (
     MCPTool,
     PromptAgentDefinition,
     PromptAgentDefinitionText,
+    ResponseTextFormatConfigurationJsonObject,
     ResponseTextFormatConfigurationJsonSchema,
+    ResponseTextFormatConfigurationText,
 )
 from azure.core.credentials_async import AsyncTokenCredential
 from azure.core.exceptions import ResourceNotFoundError
@@ -188,6 +190,40 @@ class AzureAIClient(OpenAIBaseResponsesClient):
         """Close the project_client."""
         await self._close_client_if_needed()
 
+    def _create_text_format_config(
+        self, response_format: Any
+    ) -> (
+        ResponseTextFormatConfigurationJsonSchema
+        | ResponseTextFormatConfigurationJsonObject
+        | ResponseTextFormatConfigurationText
+    ):
+        """Convert response_format into Azure text format configuration."""
+        if isinstance(response_format, type) and issubclass(response_format, BaseModel):
+            return ResponseTextFormatConfigurationJsonSchema(
+                name=response_format.__name__,
+                schema=response_format.model_json_schema(),
+            )
+
+        if isinstance(response_format, Mapping):
+            format_config = self._convert_response_format(response_format)
+            format_type = format_config.get("type")
+            if format_type == "json_schema":
+                config_kwargs: dict[str, Any] = {
+                    "name": format_config.get("name") or "response",
+                    "schema": format_config["schema"],
+                }
+                if "strict" in format_config:
+                    config_kwargs["strict"] = format_config["strict"]
+                if "description" in format_config:
+                    config_kwargs["description"] = format_config["description"]
+                return ResponseTextFormatConfigurationJsonSchema(**config_kwargs)
+            if format_type == "json_object":
+                return ResponseTextFormatConfigurationJsonObject()
+            if format_type == "text":
+                return ResponseTextFormatConfigurationText()
+
+        raise ServiceInvalidRequestError("response_format must be a Pydantic model or mapping.")
+
     async def _get_agent_reference_or_create(
         self, run_options: dict[str, Any], messages_instructions: str | None
     ) -> dict[str, str]:
@@ -228,12 +264,7 @@ class AzureAIClient(OpenAIBaseResponsesClient):
 
             if "response_format" in run_options:
                 response_format = run_options["response_format"]
-                args["text"] = PromptAgentDefinitionText(
-                    format=ResponseTextFormatConfigurationJsonSchema(
-                        name=response_format.__name__,
-                        schema=response_format.model_json_schema(),
-                    )
-                )
+                args["text"] = PromptAgentDefinitionText(format=self._create_text_format_config(response_format))
 
             # Combine instructions from messages and options
             combined_instructions = [

python/packages/azure-ai/tests/test_azure_ai_client.py

@@ -561,6 +561,56 @@ async def test_azure_ai_client_agent_creation_with_response_format(
     assert "description" in schema["properties"]
 
 
+async def test_azure_ai_client_agent_creation_with_mapping_response_format(
+    mock_project_client: MagicMock,
+) -> None:
+    """Test agent creation when response_format is provided as a mapping."""
+    client = create_test_azure_ai_client(mock_project_client, agent_name="test-agent")
+
+    mock_agent = MagicMock()
+    mock_agent.name = "test-agent"
+    mock_agent.version = "1.0"
+    mock_project_client.agents.create_version = AsyncMock(return_value=mock_agent)
+
+    runtime_schema = {
+        "title": "WeatherDigest",
+        "type": "object",
+        "properties": {
+            "location": {"type": "string"},
+            "conditions": {"type": "string"},
+            "temperature_c": {"type": "number"},
+            "advisory": {"type": "string"},
+        },
+        "required": ["location", "conditions", "temperature_c", "advisory"],
+        "additionalProperties": False,
+    }
+
+    run_options = {
+        "model": "test-model",
+        "response_format": {
+            "type": "json_schema",
+            "json_schema": {
+                "name": runtime_schema["title"],
+                "strict": True,
+                "schema": runtime_schema,
+            },
+        },
+    }
+
+    await client._get_agent_reference_or_create(run_options, None)  # type: ignore
+
+    call_args = mock_project_client.agents.create_version.call_args
+    created_definition = call_args[1]["definition"]
+
+    assert hasattr(created_definition, "text")
+    assert created_definition.text is not None
+    format_config = created_definition.text.format
+    assert isinstance(format_config, ResponseTextFormatConfigurationJsonSchema)
+    assert format_config.name == runtime_schema["title"]
+    assert format_config.schema == runtime_schema
+    assert format_config.strict is True
+
+
 async def test_azure_ai_client_prepare_options_excludes_response_format(
     mock_project_client: MagicMock,
 ) -> None:

python/packages/core/agent_framework/openai/_responses_client.py

@@ -91,18 +91,21 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
     ) -> ChatResponse:
         client = await self.ensure_client()
         run_options = await self.prepare_options(messages, chat_options)
+        response_format = run_options.pop("response_format", None)
+        text_config = run_options.pop("text", None)
+        text_format, text_config = self._prepare_text_config(response_format=response_format, text_config=text_config)
+        if text_config:
+            run_options["text"] = text_config
         try:
-            response_format = run_options.pop("response_format", None)
-            if not response_format:
+            if not text_format:
                 response = await client.responses.create(
                     stream=False,
                     **run_options,
                 )
                 chat_options.conversation_id = self.get_conversation_id(response, chat_options.store)
                 return self._create_response_content(response, chat_options=chat_options)
-            # create call does not support response_format, so we need to handle it via parse call
             parsed_response: ParsedResponse[BaseModel] = await client.responses.parse(
-                text_format=response_format,
+                text_format=text_format,
                 stream=False,
                 **run_options,
             )
@@ -134,9 +137,13 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
         client = await self.ensure_client()
         run_options = await self.prepare_options(messages, chat_options)
         function_call_ids: dict[int, tuple[str, str]] = {}  # output_index: (call_id, name)
+        response_format = run_options.pop("response_format", None)
+        text_config = run_options.pop("text", None)
+        text_format, text_config = self._prepare_text_config(response_format=response_format, text_config=text_config)
+        if text_config:
+            run_options["text"] = text_config
         try:
-            response_format = run_options.pop("response_format", None)
-            if not response_format:
+            if not text_format:
                 response = await client.responses.create(
                     stream=True,
                     **run_options,
@@ -147,9 +154,8 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                     )
                     yield update
                 return
-            # create call does not support response_format, so we need to handle it via stream call
             async with client.responses.stream(
-                text_format=response_format,
+                text_format=text_format,
                 **run_options,
             ) as response:
                 async for chunk in response:
@@ -173,6 +179,71 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
                 inner_exception=ex,
             ) from ex
 
+    def _prepare_text_config(
+        self,
+        *,
+        response_format: Any,
+        text_config: MutableMapping[str, Any] | None,
+    ) -> tuple[type[BaseModel] | None, dict[str, Any] | None]:
+        """Normalize response_format into Responses text configuration and parse target."""
+        prepared_text = dict(text_config) if isinstance(text_config, MutableMapping) else None
+        if text_config is not None and not isinstance(text_config, MutableMapping):
+            raise ServiceInvalidRequestError("text must be a mapping when provided.")
+
+        if response_format is None:
+            return None, prepared_text
+
+        if isinstance(response_format, type) and issubclass(response_format, BaseModel):
+            if prepared_text and "format" in prepared_text:
+                raise ServiceInvalidRequestError("response_format cannot be combined with explicit text.format.")
+            return response_format, prepared_text
+
+        if isinstance(response_format, Mapping):
+            format_config = self._convert_response_format(response_format)
+            if prepared_text is None:
+                prepared_text = {}
+            elif "format" in prepared_text and prepared_text["format"] != format_config:
+                raise ServiceInvalidRequestError("Conflicting response_format definitions detected.")
+            prepared_text["format"] = format_config
+            return None, prepared_text
+
+        raise ServiceInvalidRequestError("response_format must be a Pydantic model or mapping.")
+
+    def _convert_response_format(self, response_format: Mapping[str, Any]) -> dict[str, Any]:
+        """Convert Chat style response_format into Responses text format config."""
+        if "format" in response_format and isinstance(response_format["format"], Mapping):
+            return dict(response_format["format"])
+
+        format_type = response_format.get("type")
+        if format_type == "json_schema":
+            schema_section = response_format.get("json_schema", response_format)
+            if not isinstance(schema_section, Mapping):
+                raise ServiceInvalidRequestError("json_schema response_format must be a mapping.")
+            schema = schema_section.get("schema")
+            if schema is None:
+                raise ServiceInvalidRequestError("json_schema response_format requires a schema.")
+            name = (
+                schema_section.get("name")
+                or schema_section.get("title")
+                or (schema.get("title") if isinstance(schema, Mapping) else None)
+                or "response"
+            )
+            format_config: dict[str, Any] = {
+                "type": "json_schema",
+                "name": name,
+                "schema": schema,
+            }
+            if "strict" in schema_section:
+                format_config["strict"] = schema_section["strict"]
+            if "description" in schema_section and schema_section["description"] is not None:
+                format_config["description"] = schema_section["description"]
+            return format_config
+
+        if format_type in {"json_object", "text"}:
+            return {"type": format_type}
+
+        raise ServiceInvalidRequestError("Unsupported response_format provided for Responses client.")
+
     def get_conversation_id(
         self, response: OpenAIResponse | ParsedResponse[BaseModel], store: bool | None
     ) -> str | None:

python/samples/getting_started/agents/azure_ai/README.md

@@ -20,6 +20,7 @@ This folder contains examples demonstrating different ways to create and use age
 | [`azure_ai_with_file_search.py`](azure_ai_with_file_search.py) | Shows how to use the `HostedFileSearchTool` with Azure AI agents to upload files, create vector stores, and enable agents to search through uploaded documents to answer user questions. |
 | [`azure_ai_with_hosted_mcp.py`](azure_ai_with_hosted_mcp.py) | Shows how to integrate hosted Model Context Protocol (MCP) tools with Azure AI Agent. |
 | [`azure_ai_with_response_format.py`](azure_ai_with_response_format.py) | Shows how to use structured outputs (response format) with Azure AI agents using Pydantic models to enforce specific response schemas. |
+| [`azure_ai_with_runtime_json_schema.py`](azure_ai_with_runtime_json_schema.py) | Shows how to use structured outputs (response format) with Azure AI agents using a JSON schema to enforce specific response schemas. |
 | [`azure_ai_with_search_context_agentic.py`](../../context_providers/azure_ai_search/azure_ai_with_search_context_agentic.py) | Shows how to use AzureAISearchContextProvider with agentic mode. Uses Knowledge Bases for multi-hop reasoning across documents with query planning. Recommended for most scenarios - slightly slower with more token consumption for query planning, but more accurate results. |
 | [`azure_ai_with_search_context_semantic.py`](../../context_providers/azure_ai_search/azure_ai_with_search_context_semantic.py) | Shows how to use AzureAISearchContextProvider with semantic mode. Fast hybrid search with vector + keyword search and semantic ranking for RAG. Best for simple queries where speed is critical. |
 | [`azure_ai_with_sharepoint.py`](azure_ai_with_sharepoint.py) | Shows how to use SharePoint grounding with Azure AI agents to search through SharePoint content and answer user questions with proper citations. Requires a SharePoint connection configured in your Azure AI project. |

python/samples/getting_started/agents/azure_ai/azure_ai_with_runtime_json_schema.py

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+
+from agent_framework.azure import AzureAIClient
+from azure.identity.aio import AzureCliCredential
+
+"""
+Azure AI Agent Response Format Example with Runtime JSON Schema
+
+This sample demonstrates basic usage of AzureAIClient with response format,
+also known as structured outputs.
+"""
+
+
+runtime_schema = {
+    "title": "WeatherDigest",
+    "type": "object",
+    "properties": {
+        "location": {"type": "string"},
+        "conditions": {"type": "string"},
+        "temperature_c": {"type": "number"},
+        "advisory": {"type": "string"},
+    },
+    # OpenAI strict mode requires every property to appear in required.
+    "required": ["location", "conditions", "temperature_c", "advisory"],
+    "additionalProperties": False,
+}
+
+
+async def main() -> None:
+    """Example of using response_format property."""
+
+    # Since no Agent ID is provided, the agent will be automatically created.
+    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
+    # authentication option.
+    async with (
+        AzureCliCredential() as credential,
+        AzureAIClient(async_credential=credential).create_agent(
+            name="ProductMarketerAgent",
+            instructions="Return launch briefs as structured JSON.",
+        ) as agent,
+    ):
+        query = "Draft a launch brief for the Contoso Note app."
+        print(f"User: {query}")
+        result = await agent.run(
+            query,
+            # Specify type to use as response
+            additional_chat_options={
+                "response_format": {
+                    "type": "json_schema",
+                    "json_schema": {
+                        "name": runtime_schema["title"],
+                        "strict": True,
+                        "schema": runtime_schema,
+                    },
+                },
+            },
+        )
+
+        print(result.text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/getting_started/agents/openai/README.md

@@ -31,6 +31,7 @@ This folder contains examples demonstrating different ways to create and use age
 | [`openai_responses_client_with_function_tools.py`](openai_responses_client_with_function_tools.py) | Demonstrates how to use function tools with agents. Shows both agent-level tools (defined when creating the agent) and run-level tools (provided with specific queries). |
 | [`openai_responses_client_with_hosted_mcp.py`](openai_responses_client_with_hosted_mcp.py) | Shows how to integrate OpenAI agents with hosted Model Context Protocol (MCP) servers, including approval workflows and tool management for remote MCP services. |
 | [`openai_responses_client_with_local_mcp.py`](openai_responses_client_with_local_mcp.py) | Shows how to integrate OpenAI agents with local Model Context Protocol (MCP) servers for enhanced functionality and tool integration. |
+| [`openai_responses_client_with_runtime_json_schema.py`](openai_responses_client_with_runtime_json_schema.py) | Shows how to supply a runtime JSON Schema via `additional_chat_options` for structured output without defining a Pydantic model. |
 | [`openai_responses_client_with_structured_output.py`](openai_responses_client_with_structured_output.py) | Demonstrates how to use structured outputs with OpenAI agents to get structured data responses in predefined formats. |
 | [`openai_responses_client_with_thread.py`](openai_responses_client_with_thread.py) | Demonstrates thread management with OpenAI agents, including automatic thread creation for stateless conversations and explicit thread management for maintaining conversation context across multiple interactions. |
 | [`openai_responses_client_with_web_search.py`](openai_responses_client_with_web_search.py) | Shows how to use web search capabilities with OpenAI agents to retrieve and use information from the internet in responses. |

python/samples/getting_started/agents/openai/openai_chat_client_with_runtime_json_schema.py

@@ -73,7 +73,7 @@ async def streaming_example() -> None:
     query = "Give a brief weather digest for Portland."
     print(f"User: {query}")
 
-    chunks = []
+    chunks: list[str] = []
     async for chunk in agent.run_stream(
         query,
         additional_chat_options={

python/samples/getting_started/agents/openai/openai_responses_client_with_runtime_json_schema.py

@@ -0,0 +1,110 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import json
+
+from agent_framework.openai import OpenAIResponsesClient
+
+"""
+OpenAI Chat Client Runtime JSON Schema Example
+
+Demonstrates structured outputs when the schema is only known at runtime.
+Uses additional_chat_options to pass a JSON Schema payload directly to OpenAI
+without defining a Pydantic model up front.
+"""
+
+
+runtime_schema = {
+    "title": "WeatherDigest",
+    "type": "object",
+    "properties": {
+        "location": {"type": "string"},
+        "conditions": {"type": "string"},
+        "temperature_c": {"type": "number"},
+        "advisory": {"type": "string"},
+    },
+    # OpenAI strict mode requires every property to appear in required.
+    "required": ["location", "conditions", "temperature_c", "advisory"],
+    "additionalProperties": False,
+}
+
+
+async def non_streaming_example() -> None:
+    print("=== Non-streaming runtime JSON schema example ===")
+
+    agent = OpenAIResponsesClient().create_agent(
+        name="RuntimeSchemaAgent",
+        instructions="Return only JSON that matches the provided schema. Do not add commentary.",
+    )
+
+    query = "Give a brief weather digest for Seattle."
+    print(f"User: {query}")
+
+    response = await agent.run(
+        query,
+        additional_chat_options={
+            "response_format": {
+                "type": "json_schema",
+                "json_schema": {
+                    "name": runtime_schema["title"],
+                    "strict": True,
+                    "schema": runtime_schema,
+                },
+            },
+        },
+    )
+
+    print("Model output:")
+    print(response.text)
+
+    parsed = json.loads(response.text)
+    print("Parsed dict:")
+    print(parsed)
+
+
+async def streaming_example() -> None:
+    print("=== Streaming runtime JSON schema example ===")
+
+    agent = OpenAIResponsesClient().create_agent(
+        name="RuntimeSchemaAgent",
+        instructions="Return only JSON that matches the provided schema. Do not add commentary.",
+    )
+
+    query = "Give a brief weather digest for Portland."
+    print(f"User: {query}")
+
+    chunks: list[str] = []
+    async for chunk in agent.run_stream(
+        query,
+        additional_chat_options={
+            "response_format": {
+                "type": "json_schema",
+                "json_schema": {
+                    "name": runtime_schema["title"],
+                    "strict": True,
+                    "schema": runtime_schema,
+                },
+            },
+        },
+    ):
+        if chunk.text:
+            chunks.append(chunk.text)
+
+    raw_text = "".join(chunks)
+    print("Model output:")
+    print(raw_text)
+
+    parsed = json.loads(raw_text)
+    print("Parsed dict:")
+    print(parsed)
+
+
+async def main() -> None:
+    print("=== OpenAI Chat Client with runtime JSON Schema ===")
+
+    await non_streaming_example()
+    await streaming_example()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())