summaryrefslogtreecommitdiff
path: root/script/smoke_generate_image.py
diff options
context:
space:
mode:
Diffstat (limited to 'script/smoke_generate_image.py')
-rw-r--r--script/smoke_generate_image.py60
1 files changed, 60 insertions, 0 deletions
diff --git a/script/smoke_generate_image.py b/script/smoke_generate_image.py
new file mode 100644
index 0000000..9b52253
--- /dev/null
+++ b/script/smoke_generate_image.py
@@ -0,0 +1,60 @@
+"""Run generate_image through an in-process MCP client with the real GPU backend."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import base64
+from pathlib import Path
+
+from mcp import Client
+from mcp.types import ImageContent
+
+from mcp_sd15.config import load_config
+from mcp_sd15.model_manager import ModelManager
+from mcp_sd15.server import create_server
+from mcp_sd15.service import ImageGenerationService
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--config", type=Path, required=True)
+ parser.add_argument("--output", type=Path, required=True)
+ parser.add_argument(
+ "--prompt",
+ default="a small red cube on a white table, studio photograph, detailed",
+ )
+ parser.add_argument("--negative-prompt", default="blurry, low quality")
+ parser.add_argument("--size", default="square_512")
+ return parser.parse_args()
+
+
+async def run(args: argparse.Namespace) -> None:
+ config = load_config(args.config)
+ backend = ModelManager(config)
+ server = create_server(ImageGenerationService(config, backend))
+ async with Client(server, raise_exceptions=True) as client:
+ result = await client.call_tool(
+ "generate_image",
+ {
+ "positive_prompt": args.prompt,
+ "negative_prompt": args.negative_prompt,
+ "size": args.size,
+ },
+ )
+ if result.is_error or len(result.content) != 1:
+ raise RuntimeError(f"generate_image failed: {result}")
+ content = result.content[0]
+ if not isinstance(content, ImageContent) or content.mime_type != "image/png":
+ raise RuntimeError(f"generate_image returned unexpected content: {content}")
+ args.output.parent.mkdir(parents=True, exist_ok=True)
+ args.output.write_bytes(base64.b64decode(content.data, validate=True))
+ print(f"generated={args.output.resolve()}")
+
+
+def main() -> None:
+ asyncio.run(run(parse_args()))
+
+
+if __name__ == "__main__":
+ main()