diff options
| author | tanaka takashi <isazaka@gmail.com> | 2026-09-12 18:04:56 +0900 |
|---|---|---|
| committer | tanaka takashi <isazaka@gmail.com> | 2026-09-12 18:04:56 +0900 |
| commit | cba15a62829a0fe57a03c8031f981e8f790e2efc (patch) | |
| tree | 3bc0ef1a243f356f55c2fa97f63c3bfd5c77e8fc /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.py | 62 |
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() |
