From 9c2850f02ea32961d9eb5f0f301f6a0106ffdd3c Mon Sep 17 00:00:00 2001 From: octo-patch <266937838+octo-patch@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:46:06 +0800 Subject: [PATCH] feat: add MiniMax image generation tool --- tests/tools/test_minimax_image.py | 222 +++++++++++++++++++++++ tools/graphics/minimax_image.py | 290 ++++++++++++++++++++++++++++++ 2 files changed, 512 insertions(+) create mode 100644 tests/tools/test_minimax_image.py create mode 100644 tools/graphics/minimax_image.py diff --git a/tests/tools/test_minimax_image.py b/tests/tools/test_minimax_image.py new file mode 100644 index 00000000..ed49af19 --- /dev/null +++ b/tests/tools/test_minimax_image.py @@ -0,0 +1,222 @@ +"""Contract tests for the MiniMax image generation tool.""" + +from __future__ import annotations + +import base64 + +import pytest +import requests + +from tools.base_tool import ToolStatus +from tools.graphics import minimax_image +from tools.graphics.minimax_image import MiniMaxImage +from tools.tool_registry import ToolRegistry + + +class FakeResponse: + def __init__(self, *, json_data=None, content: bytes = b"") -> None: + self._json_data = json_data + self.content = content + + def raise_for_status(self) -> None: + return None + + def json(self): + return self._json_data + + +@pytest.fixture(autouse=True) +def clear_minimax_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("MINIMAX_API_KEY", raising=False) + monkeypatch.delenv("MINIMAX_REGION", raising=False) + monkeypatch.delenv("MINIMAX_BASE_URL", raising=False) + + +def test_registry_registers_minimax_image_tool() -> None: + registry = ToolRegistry() + assert registry.register_module(minimax_image) == ["minimax_image"] + + tool = registry.get("minimax_image") + assert tool is not None + assert tool.provider == "minimax" + assert tool.capability == "image_generation" + assert tool.input_schema["properties"]["model"]["enum"] == [ + "image-01", + "image-01-live", + ] + + +def test_status_requires_api_key(monkeypatch: pytest.MonkeyPatch) -> None: + tool = MiniMaxImage() + assert tool.get_status() == ToolStatus.UNAVAILABLE + + monkeypatch.setenv("MINIMAX_API_KEY", "test-key") + assert tool.get_status() == ToolStatus.AVAILABLE + + +@pytest.mark.parametrize( + ("region", "expected_base_url"), + [ + ("global", "https://api.minimax.io"), + ("global_en", "https://api.minimax.io"), + ("cn", "https://api.minimaxi.com"), + ("cn_zh", "https://api.minimaxi.com"), + ], +) +def test_region_routes_to_official_endpoint( + monkeypatch: pytest.MonkeyPatch, region: str, expected_base_url: str +) -> None: + monkeypatch.setenv("MINIMAX_REGION", region) + assert MiniMaxImage()._base_url() == expected_base_url + + +def test_url_response_downloads_all_images( + monkeypatch: pytest.MonkeyPatch, tmp_path +) -> None: + monkeypatch.setenv("MINIMAX_API_KEY", "test-key") + monkeypatch.setenv("MINIMAX_REGION", "cn") + captured = {} + + def fake_post(url, *, headers, json, timeout): + captured.update(url=url, headers=headers, payload=json, timeout=timeout) + return FakeResponse( + json_data={ + "id": "request-1", + "data": { + "image_urls": [ + "https://example.test/one.png", + "https://example.test/two.png", + ] + }, + "metadata": {"success_count": 2, "failed_count": 0}, + "base_resp": {"status_code": 0, "status_msg": "success"}, + } + ) + + def fake_get(url, *, timeout): + assert timeout == 120 + return FakeResponse(content=url.rsplit("/", 1)[-1].encode()) + + monkeypatch.setattr(requests, "post", fake_post) + monkeypatch.setattr(requests, "get", fake_get) + + output_path = tmp_path / "image.png" + result = MiniMaxImage().execute( + { + "prompt": "A lighthouse at dusk", + "model": "image-01-live", + "subject_reference": [ + {"type": "character", "image_file": "https://example.test/ref.png"} + ], + "aspect_ratio": "16:9", + "response_format": "url", + "seed": 42, + "n": 2, + "prompt_optimizer": True, + "output_path": str(output_path), + } + ) + + assert result.success + assert captured["url"] == "https://api.minimaxi.com/v1/image_generation" + assert captured["headers"]["Authorization"] == "Bearer test-key" + assert captured["payload"] == { + "model": "image-01-live", + "prompt": "A lighthouse at dusk", + "response_format": "url", + "n": 2, + "prompt_optimizer": True, + "subject_reference": [ + {"type": "character", "image_file": "https://example.test/ref.png"} + ], + "aspect_ratio": "16:9", + "seed": 42, + } + assert result.artifacts == [ + str(tmp_path / "image_1.png"), + str(tmp_path / "image_2.png"), + ] + assert (tmp_path / "image_1.png").read_bytes() == b"one.png" + assert (tmp_path / "image_2.png").read_bytes() == b"two.png" + assert result.data["metadata"] == {"success_count": 2, "failed_count": 0} + + +def test_base64_response_writes_inline_images( + monkeypatch: pytest.MonkeyPatch, tmp_path +) -> None: + monkeypatch.setenv("MINIMAX_API_KEY", "test-key") + image_bytes = b"inline image" + + monkeypatch.setattr( + requests, + "post", + lambda *args, **kwargs: FakeResponse( + json_data={ + "data": { + "image_base64": [ + "data:image/png;base64," + + base64.b64encode(image_bytes).decode("ascii") + ] + }, + "metadata": {"success_count": "1", "failed_count": "0"}, + "base_resp": {"status_code": 0}, + } + ), + ) + monkeypatch.setattr( + requests, + "get", + lambda *args, **kwargs: pytest.fail("base64 output must not be downloaded"), + ) + + output_path = tmp_path / "inline.png" + result = MiniMaxImage().execute( + { + "prompt": "A paper-cut forest", + "response_format": "base64", + "output_path": str(output_path), + } + ) + + assert result.success + assert output_path.read_bytes() == image_bytes + assert result.data["response_format"] == "base64" + + +def test_base_response_error_is_returned(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("MINIMAX_API_KEY", "test-key") + monkeypatch.setattr( + requests, + "post", + lambda *args, **kwargs: FakeResponse( + json_data={ + "base_resp": { + "status_code": 1008, + "status_msg": "insufficient balance", + } + } + ), + ) + + result = MiniMaxImage().execute({"prompt": "A mountain cabin"}) + + assert not result.success + assert result.error == "MiniMax API error 1008: insufficient balance" + + +def test_width_and_height_must_be_provided_together( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("MINIMAX_API_KEY", "test-key") + monkeypatch.setattr( + requests, + "post", + lambda *args, **kwargs: pytest.fail("invalid inputs must not call the API"), + ) + + result = MiniMaxImage().execute( + {"prompt": "A mountain cabin", "width": 1024} + ) + + assert not result.success + assert "width and height must be set together" in (result.error or "") diff --git a/tools/graphics/minimax_image.py b/tools/graphics/minimax_image.py new file mode 100644 index 00000000..47bac617 --- /dev/null +++ b/tools/graphics/minimax_image.py @@ -0,0 +1,290 @@ +"""MiniMax image generation through the first-party API.""" + +from __future__ import annotations + +import base64 +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolTier, +) + + +MODELS = ["image-01", "image-01-live"] +DEFAULT_MODEL = "image-01" +DEFAULT_REGION = "global" +REGION_BASE_URLS = { + "global": "https://api.minimax.io", + "global_en": "https://api.minimax.io", + "cn": "https://api.minimaxi.com", + "cn_zh": "https://api.minimaxi.com", +} + + +class MiniMaxImage(BaseTool): + name = "minimax_image" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "image_generation" + provider = "minimax" + stability = ToolStability.BETA + execution_mode = ExecutionMode.SYNC + determinism = Determinism.SEEDED + runtime = ToolRuntime.API + + dependencies = ["env:MINIMAX_API_KEY"] + install_instructions = ( + "Set MINIMAX_API_KEY to your MiniMax API key. " + "Optionally set MINIMAX_REGION to global or cn." + ) + agent_skills = ["flux-best-practices"] + + capabilities = ["generate_image", "text_to_image"] + supports = { + "multiple_outputs": True, + "aspect_ratio": True, + "custom_dimensions": True, + "seed": True, + "subject_reference": True, + "url_response": True, + "base64_response": True, + } + best_for = [ + "first-party MiniMax image generation", + "seeded multi-image generation", + "global and mainland China API routing", + ] + not_good_for = ["offline generation"] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": {"type": "string", "maxLength": 1500}, + "model": { + "type": "string", + "enum": MODELS, + "default": DEFAULT_MODEL, + }, + "subject_reference": { + "type": "array", + "items": { + "type": "object", + "required": ["type", "image_file"], + "properties": { + "type": {"type": "string", "enum": ["character"]}, + "image_file": {"type": "string"}, + }, + }, + }, + "aspect_ratio": { + "type": "string", + "enum": ["1:1", "16:9", "4:3", "3:2", "2:3", "3:4", "9:16", "21:9"], + "default": "1:1", + }, + "width": {"type": "integer", "minimum": 512, "maximum": 2048, "multipleOf": 8}, + "height": {"type": "integer", "minimum": 512, "maximum": 2048, "multipleOf": 8}, + "response_format": { + "type": "string", + "enum": ["url", "base64"], + "default": "url", + }, + "seed": {"type": "integer"}, + "n": {"type": "integer", "minimum": 1, "maximum": 9, "default": 1}, + "prompt_optimizer": {"type": "boolean", "default": False}, + "output_path": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=100, network_required=True + ) + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) + idempotency_key_fields = [ + "prompt", + "model", + "subject_reference", + "aspect_ratio", + "width", + "height", + "response_format", + "seed", + "n", + "prompt_optimizer", + ] + side_effects = [ + "writes image files to output_path", + "calls the MiniMax image generation API", + ] + user_visible_verification = [ + "Inspect generated images for prompt adherence and visual quality" + ] + + @staticmethod + def _region() -> str: + region = os.environ.get("MINIMAX_REGION", DEFAULT_REGION).strip().lower() + return region if region in REGION_BASE_URLS else DEFAULT_REGION + + def _base_url(self) -> str: + override = os.environ.get("MINIMAX_BASE_URL") + if override: + return override.rstrip("/") + return REGION_BASE_URLS[self._region()] + + @staticmethod + def _base_resp_error(data: dict[str, Any]) -> str | None: + base_resp = data.get("base_resp") or {} + status_code = base_resp.get("status_code") + if status_code in (None, 0): + return None + status_msg = base_resp.get("status_msg") or "unknown error" + return f"MiniMax API error {status_code}: {status_msg}" + + @staticmethod + def _output_paths(output_path: str | None, count: int) -> list[Path]: + path = Path(output_path or "minimax_image.png") + if not path.suffix: + path = path.with_suffix(".png") + if count == 1: + return [path] + return [ + path.with_name(f"{path.stem}_{index}{path.suffix}") + for index in range(1, count + 1) + ] + + @staticmethod + def _build_payload(inputs: dict[str, Any]) -> dict[str, Any]: + model = inputs.get("model", DEFAULT_MODEL) + if model not in MODELS: + raise ValueError(f"Unsupported MiniMax image model '{model}'.") + + prompt = inputs.get("prompt") + if not isinstance(prompt, str) or not prompt: + raise ValueError("MiniMax image generation requires 'prompt'.") + if len(prompt) > 1500: + raise ValueError("MiniMax image prompt must not exceed 1500 characters.") + + width = inputs.get("width") + height = inputs.get("height") + if (width is None) != (height is None): + raise ValueError("MiniMax image width and height must be set together.") + + payload: dict[str, Any] = { + "model": model, + "prompt": prompt, + "response_format": inputs.get("response_format", "url"), + "n": inputs.get("n", 1), + "prompt_optimizer": inputs.get("prompt_optimizer", False), + } + for field in ( + "subject_reference", + "aspect_ratio", + "width", + "height", + "seed", + ): + if inputs.get(field) is not None: + payload[field] = inputs[field] + return payload + + @staticmethod + def _decode_base64_image(value: str) -> bytes: + encoded = value.split(",", 1)[1] if value.startswith("data:") else value + return base64.b64decode(encoded) + + @staticmethod + def _safe_error(exc: Exception, api_key: str) -> str: + return str(exc).replace(api_key, "[redacted]") if api_key else str(exc) + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = os.environ.get("MINIMAX_API_KEY", "") + if not api_key: + return ToolResult( + success=False, + error="MINIMAX_API_KEY not set. " + self.install_instructions, + ) + + import requests + + start = time.time() + try: + payload = self._build_payload(inputs) + response = requests.post( + f"{self._base_url()}/v1/image_generation", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json=payload, + timeout=180, + ) + response.raise_for_status() + data = response.json() + + base_error = self._base_resp_error(data) + if base_error: + return ToolResult(success=False, error=base_error) + + response_format = payload["response_format"] + data_object = data.get("data") or {} + image_values = data_object.get( + "image_base64" if response_format == "base64" else "image_urls" + ) or [] + if not image_values: + return ToolResult( + success=False, + error=f"MiniMax returned no {response_format} image outputs.", + ) + + output_paths = self._output_paths( + inputs.get("output_path"), len(image_values) + ) + for path, value in zip(output_paths, image_values): + path.parent.mkdir(parents=True, exist_ok=True) + if response_format == "base64": + path.write_bytes(self._decode_base64_image(value)) + else: + download = requests.get(value, timeout=120) + download.raise_for_status() + path.write_bytes(download.content) + except Exception as exc: + return ToolResult( + success=False, + error=( + "MiniMax image generation failed: " + f"{self._safe_error(exc, api_key)}" + ), + ) + + outputs = [str(path) for path in output_paths] + return ToolResult( + success=True, + data={ + "provider": "minimax", + "model": payload["model"], + "prompt": payload["prompt"], + "region": self._region(), + "response_format": payload["response_format"], + "output": outputs[0], + "outputs": outputs, + "images_generated": len(outputs), + "metadata": data.get("metadata") or {}, + "request_id": data.get("id"), + }, + artifacts=outputs, + duration_seconds=round(time.time() - start, 2), + model=payload["model"], + )