diff options
Diffstat (limited to 'script/smoke_generate_image.py')
| -rw-r--r-- | script/smoke_generate_image.py | 60 |
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() |
