Add MiniMax image generation support

sdk/python/agentfield/media_providers.py

@@ -745,7 +745,7 @@ def name(self) -> str:
 
     @property
     def supported_modalities(self) -> List[str]:
-        return ["video", "music", "audio"]
+        return ["video", "music", "audio", "image"]
 
     @property
     def video_model_metadata(self) -> Dict[str, Dict[str, Any]]:
@@ -758,9 +758,126 @@ async def generate_image(
         model: Optional[str] = None,
         size: str = "1024x1024",
         quality: str = "standard",
+        subject_reference: Optional[Any] = None,
+        aspect_ratio: Optional[str] = None,
+        width: Optional[int] = None,
+        height: Optional[int] = None,
+        response_format: str = "url",
+        seed: Optional[int] = None,
+        n: Optional[int] = None,
+        prompt_optimizer: Optional[bool] = None,
         **kwargs,
     ) -> MultimodalResponse:
-        raise NotImplementedError("minimax does not support image generation")
+        """Generate images via the MiniMax image_generation endpoint.
+
+        URL responses remain available for 24 hours. ``b64_json`` is accepted
+        as the unified SDK alias for MiniMax's ``base64`` response format.
+        """
+        import os
+
+        import aiohttp
+
+        api_key = self._api_key or os.environ.get("MINIMAX_API_KEY")
+        if not api_key:
+            raise ValueError(
+                "MiniMax API key required. Set MINIMAX_API_KEY or pass api_key "
+                "to MiniMaxProvider."
+            )
+
+        send_model = self._strip_prefix(
+            model if model is not None else "image-01"
+        ).strip()
+        if not send_model:
+            raise ValueError("MiniMax image generation requires a model")
+
+        normalized_format = (
+            "base64" if response_format == "b64_json" else response_format
+        )
+        if normalized_format not in {"url", "base64"}:
+            raise ValueError("MiniMax image response_format must be url or base64")
+
+        body: Dict[str, Any] = {
+            "model": send_model,
+            "prompt": prompt,
+            "response_format": normalized_format,
+        }
+        optional_fields = {
+            "subject_reference": subject_reference,
+            "aspect_ratio": aspect_ratio,
+            "width": width,
+            "height": height,
+            "seed": seed,
+            "n": n,
+            "prompt_optimizer": prompt_optimizer,
+        }
+        body.update(
+            {key: value for key, value in optional_fields.items() if value is not None}
+        )
+
+        if aspect_ratio is None and width is None and height is None and size:
+            size_match = re.fullmatch(r"\s*(\d+)\s*[xX]\s*(\d+)\s*", size)
+            if size_match:
+                body["width"] = int(size_match.group(1))
+                body["height"] = int(size_match.group(2))
+
+        headers = {
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        }
+        timeout = aiohttp.ClientTimeout(total=120.0)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.post(
+                f"{self._base_url}/image_generation",
+                headers=headers,
+                json=body,
+            ) as response:
+                if response.status >= 400:
+                    detail = await response.text()
+                    raise RuntimeError(
+                        f"MiniMax image generation failed ({response.status}): "
+                        f"{detail[:500]}"
+                    )
+                data = await response.json()
+
+        if not isinstance(data, dict):
+            raise RuntimeError("MiniMax image generation returned an invalid response")
+        base_resp = data.get("base_resp") or {}
+        if not isinstance(base_resp, dict):
+            raise RuntimeError("MiniMax image generation returned an invalid response")
+        status_code = base_resp.get("status_code")
+        if status_code not in (None, 0):
+            status_msg = base_resp.get("status_msg") or "unknown error"
+            raise RuntimeError(
+                f"MiniMax image generation failed ({status_code}): {status_msg}"
+            )
+
+        response_data = data.get("data") or {}
+        if not isinstance(response_data, dict):
+            raise RuntimeError("MiniMax image generation returned an invalid response")
+        response_key = "image_base64" if normalized_format == "base64" else "image_urls"
+        image_values = response_data.get(response_key)
+        if not isinstance(image_values, list):
+            raise RuntimeError(f"MiniMax image generation returned no {response_key}")
+
+        images: List[ImageOutput] = []
+        for value in image_values:
+            if not isinstance(value, str) or not value:
+                continue
+            if normalized_format == "base64":
+                encoded = value.split(",", 1)[1] if value.startswith("data:") else value
+                images.append(ImageOutput(b64_json=encoded))
+            else:
+                images.append(ImageOutput(url=value))
+        if not images:
+            raise RuntimeError("MiniMax image generation returned no images")
+
+        return MultimodalResponse(
+            text=prompt,
+            audio=None,
+            images=images,
+            files=[],
+            raw_response=data,
+        )
 
     async def generate_audio(
         self,

sdk/python/tests/test_minimax_image.py

@@ -0,0 +1,191 @@
+import base64
+
+import pytest
+
+from agentfield.media_providers import (
+    MINIMAX_CN_BASE_URL,
+    MINIMAX_GLOBAL_BASE_URL,
+    MiniMaxProvider,
+)
+from agentfield.media_router import MediaRouter
+
+
+class FakeResponse:
+    def __init__(self, payload, status=200):
+        self.payload = payload
+        self.status = status
+
+    async def json(self):
+        return self.payload
+
+    async def text(self):
+        return str(self.payload)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+
+class CaptureSession:
+    def __init__(self, response):
+        self.response = response
+        self.calls = []
+
+    def post(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        return self.response
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+
+@pytest.mark.asyncio
+async def test_minimax_image_url_output_uses_global_endpoint_and_request_fields(
+    monkeypatch,
+):
+    payload = {
+        "base_resp": {"status_code": 0},
+        "data": {
+            "image_urls": [
+                "https://example.test/one.png",
+                "https://example.test/two.png",
+            ]
+        },
+        "metadata": {"success_count": 2, "failed_count": 0},
+    }
+    session = CaptureSession(FakeResponse(payload))
+    monkeypatch.setattr("aiohttp.ClientSession", lambda **kwargs: session)
+    provider = MiniMaxProvider(api_key="unit-value")
+
+    result = await provider.generate_image(
+        prompt="A luminous city at dusk",
+        model="minimax/image-01-live",
+        subject_reference=["reference-image"],
+        aspect_ratio="16:9",
+        response_format="url",
+        seed=17,
+        n=2,
+        prompt_optimizer=False,
+    )
+
+    assert session.calls[0][0] == f"{MINIMAX_GLOBAL_BASE_URL}/image_generation"
+    assert session.calls[0][1]["json"] == {
+        "model": "image-01-live",
+        "prompt": "A luminous city at dusk",
+        "response_format": "url",
+        "subject_reference": ["reference-image"],
+        "aspect_ratio": "16:9",
+        "seed": 17,
+        "n": 2,
+        "prompt_optimizer": False,
+    }
+    assert session.calls[0][1]["headers"]["Authorization"] == "Bearer unit-value"
+    assert [image.url for image in result.images] == payload["data"]["image_urls"]
+    assert result.raw_response["metadata"]["success_count"] == 2
+
+
+@pytest.mark.asyncio
+async def test_minimax_image_base64_output_uses_cn_endpoint_and_default_model(
+    monkeypatch,
+):
+    image_bytes = b"generated-image"
+    encoded = base64.b64encode(image_bytes).decode("ascii")
+    session = CaptureSession(
+        FakeResponse(
+            {
+                "base_resp": {"status_code": 0},
+                "data": {"image_base64": [encoded]},
+            }
+        )
+    )
+    monkeypatch.setattr("aiohttp.ClientSession", lambda **kwargs: session)
+    provider = MiniMaxProvider(api_key="unit-value", base_url=MINIMAX_CN_BASE_URL)
+
+    result = await provider.generate_image(
+        prompt="A geometric landscape",
+        size="1280x720",
+        response_format="b64_json",
+    )
+
+    assert session.calls[0][0] == f"{MINIMAX_CN_BASE_URL}/image_generation"
+    assert session.calls[0][1]["json"] == {
+        "model": "image-01",
+        "prompt": "A geometric landscape",
+        "response_format": "base64",
+        "width": 1280,
+        "height": 720,
+    }
+    assert result.images[0].url is None
+    assert result.images[0].get_bytes() == image_bytes
+
+
+@pytest.mark.asyncio
+async def test_minimax_image_validates_credentials_format_and_api_errors(monkeypatch):
+    monkeypatch.delenv("MINIMAX_API_KEY", raising=False)
+    provider = MiniMaxProvider()
+    with pytest.raises(ValueError, match="API key required"):
+        await provider.generate_image("An image")
+
+    provider = MiniMaxProvider(api_key="unit-value")
+    for model in ("", " ", "minimax/   "):
+        with pytest.raises(ValueError, match="requires a model"):
+            await provider.generate_image("An image", model=model)
+    with pytest.raises(ValueError, match="response_format"):
+        await provider.generate_image("An image", response_format="binary")
+
+    session = CaptureSession(
+        FakeResponse(
+            {
+                "base_resp": {
+                    "status_code": 1004,
+                    "status_msg": "generation rejected",
+                }
+            }
+        )
+    )
+    monkeypatch.setattr("aiohttp.ClientSession", lambda **kwargs: session)
+    with pytest.raises(RuntimeError, match="generation rejected"):
+        await provider.generate_image("An image")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("payload", "error"),
+    [
+        ("not a dictionary", "invalid response"),
+        ({"base_resp": "invalid"}, "invalid response"),
+        ({"data": "invalid"}, "invalid response"),
+        ({"data": {}}, "no image_urls"),
+        ({"data": {"image_urls": []}}, "no images"),
+    ],
+)
+async def test_minimax_image_rejects_invalid_responses(monkeypatch, payload, error):
+    session = CaptureSession(FakeResponse(payload))
+    monkeypatch.setattr("aiohttp.ClientSession", lambda **kwargs: session)
+    provider = MiniMaxProvider(api_key="unit-value")
+
+    with pytest.raises(RuntimeError, match=error):
+        await provider.generate_image("An image")
+
+
+@pytest.mark.asyncio
+async def test_minimax_image_checks_http_status(monkeypatch):
+    session = CaptureSession(FakeResponse("upstream unavailable", status=503))
+    monkeypatch.setattr("aiohttp.ClientSession", lambda **kwargs: session)
+    provider = MiniMaxProvider(api_key="unit-value")
+
+    with pytest.raises(RuntimeError, match=r"failed \(503\): upstream unavailable"):
+        await provider.generate_image("An image")
+
+
+def test_minimax_image_models_route_to_provider():
+    provider = MiniMaxProvider(api_key="unit-value")
+    router = MediaRouter()
+    router.register("minimax/", provider)
+
+    assert router.resolve("minimax/image-01", "image") is provider

sdk/python/tests/test_minimax_video.py

@@ -231,8 +231,6 @@ async def test_minimax_video_validates_credentials_and_duration(monkeypatch):
             model="minimax/video-model",
             duration=6.5,
         )
-    with pytest.raises(NotImplementedError, match="image generation"):
-        await provider.generate_image("An image")
 
 
 @pytest.mark.asyncio
@@ -611,4 +609,4 @@ def test_minimax_provider_configuration_and_registry():
         base_url=MINIMAX_GLOBAL_BASE_URL,
     )
     assert isinstance(provider, MiniMaxProvider)
-    assert provider.supported_modalities == ["video", "music", "audio"]
+    assert provider.supported_modalities == ["video", "music", "audio", "image"]