summaryrefslogtreecommitdiff
path: root/src/mcp_sd15/images.py
diff options
context:
space:
mode:
authortanaka takashi <isazaka@gmail.com>2026-09-12 18:04:56 +0900
committertanaka takashi <isazaka@gmail.com>2026-09-12 18:04:56 +0900
commitcba15a62829a0fe57a03c8031f981e8f790e2efc (patch)
tree3bc0ef1a243f356f55c2fa97f63c3bfd5c77e8fc /src/mcp_sd15/images.py
feat: Stable Diffusion 1.x MCPサーバを実装
仕様書・設計書に基づき、ローカルCUDA GPU上のStable Diffusion 1.x互換モデルをMCP経由で利用するサーバを新規実装。 実装内容: - generate_image、generate_guided_image、inpaint_imageの3ツールをMCP SDK v2で公開 - TOMLからcheckpoint、VAE、複数LoRAと重み、ControlNet、生成設定を読み込み - モデル種別ごとの固定ディレクトリ解決と起動時の設定・ファイル存在検証を追加 - Depth/MiDaS、OpenPose body/full、ControlNet Inpaintの前処理とパイプライン切替を実装 - FP16/CUDA、1枚固定生成、ControlNetオンデマンドロード、単一GPU直列実行を実装 - Base64 PNG/JPEG入力の形式・サイズ検証と任意のdata URI入力に対応 - seed、サイズ、モード、時刻、処理時間をPNG metadataへ記録 - 任意のローカルPNG保存、生成エラー分類、CUDA OOM後のキャッシュ回復を追加 - --configと--versionを備えたstdio CLIエントリポイントを追加 開発・運用: - Python 3.11およびuvによる依存・lock管理を構成 - config.example.tomlとセットアップ・運用手順をREADMEへ追加 - Ruff、Pyright、pytestの設定を追加 - 設定、パス解決、schema、画像検証、サービスルーティング、排他制御、MCP画像返却をテスト 検証結果: - pytest: 30 passed - Ruff lint/format: passed - Pyright: 0 errors, 0 warnings - CLI versionおよびMCPクライアント経由のPNG返却を確認 - 実GPU smoke testはCUDA GPUとモデルファイルがないため対象外
Diffstat (limited to 'src/mcp_sd15/images.py')
-rw-r--r--src/mcp_sd15/images.py62
1 files changed, 62 insertions, 0 deletions
diff --git a/src/mcp_sd15/images.py b/src/mcp_sd15/images.py
new file mode 100644
index 0000000..4b9cf3c
--- /dev/null
+++ b/src/mcp_sd15/images.py
@@ -0,0 +1,62 @@
+"""Safe conversion between MCP Base64 inputs and Pillow images."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import io
+import json
+from typing import Any
+
+from PIL import Image, PngImagePlugin, UnidentifiedImageError
+
+from .errors import InvalidImageError
+
+MAX_ENCODED_LENGTH = 36_000_000
+MAX_PIXELS = 40_000_000
+SUPPORTED_FORMATS = frozenset({"PNG", "JPEG"})
+
+
+def decode_image(value: str, *, mask: bool = False) -> Image.Image:
+ """Decode a bounded PNG/JPEG Base64 value or data URI and fully validate it."""
+ if not isinstance(value, str) or len(value) > MAX_ENCODED_LENGTH:
+ raise InvalidImageError("image Base64 input is missing or too large")
+ payload = value.strip()
+ if payload.startswith("data:"):
+ try:
+ header, payload = payload.split(",", 1)
+ except ValueError as exc:
+ raise InvalidImageError("malformed image data URI") from exc
+ if header.lower() not in {
+ "data:image/png;base64",
+ "data:image/jpeg;base64",
+ "data:image/jpg;base64",
+ }:
+ raise InvalidImageError("data URI must contain a Base64 PNG or JPEG")
+ try:
+ raw = base64.b64decode(payload, validate=True)
+ except (binascii.Error, ValueError) as exc:
+ raise InvalidImageError("image is not valid Base64") from exc
+ try:
+ with Image.open(io.BytesIO(raw)) as opened:
+ if opened.format not in SUPPORTED_FORMATS:
+ raise InvalidImageError("only PNG and JPEG images are supported")
+ if opened.width * opened.height > MAX_PIXELS:
+ raise InvalidImageError("image pixel count exceeds the safety limit")
+ opened.load()
+ return opened.convert("L" if mask else "RGB")
+ except InvalidImageError:
+ raise
+ except (OSError, UnidentifiedImageError, Image.DecompressionBombError) as exc:
+ raise InvalidImageError("input is not a valid PNG or JPEG image") from exc
+
+
+def encode_png(image: Image.Image, metadata: dict[str, Any]) -> bytes:
+ """Serialize one RGB image as PNG with stable textual generation metadata."""
+ info = PngImagePlugin.PngInfo()
+ for key, value in metadata.items():
+ text = json.dumps(value) if isinstance(value, (dict, list)) else str(value)
+ info.add_text(str(key), text)
+ output = io.BytesIO()
+ image.convert("RGB").save(output, format="PNG", pnginfo=info)
+ return output.getvalue()