Python: feature: support gpt-image-1 (#12621)

This pull request was created in response to the issue
#12500 (comment).
The current AzureTextToImage implementation only works with DALL-E 3.
For gpt-image-1, the response format has changed from a URL to base64
only, which the current code does not support.

Additionally, gpt-image-1 introduces a new image editing feature that
also needs to be supported. Since breaking changes are required, new
methods have been added.

Minimal code that reproduces the problem
```python
service = AzureTextToImage(
        service_id=service_id,
        deployment_name="image-1",
        endpoint=AZURE_OPENAI_IMAGE_ENDPOINT,
        api_key=AZURE_OPENAI_IMAGE_API_KEY,
    )
)

settings = service.get_prompt_execution_settings_class()(service_id="image1")
settings.prompt = "sky"
settings.size = ImageSize(width=1024, height=1024)
settings.quality = "low"
r = await service.generate_image(settings=settings)
```

Example of use with newly added methods
```python
from semantic_kernel.connectors.ai.open_ai import AzureTextToImage
service = AzureTextToImage(
    service_id="image1",
    deployment_name="gpt-image-1",
    endpoint=AZURE_OPENAI_IMAGE_ENDPOINT,
    api_key=AZURE_OPENAI_IMAGE_API_KEY,
    api_version="2025-04-01-preview",
)
settings = service.get_prompt_execution_settings_class()(service_id="image1")
settings.n = 3
images_b64 = await service.generate_images("A cute cat wearing a whimsical striped hat", settings=settings)
```

```python
  from semantic_kernel.connectors.ai.open_ai import AzureTextToImage
  service = AzureTextToImage(
      service_id="image1",
      deployment_name="gpt-image-1",
      endpoint=AZURE_OPENAI_IMAGE_ENDPOINT,
      api_key=AZURE_OPENAI_IMAGE_API_KEY,
      api_version="2025-04-01-preview",
  )
  file_paths = ["./new_images/img_1.png", "./new_images/img_2.png"]
  settings = service.get_prompt_execution_settings_class()(service_id="image1")
  settings.n = 2
  results = await service.edit_image(
      prompt="Make the cat wear a wizard hat",
      image_paths=file_paths,
      settings=settings,
  )
```

## Problems Identified
1. Assumption of URL-based responses. The current implementation assumes
a response format that includes an image url, which is not the case for
gpt-image-1. See:
https://github.com/microsoft/semantic-kernel/blob/8d1b3fd55155bb65e9671383366d2672c4582fb0/python/semantic_kernel/connectors/ai/open_ai/services/open_ai_text_to_image_base.py#L69

Author: ymuichiro

python/semantic_kernel/connectors/ai/open_ai/prompt_execution_settings/open_ai_text_to_image_execution_settings.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import logging
-from typing import Any
+from typing import Any, Literal
 
 from pydantic import Field, model_validator
 
@@ -40,6 +40,10 @@ class OpenAITextToImageExecutionSettings(PromptExecutionSettings):
     size: ImageSize | None = None
     quality: str | None = None
     style: str | None = None
+    output_compression: int | None = None
+    background: Literal["transparent", "opaque", "auto"] | None = None
+    n: int | None = Field(default=1, ge=1, le=10)
+    moderation: Literal["auto", "low"] | None = None
 
     @model_validator(mode="before")
     @classmethod

python/semantic_kernel/connectors/ai/open_ai/services/open_ai_handler.py

@@ -5,6 +5,7 @@
 from typing import Any, Union
 
 from openai import AsyncOpenAI, AsyncStream, BadRequestError, _legacy_response
+from openai._types import NOT_GIVEN, FileTypes, NotGiven
 from openai.lib._parsing._completions import type_to_response_format_param
 from openai.types import Completion, CreateEmbeddingResponse
 from openai.types.audio import Transcription
@@ -122,12 +123,41 @@ async def _send_embedding_request(self, settings: OpenAIEmbeddingPromptExecution
     async def _send_text_to_image_request(self, settings: OpenAITextToImageExecutionSettings) -> ImagesResponse:
         """Send a request to the OpenAI text to image endpoint."""
         try:
-            return await self.client.images.generate(
+            response: ImagesResponse = await self.client.images.generate(
                 **settings.prepare_settings_dict(),
             )
+            self.store_usage(response)
+            return response
         except Exception as ex:
             raise ServiceResponseException(f"Failed to generate image: {ex}") from ex
 
+    async def _send_image_edit_request(
+        self,
+        image: list[FileTypes],
+        settings: OpenAITextToImageExecutionSettings,
+        mask: FileTypes | NotGiven = NOT_GIVEN,
+    ) -> ImagesResponse:
+        """Send a request to the OpenAI image edit endpoint.
+
+        Args:
+            image: List of image files to edit. Accepts file paths or bytes.
+            settings: Image edit execution settings.
+            mask: Optional mask image. Accepts file path or bytes.
+
+        Returns:
+            ImagesResponse: The response from the image edit API.
+        """
+        try:
+            response: ImagesResponse = await self.client.images.edit(
+                image=image,
+                mask=mask,
+                **settings.prepare_settings_dict(),
+            )
+            self.store_usage(response)
+            return response
+        except Exception as ex:
+            raise ServiceResponseException(f"Failed to edit image: {ex}") from ex
+
     async def _send_audio_to_text_request(self, settings: OpenAIAudioToTextExecutionSettings) -> Transcription:
         """Send a request to the OpenAI audio to text endpoint."""
         if not settings.filename:
@@ -187,12 +217,19 @@ def store_usage(
         | Completion
         | AsyncStream[ChatCompletionChunk]
         | AsyncStream[Completion]
-        | CreateEmbeddingResponse,
+        | CreateEmbeddingResponse
+        | ImagesResponse,
     ):
         """Store the usage information from the response."""
-        if not isinstance(response, AsyncStream) and response.usage:
+        if isinstance(response, ImagesResponse) and hasattr(response, "usage") and response.usage:
+            logger.info(f"OpenAI image usage: {response.usage}")
+            self.prompt_tokens += response.usage.input_tokens
+            self.total_tokens += response.usage.total_tokens
+            self.completion_tokens += response.usage.output_tokens
+            return
+        if not isinstance(response, AsyncStream) and not isinstance(response, ImagesResponse) and response.usage:
             logger.info(f"OpenAI usage: {response.usage}")
             self.prompt_tokens += response.usage.prompt_tokens
             self.total_tokens += response.usage.total_tokens
             if hasattr(response.usage, "completion_tokens"):
-                self.completion_tokens += response.usage.completion_tokens
+                self.completion_tokens += response.usage.completion_tokens  # type: ignore

python/semantic_kernel/connectors/ai/open_ai/services/open_ai_text_to_image_base.py

@@ -1,8 +1,10 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from typing import Any
+from pathlib import Path
+from typing import IO, Any
 from warnings import warn
 
+from openai._types import NOT_GIVEN, FileTypes, NotGiven
 from openai.types.images_response import ImagesResponse
 
 from semantic_kernel.connectors.ai.open_ai.prompt_execution_settings.open_ai_text_to_image_execution_settings import (
@@ -38,6 +40,7 @@ async def generate_image(
         Returns:
             bytes | str: Image bytes or image URL.
         """
+        warn("generate_image is deprecated. Use generate_images.", DeprecationWarning, stacklevel=2)
         if not settings:
             settings = OpenAITextToImageExecutionSettings(**kwargs)
         if not isinstance(settings, OpenAITextToImageExecutionSettings):
@@ -70,6 +73,177 @@ async def generate_image(
 
         return response.data[0].url
 
+    async def generate_images(
+        self,
+        prompt: str,
+        settings: PromptExecutionSettings | None = None,
+        **kwargs: Any,
+    ) -> list[str]:
+        """Generate one or more images from text. Returns URLs or base64-encoded images.
+
+        Args:
+            prompt: Description of the image(s) to generate.
+            settings: Execution settings for the prompt.
+            kwargs: Additional arguments, check the openai images.generate documentation for the supported arguments.
+
+        Returns:
+            list[str]: Image URLs or base64-encoded images.
+
+        Example:
+            Generate images and save them as PNG files:
+
+            ```python
+            from semantic_kernel.connectors.ai.open_ai import AzureTextToImage
+            import base64, os
+
+            service = AzureTextToImage(
+                service_id="image1",
+                deployment_name="gpt-image-1",
+                endpoint="https://your-endpoint.cognitiveservices.azure.com",
+                api_key="your-api-key",
+                api_version="2025-04-01-preview",
+            )
+            settings = service.get_prompt_execution_settings_class()(service_id="image1")
+            settings.n = 3
+            images_b64 = await service.generate_images("A cute cat wearing a whimsical striped hat", settings=settings)
+            ```
+        """
+        if not settings:
+            settings = OpenAITextToImageExecutionSettings(**kwargs)
+        if not isinstance(settings, OpenAITextToImageExecutionSettings):
+            settings = OpenAITextToImageExecutionSettings.from_prompt_execution_settings(settings)
+        if prompt:
+            settings.prompt = prompt
+
+        if not settings.prompt:
+            raise ServiceInvalidRequestError("Prompt is required.")
+
+        if not settings.ai_model_id:
+            settings.ai_model_id = self.ai_model_id
+
+        response = await self._send_request(settings)
+
+        assert isinstance(response, ImagesResponse)  # nosec
+        if not response.data or not isinstance(response.data, list) or len(response.data) == 0:
+            raise ServiceResponseException("Failed to generate image.")
+
+        results: list[str] = []
+        for image in response.data:
+            url: str | None = getattr(image, "url", None)
+            b64_json: str | None = getattr(image, "b64_json", None)
+            if url:
+                results.append(url)
+            elif b64_json:
+                results.append(b64_json)
+            else:
+                continue
+
+        if len(results) == 0:
+            raise ServiceResponseException("No valid image data found in response.")
+        return results
+
+    async def edit_image(
+        self,
+        prompt: str,
+        image_paths: list[str] | None = None,
+        image_files: list[IO[bytes]] | None = None,
+        mask_path: str | None = None,
+        mask_file: IO[bytes] | None = None,
+        settings: PromptExecutionSettings | None = None,
+        **kwargs: Any,
+    ) -> list[str]:
+        """Edit images using the OpenAI image edit API.
+
+        Args:
+            prompt: Instructional prompt for image editing.
+            image_paths: List of image file paths to edit.
+            image_files: List of file-like objects (opened in binary mode) to edit.
+            mask_path: Optional mask image file path.
+            mask_file: Optional mask image file-like object (opened in binary mode).
+            settings: Optional execution settings. If not provided, will be constructed from kwargs.
+            kwargs: Additional API parameters.
+
+        Returns:
+            list[str]: List of edited image URLs or base64-encoded strings.
+
+        Example:
+            Edit images from file path and save results:
+
+            ```python
+            from semantic_kernel.connectors.ai.open_ai import AzureTextToImage
+            import base64, os
+
+            service = AzureTextToImage(
+                service_id="image1",
+                deployment_name="gpt-image-1",
+                endpoint="https://your-endpoint.cognitiveservices.azure.com",
+                api_key="your-api-key",
+                api_version="2025-04-01-preview",
+            )
+            file_paths = ["./new_images/img_1.png", "./new_images/img_2.png"]
+            settings = service.get_prompt_execution_settings_class()(service_id="image1")
+            settings.n = 2
+            results = await service.edit_image(
+                prompt="Make the cat wear a wizard hat",
+                image_paths=file_paths,
+                settings=settings,
+            )
+            ```
+
+            Edit images from file object:
+
+            ```python
+            with open("./new_images/img_1.png", "rb") as f:
+                results = await service.edit_image(
+                    prompt="Make the cat wear a wizard hat",
+                    image_files=[f],
+                )
+            ```
+        """
+        if not settings:
+            settings = OpenAITextToImageExecutionSettings(**kwargs)
+        if not isinstance(settings, OpenAITextToImageExecutionSettings):
+            settings = OpenAITextToImageExecutionSettings.from_prompt_execution_settings(settings)
+        settings.prompt = prompt
+
+        if not settings.prompt:
+            raise ServiceInvalidRequestError("Prompt is required.")
+        if (image_paths is None and image_files is None) or (image_paths is not None and image_files is not None):
+            raise ServiceInvalidRequestError("Provide either 'image_paths' or 'image_files', and only one.")
+
+        images: list[FileTypes] = []
+        if image_paths is not None:
+            images = [Path(p) for p in image_paths]
+        elif image_files is not None:
+            images = list(image_files)
+
+        mask: FileTypes | NotGiven = NOT_GIVEN
+        if mask_path is not None:
+            mask = Path(mask_path)
+        elif mask_file is not None:
+            mask = mask_file
+
+        response: ImagesResponse = await self._send_image_edit_request(
+            image=images,
+            mask=mask,
+            settings=settings,
+        )
+
+        if not response or not response.data or not isinstance(response.data, list):
+            raise ServiceResponseException("Failed to edit image.")
+
+        results: list[str] = []
+        for img in response.data:
+            b64_json: str | None = getattr(img, "b64_json", None)
+            url: str | None = getattr(img, "url", None)
+            if b64_json:
+                results.append(b64_json)
+            elif url:
+                results.append(url)
+        if not results:
+            raise ServiceResponseException("No valid image data found in response.")
+        return results
+
     def get_prompt_execution_settings_class(self) -> type[PromptExecutionSettings]:
         """Get the request settings class."""
         return OpenAITextToImageExecutionSettings

python/tests/unit/connectors/ai/open_ai/services/test_azure_text_to_image.py

@@ -81,15 +81,19 @@ def test_azure_text_to_image_init_with_from_dict(azure_openai_unit_test_env) ->
 @patch.object(AsyncImages, "generate", return_value=AsyncMock(spec=ImagesResponse))
 async def test_azure_text_to_image_calls_with_parameters(mock_generate, azure_openai_unit_test_env) -> None:
     mock_generate.return_value.data = [Image(url="abc")]
+    mock_generate.return_value.usage = None
 
     prompt = "A painting of a vase with flowers"
     width = 512
 
-    azure_text_to_image = AzureTextToImage()
+    azure_text_to_image = AzureTextToImage(
+        deployment_name=azure_openai_unit_test_env["AZURE_OPENAI_TEXT_TO_IMAGE_DEPLOYMENT_NAME"]
+    )
     await azure_text_to_image.generate_image(prompt, width=width, height=width)
 
     mock_generate.assert_awaited_once_with(
         prompt=prompt,
         model=azure_openai_unit_test_env["AZURE_OPENAI_TEXT_TO_IMAGE_DEPLOYMENT_NAME"],
         size=f"{width}x{width}",
+        n=1,
     )