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
61
62
63
64
|
"""Verify generate_image over the Streamable HTTP transport with the real GPU backend.
This script connects to a running ``mcp-sd15`` server that was started with
``transport = "streamable-http"`` and calls ``generate_image`` through the HTTP
endpoint. It exercises the full request path: HTTP connect → MCP initialize →
tools/call → PNG response.
"""
from __future__ import annotations
import argparse
import asyncio
import base64
from pathlib import Path
from mcp import Client
from mcp.types import ImageContent
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--url", default="http://127.0.0.1:8123/mcp")
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:
async with Client(args.url, raise_exceptions=True) as client:
tools = (await client.list_tools()).tools
tool_names = {tool.name for tool in tools}
if tool_names != {"generate_image", "generate_guided_image", "inpaint_image"}:
raise RuntimeError(f"unexpected tools: {tool_names}")
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__":
raise SystemExit(main())
|
