1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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()
|
