From cba15a62829a0fe57a03c8031f981e8f790e2efc Mon Sep 17 00:00:00 2001 From: tanaka takashi Date: Sat, 12 Sep 2026 18:04:56 +0900 Subject: feat: Stable Diffusion 1.x MCPサーバを実装 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 仕様書・設計書に基づき、ローカル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とモデルファイルがないため対象外 --- src/mcp_sd15/model_manager.py | 142 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 src/mcp_sd15/model_manager.py (limited to 'src/mcp_sd15/model_manager.py') diff --git a/src/mcp_sd15/model_manager.py b/src/mcp_sd15/model_manager.py new file mode 100644 index 0000000..c2bd166 --- /dev/null +++ b/src/mcp_sd15/model_manager.py @@ -0,0 +1,142 @@ +"""Diffusers-backed generation and on-demand ControlNet management.""" + +from __future__ import annotations + +import gc +from typing import Any + +import numpy as np +from PIL import Image + +from .config import AppConfig +from .errors import GenerationError +from .pipeline_factory import PipelineFactory +from .preprocessors.depth import DepthPreprocessor +from .preprocessors.openpose import OpenPosePreprocessor +from .schema import GuidedImageRequest, InpaintImageRequest, TextToImageRequest + + +class ModelManager: + """Own all model state; construction validates model compatibility at startup.""" + + def __init__(self, config: AppConfig) -> None: + self.config = config + self.factory = PipelineFactory(config) + self.base = self.factory.create_base() + self._control_mode: str | None = None + self._controlnet: Any | None = None + self._control_pipeline: Any | None = None + self._depth = DepthPreprocessor() + self._openpose = OpenPosePreprocessor() + + def text_to_image(self, request: TextToImageRequest, seed: int) -> Image.Image: + width, height = request.size.dimensions + result = self.base( + prompt=request.positive_prompt, + negative_prompt=request.negative_prompt, + width=width, + height=height, + num_inference_steps=self.config.generation.steps, + guidance_scale=self.config.generation.cfg_scale, + num_images_per_prompt=1, + generator=self._generator(seed), + ) + return self._one_image(result) + + def guided_image( + self, request: GuidedImageRequest, image: Image.Image, seed: int + ) -> Image.Image: + width, height = request.size.dimensions + image = image.resize((width, height), Image.Resampling.LANCZOS) + if request.control_mode == "depth": + control_image = self._depth(image) + else: + control_image = self._openpose(image, full=request.control_mode == "openpose_full") + pipeline = self._pipeline_for(str(request.control_mode), inpaint=False) + result = pipeline( + prompt=request.positive_prompt, + negative_prompt=request.negative_prompt, + image=control_image, + width=width, + height=height, + num_inference_steps=self.config.generation.steps, + guidance_scale=self.config.generation.cfg_scale, + num_images_per_prompt=1, + generator=self._generator(seed), + ) + return self._one_image(result) + + def inpaint_image( + self, + request: InpaintImageRequest, + image: Image.Image, + mask: Image.Image, + seed: int, + ) -> Image.Image: + width, height = request.size.dimensions + image = image.resize((width, height), Image.Resampling.LANCZOS) + mask = mask.resize((width, height), Image.Resampling.NEAREST) + control_image = self._inpaint_condition(image, mask) + pipeline = self._pipeline_for("inpaint", inpaint=True) + result = pipeline( + prompt=request.positive_prompt, + negative_prompt=request.negative_prompt, + image=image, + mask_image=mask, + control_image=control_image, + width=width, + height=height, + num_inference_steps=self.config.generation.steps, + guidance_scale=self.config.generation.cfg_scale, + num_images_per_prompt=1, + generator=self._generator(seed), + ) + return self._one_image(result) + + def _pipeline_for(self, mode: str, *, inpaint: bool) -> Any: + model_mode = "openpose" if mode == "openpose_full" else mode + if self._control_mode == model_mode and self._control_pipeline is not None: + return self._control_pipeline + self._release_controlnet() + self._controlnet = self.factory.create_controlnet(self.config.controlnet_path(model_mode)) + self._control_pipeline = ( + self.factory.create_inpaint_pipeline(self.base, self._controlnet) + if inpaint + else self.factory.create_guided_pipeline(self.base, self._controlnet) + ) + self._control_mode = model_mode + return self._control_pipeline + + def _release_controlnet(self) -> None: + self._control_pipeline = None + self._controlnet = None + self._control_mode = None + gc.collect() + try: + import torch + + torch.cuda.empty_cache() + except ImportError: + pass + + @staticmethod + def _generator(seed: int) -> Any: + import torch + + return torch.Generator(device="cuda").manual_seed(seed) + + @staticmethod + def _inpaint_condition(image: Image.Image, mask: Image.Image) -> Any: + import torch + + image_array = np.asarray(image.convert("RGB"), dtype=np.float32) / 255.0 + mask_array = np.asarray(mask.convert("L"), dtype=np.float32) / 255.0 + image_array[mask_array > 0.5] = -1.0 + return torch.from_numpy(image_array[None].transpose(0, 3, 1, 2)) + + @staticmethod + def _one_image(result: Any) -> Image.Image: + images = getattr(result, "images", None) + if not images or len(images) != 1 or not isinstance(images[0], Image.Image): + raise GenerationError("Diffusers did not return exactly one image") + return images[0] -- cgit v1.2.3