Google ColabとMCPで画像生成サーバーを自作！手元のAIから透過素材を量産する方法

# ============================================================
# LINE Sticker MCP Server
# Google Colab / ONE CELL
#
# MCP 1.30.0
# FastMCP
# Streamable HTTP
# Cloudflare Tunnel
#
# HF_TOKEN       : optional
# CLOUDFLARE_TOKEN: required
# ============================================================


# ============================================================
# 0. Imports
# ============================================================

import os
import sys
import subprocess
import time
import importlib.util
import importlib.metadata
import threading
import io
import base64
import socket
import warnings


# ============================================================
# 1. Configuration（★ユーザー設定エリア★）
# ============================================================

# --- 【要編集】Cloudflare Tunnel & 待機設定 -------------------
# Cloudflare Tunnelで設定したご自身の公開URL
PUBLIC_URL = "https://[your-tunnel-domain]"  # [your-tunnel-domain] 例: https://ai-mcp.your-domain.com

# Cloudflare Tunnelの転送先（Service）に指定したポート
PORT = 11434  # [your-port] 例: 11434 や 8000 など（Tunnel側の設定と一致させる）

# --- 【任意設定】MCP・モデル・画像生成パラメータ -------------
HOST = "0.0.0.0"
MCP_PATH = "/mcp"

MODEL_ID = "stabilityai/sdxl-turbo"

IMAGE_WIDTH = 512
IMAGE_HEIGHT = 512
INFERENCE_STEPS = 4
GUIDANCE_SCALE = 0.0

MCP_VERSION_REQUIRED = "1.30.0"

CLOUDFLARED_PATH = "/usr/local/bin/cloudflared"

SERVER_START_TIMEOUT = 30
SERVER_CHECK_INTERVAL = 0.5


# ============================================================
# 2. Utility
# ============================================================

def print_section(title: str):
    print()
    print("=" * 70)
    print(title)
    print("=" * 70)


def get_colab_secret(name: str) -> str:
    """
    Google Colab Secretを取得。
    Colab以外では空文字を返す。
    """
    try:
        from google.colab import userdata

        value = userdata.get(name)

        if value:
            return str(value).strip()

    except Exception:
        pass

    return ""


def get_secret_or_env(name: str) -> str:
    """
    Colab Secret → 環境変数の順で取得。
    """
    value = get_colab_secret(name)

    if not value:
        value = os.environ.get(name, "")

    return str(value).strip()


def package_version(package_name: str) -> str:
    """
    importlib.metadataから安全にバージョン取得。
    """
    try:
        value = importlib.metadata.version(package_name)

        if value:
            return str(value)

    except Exception:
        pass

    return "unknown"


def is_port_open(host: str, port: int) -> bool:
    """
    TCPポートがLISTEN状態か確認。
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    try:
        sock.settimeout(0.5)
        result = sock.connect_ex((host, port))
        return result == 0

    except Exception:
        return False

    finally:
        sock.close()


# ============================================================
# 3. Initial settings / Secrets
# ============================================================

print("=" * 70)
print("LINE Sticker MCP Server")
print("=" * 70)
print("初期設定")
print()


# ------------------------------------------------------------
# HF_TOKEN
# ------------------------------------------------------------
#
# 公開モデルのため必須ではない。
# 設定されていればHugging Face Hubのレート制限緩和等に利用。
#

HF_TOKEN = get_secret_or_env("HF_TOKEN")

if HF_TOKEN:
    os.environ["HF_TOKEN"] = HF_TOKEN
    print("HF_TOKEN        : configured")
else:
    print("HF_TOKEN        : not configured (optional)")


# ------------------------------------------------------------
# CLOUDFLARE_TOKEN
# ------------------------------------------------------------

CLOUDFLARE_TOKEN = get_secret_or_env("CLOUDFLARE_TOKEN")

if CLOUDFLARE_TOKEN:
    print("CLOUDFLARE_TOKEN: configured")
else:
    print("CLOUDFLARE_TOKEN: NOT CONFIGURED")


if not CLOUDFLARE_TOKEN:
    raise RuntimeError(
        "\n"
        "CLOUDFLARE_TOKEN が設定されていません。\n\n"
        "Google Colab 左側の「シークレット（鍵アイコン）」に、\n"
        "以下を登録してください。\n\n"
        "名前:\n"
        "  CLOUDFLARE_TOKEN\n\n"
        "値:\n"
        "  Cloudflare Tunnel Token\n\n"
        "さらに「ノートブック アクセス」をONにしてください。"
    )


# ============================================================
# 4. Dependency check
# ============================================================

print_section("依存パッケージ確認")

# ------------------------------------------------------------
# MCPは1.30.0を固定
# ------------------------------------------------------------

required_modules = [
    "mcp",
    "diffusers",
    "transformers",
    "accelerate",
    "rembg",
    "uvicorn",
    "timm",
]


missing_modules = [
    module
    for module in required_modules
    if importlib.util.find_spec(module) is None
]


# ------------------------------------------------------------
# MCP version check
# ------------------------------------------------------------

installed_mcp_version = package_version("mcp")

mcp_needs_install = (
    installed_mcp_version != MCP_VERSION_REQUIRED
)


if missing_modules or mcp_needs_install:

    print("必要パッケージの追加・調整を行います。")

    if missing_modules:
        print(f"未検出: {missing_modules}")

    if mcp_needs_install:
        print(
            f"MCP version: {installed_mcp_version}"
            f" → {MCP_VERSION_REQUIRED}"
        )

    print()

    install_packages = [
        f"mcp=={MCP_VERSION_REQUIRED}",
        "diffusers",
        "transformers",
        "accelerate",
        "rembg[gpu]",
        "uvicorn",
        "timm",
    ]

    subprocess.run(
        [
            sys.executable,
            "-m",
            "pip",
            "install",
            "-q",
            *install_packages,
        ],
        check=True,
    )

    print("✅ パッケージ調整完了")

    # pipでMCPのバージョンを変更した場合、
    # 既にロードされたPythonモジュールとの混在を避ける。
    if mcp_needs_install:

        print()
        print("⚠️ MCPパッケージを変更しました。")

        # Google Colabではsys.modulesに旧MCPが残る可能性があるため、
        # 必要な場合はランタイム再起動を促す。
        current_mcp = sys.modules.get("mcp")

        if current_mcp is not None:
            raise RuntimeError(
                "\n"
                "MCPのバージョンを変更しました。\n"
                "Colabのランタイムを再起動してから、"
                "このセルをもう一度実行してください。\n\n"
                f"Required MCP: {MCP_VERSION_REQUIRED}\n"
                f"Previous MCP: {installed_mcp_version}"
            )

else:

    print("✅ 必要パッケージはすべてインストール済み")


# ============================================================
# 5. NumPy compatibility patch
# ============================================================

print_section("NumPy互換性確認")

import numpy as np


numpy_patch_applied = False

try:

    import numpy._core._multiarray_umath as _umath

    if not hasattr(_umath, "_blas_supports_fpe"):

        _umath._blas_supports_fpe = (
            lambda x=None: False
        )

        numpy_patch_applied = True

        print("⚠️ NumPy互換性パッチを適用しました")

    else:

        print("✅ NumPy互換性パッチは不要です")

except Exception as e:

    print(
        "⚠️ NumPy互換性確認中に例外が発生しました:"
    )
    print(type(e).__name__, e)


# ============================================================
# 6. GPU / CUDA
# ============================================================

print_section("GPU / CUDA確認")

import torch


print(f"Python        : {sys.version.split()[0]}")
print(f"NumPy         : {np.__version__}")
print(f"PyTorch       : {torch.__version__}")
print(f"CUDA          : {torch.version.cuda}")
print(f"CUDA available: {torch.cuda.is_available()}")


if not torch.cuda.is_available():

    raise RuntimeError(
        "\n"
        "CUDAが利用できません。\n\n"
        "Google Colabで\n"
        "「ランタイム」→「ランタイムのタイプを変更」\n"
        "からGPUを有効にしてください。\n\n"
        "推奨GPU: NVIDIA L4"
    )


GPU_NAME = torch.cuda.get_device_name(0)

print(f"GPU           : {GPU_NAME}")


# ============================================================
# 7. MCP
# ============================================================

print_section("MCP確認")


import mcp

MCP_VERSION = package_version("mcp")

print(f"MCP version   : {MCP_VERSION}")


if MCP_VERSION != MCP_VERSION_REQUIRED:

    raise RuntimeError(
        "\n"
        f"MCP {MCP_VERSION_REQUIRED} が必要ですが、"
        f"現在は {MCP_VERSION} です。\n\n"
        "Colabランタイムを再起動してから、"
        "このセルをもう一度実行してください。"
    )


try:

    from mcp.server.fastmcp import FastMCP

except Exception as e:

    raise RuntimeError(
        "\n"
        "MCP 1.x / FastMCP を読み込めませんでした。\n\n"
        f"MCP version: {MCP_VERSION}\n"
        f"Original error: {type(e).__name__}: {e}\n"
    )


print("✅ FastMCP import OK")


# ============================================================
# 8. Diffusers
# ============================================================

print_section("Diffusers")


from diffusers import AutoPipelineForText2Image


DIFFUSERS_VERSION = package_version("diffusers")

print(f"Diffusers     : {DIFFUSERS_VERSION}")
print("✅ Diffusers import OK")


# ============================================================
# 9. SDXL Turbo
# ============================================================

print_section("SDXL Turbo ロード")

print(f"Model         : {MODEL_ID}")
print(f"Resolution    : {IMAGE_WIDTH}x{IMAGE_HEIGHT}")
print(f"Steps         : {INFERENCE_STEPS}")
print(f"Guidance      : {GUIDANCE_SCALE}")


# ------------------------------------------------------------
# HF Token
# ------------------------------------------------------------

pipeline_kwargs = {
    "dtype": torch.float16,
    "variant": "fp16",
}

if HF_TOKEN:
    pipeline_kwargs["token"] = HF_TOKEN


pipe = AutoPipelineForText2Image.from_pretrained(
    MODEL_ID,
    **pipeline_kwargs,
)


pipe = pipe.to("cuda")


# Attention slicing
pipe.enable_attention_slicing()


print("✅ SDXL Turbo ロード成功")


# ============================================================
# 10. BiRefNet
# ============================================================

print_section("BiRefNet ロード")


from rembg import new_session, remove


rembg_session = new_session(
    "birefnet-general"
)


print("✅ BiRefNet ロード成功")


# ============================================================
# 11. MCP Server
# ============================================================

print_section("MCP Server 構築")


# IMPORTANT:
#
# MCP 1.30.0ではhost / portをFastMCP()に渡す。
#
# run()側にはhost / portを渡さない。
#
# これが今回のTypeError対策。
#

mcp_server = FastMCP(
    "LINE Sticker Generator",
    host=HOST,
    port=PORT,
)


print("Transport     : Streamable HTTP")
print(f"MCP endpoint  : {MCP_PATH}")
print("MCP host      :", HOST)
print("MCP port      :", PORT)
print("✅ FastMCP Server 作成完了")


# ============================================================
# 12. Image generation lock
# ============================================================

# GPU上のSDXL / BiRefNet処理を同時実行しない。
#
# MCPクライアントから複数リクエストが来ても、
# GPUメモリの急増を避けるため1件ずつ処理する。

generation_lock = threading.Lock()


# ============================================================
# 13. Image generation
# ============================================================

def create_transparent_sticker(
    prompt: str,
    extra_style: str = "",
    seed: int | None = None,
):
    """
    LINEスタンプ用の透明PNGを生成。
    """

    if not prompt or not prompt.strip():

        raise ValueError(
            "promptが空です。"
        )


    # --------------------------------------------------------
    # Prompt
    # --------------------------------------------------------

    final_prompt = f"""
A high-quality Japanese LINE sticker illustration.

{prompt}

{extra_style}

Cute expressive cartoon character,
clear facial expression,
clear body pose,
thick clean outline,
simple polished illustration,
vibrant but soft colors,
centered composition,
full character visible,
isolated character,
sticker artwork,
no background,
no scenery,
no frame,
no border,
no text,
no letters,
no typography,
no watermark,
no logo.
"""


    # --------------------------------------------------------
    # Seed
    # --------------------------------------------------------

    if seed is None:

        seed = torch.randint(
            0,
            2**31 - 1,
            (1,),
            device="cuda",
        ).item()


    generator = torch.Generator(
        device="cuda"
    ).manual_seed(seed)


    # --------------------------------------------------------
    # GPU処理
    # --------------------------------------------------------

    with generation_lock:

        # ----------------------------------------------------
        # SDXL Turbo
        # ----------------------------------------------------

        with torch.inference_mode():

            result = pipe(
                prompt=final_prompt,
                num_inference_steps=INFERENCE_STEPS,
                guidance_scale=GUIDANCE_SCALE,
                height=IMAGE_HEIGHT,
                width=IMAGE_WIDTH,
                generator=generator,
            )


        image = result.images[0]


        # ----------------------------------------------------
        # RGBA
        # ----------------------------------------------------

        image = image.convert("RGBA")


        # ----------------------------------------------------
        # Background Removal
        # ----------------------------------------------------

        transparent_image = remove(
            image,
            session=rembg_session,
        )


        # ----------------------------------------------------
        # PNG → Base64
        # ----------------------------------------------------

        buffer = io.BytesIO()

        transparent_image.save(
            buffer,
            format="PNG",
            optimize=True,
        )

        image_base64 = base64.b64encode(
            buffer.getvalue()
        ).decode("utf-8")


        # ----------------------------------------------------
        # GPU memory cleanup
        # ----------------------------------------------------

        del result


    return {
        "success": True,
        "format": "PNG",
        "width": transparent_image.width,
        "height": transparent_image.height,
        "seed": seed,
        "mime_type": "image/png",
        "data": image_base64,
    }


# ============================================================
# 14. MCP Tool: generate
# ============================================================

@mcp_server.tool()
def generate_transparent_stamp(
    prompt: str,
    extra_style: str = "",
    seed: int | None = None,
) -> dict:
    """
    LINEスタンプ用の透明PNG画像を生成します。

    prompt:
        キャラクター、表情、ポーズ、シチュエーション。

    extra_style:
        絵柄や追加スタイル。

    seed:
        再現性が必要な場合に指定。
    """

    try:

        return create_transparent_sticker(
            prompt=prompt,
            extra_style=extra_style,
            seed=seed,
        )

    except Exception as e:

        return {
            "success": False,
            "error": type(e).__name__,
            "message": str(e),
        }


# ============================================================
# 15. MCP Tool: status
# ============================================================

@mcp_server.tool()
def server_status() -> dict:
    """
    MCPサーバーとGPUの状態を返します。
    """

    return {
        "status": "ok",
        "server": "LINE Sticker Generator",
        "model": MODEL_ID,
        "transport": "streamable-http",
        "endpoint": f"{PUBLIC_URL}{MCP_PATH}",
        "device": "cuda",
        "gpu": torch.cuda.get_device_name(0),
        "cuda": torch.version.cuda,
        "torch": torch.__version__,
        "mcp": MCP_VERSION,
        "diffusers": DIFFUSERS_VERSION,
    }


# ============================================================
# 16. Local MCP server startup
# ============================================================

print_section("MCP Server 起動")


mcp_server_error = None


def run_mcp_server():

    global mcp_server_error

    try:

        print(
            "MCP Server starting..."
        )

        # ====================================================
        # IMPORTANT:
        #
        # MCP 1.30.0
        #
        # host / port:
        #     FastMCP(...)
        #
        # transport:
        #     run(...)
        #
        # ====================================================

        mcp_server.run(
            transport="streamable-http",
        )

    except Exception as e:

        mcp_server_error = e

        print()
        print("❌ MCP Serverでエラーが発生しました。")
        print(
            f"{type(e).__name__}: {e}"
        )


server_thread = threading.Thread(
    target=run_mcp_server,
    daemon=True,
    name="mcp-server",
)


server_thread.start()


print("MCP Server起動待機中...")


# ============================================================
# 17. Wait for local server
# ============================================================

deadline = (
    time.time()
    + SERVER_START_TIMEOUT
)


server_started = False


while time.time() < deadline:

    if is_port_open(
        "127.0.0.1",
        PORT,
    ):

        server_started = True
        break


    if not server_thread.is_alive():

        break


    time.sleep(
        SERVER_CHECK_INTERVAL
    )


# ------------------------------------------------------------
# Startup failure
# ------------------------------------------------------------

if not server_started:

    if mcp_server_error is not None:

        raise RuntimeError(
            "\n"
            "MCP Serverの起動に失敗しました。\n\n"
            f"{type(mcp_server_error).__name__}: "
            f"{mcp_server_error}"
        )

    if not server_thread.is_alive():

        raise RuntimeError(
            "\n"
            "MCP Serverスレッドが終了しました。\n"
            "上のMCP Server起動ログを確認してください。"
        )

    raise RuntimeError(
        "\n"
        f"MCP Serverが"
        f"{SERVER_START_TIMEOUT}秒以内に"
        f"Port {PORT} で起動しませんでした。\n\n"
        "MCP Serverの起動ログを確認してください。"
    )


print("✅ MCP Server起動成功")
print(
    f"Local MCP URL : "
    f"http://127.0.0.1:{PORT}{MCP_PATH}"
)


# ============================================================
# 18. cloudflared
# ============================================================

print_section("cloudflared確認")


if not os.path.exists(CLOUDFLARED_PATH):

    print(
        "cloudflaredをインストールします..."
    )

    subprocess.run(
        [
            "wget",
            "-q",
            "https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64",
            "-O",
            CLOUDFLARED_PATH,
        ],
        check=True,
    )

    subprocess.run(
        [
            "chmod",
            "+x",
            CLOUDFLARED_PATH,
        ],
        check=True,
    )

    print(
        "✅ cloudflared インストール完了"
    )

else:

    print(
        "✅ cloudflared はインストール済み"
    )


# ============================================================
# 19. Stop previous Cloudflare Tunnel
# ============================================================

print_section("既存Cloudflare Tunnel確認")


subprocess.run(
    ["pkill", "-f", "cloudflared"],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
)


time.sleep(2)


# ============================================================
# 20. Start Cloudflare Tunnel
# ============================================================

print_section("Cloudflare Tunnel 起動")


cloudflared_process = subprocess.Popen(
    [
        CLOUDFLARED_PATH,
        "tunnel",
        "--no-autoupdate",
        "run",
        "--token",
        CLOUDFLARE_TOKEN,
    ],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
)


# 少しだけプロセス状態を確認
time.sleep(2)


if cloudflared_process.poll() is not None:

    raise RuntimeError(
        "\n"
        "Cloudflare Tunnelが起動直後に終了しました。\n\n"
        "Cloudflare Tunnel Tokenが正しいか確認してください。"
    )


print(
    "✅ Cloudflare Tunnel 起動"
)


# ============================================================
# 21. Final status
# ============================================================

print()
print("=" * 70)
print("🚀 LINE STICKER MCP SERVER READY")
print("=" * 70)

print()
print(f"Server        : LINE Sticker Generator")
print(f"MCP           : {MCP_VERSION} / FastMCP")
print(f"Transport     : Streamable HTTP")
print(f"Model         : {MODEL_ID}")
print(f"Resolution    : {IMAGE_WIDTH}x{IMAGE_HEIGHT}")
print(f"Steps         : {INFERENCE_STEPS}")
print(f"GPU           : {GPU_NAME}")
print(f"PyTorch       : {torch.__version__}")
print(f"CUDA          : {torch.version.cuda}")
print(f"Diffusers     : {DIFFUSERS_VERSION}")
print(f"Port          : {PORT}")
print(f"Public URL    : {PUBLIC_URL}")
print(f"MCP URL       : {PUBLIC_URL}{MCP_PATH}")

print()
print("MCP Tools:")
print("  generate_transparent_stamp")
print("  server_status")

print()
print("HF_TOKEN      :", "configured" if HF_TOKEN else "not configured (optional)")
print("Cloudflare    : configured")
print()
print("Status        : ✅ READY")
print("=" * 70)