import json
import os
from datetime import timedelta
from pathlib import Path
os.environ["OTEL_SDK_DISABLED"] = "true"
from strands.tools.mcp import MCPClient
def metadata(result):
if result.get("status") != "success" or result.get("isError"):
raise RuntimeError("Magic Hour returned an error; retain the saved project ID.")
if isinstance(result.get("structuredContent"), dict):
return result["structuredContent"]
for block in result.get("content", []):
if "text" in block:
try:
value = json.loads(block["text"])
except json.JSONDecodeError:
continue
if isinstance(value, dict):
return value
raise RuntimeError("Magic Hour returned no project or account metadata.")
project_file = Path("magic-hour-image-project.json")
with MCPClient(
url="https://mcp.magichour.ai/",
headers={"Authorization": "Bearer " + os.environ["MAGIC_HOUR_API_KEY"]},
startup_timeout=30,
) as client:
print(f"Discovered {len(client.list_tools_sync())} tools.")
metadata(
client.call_tool_sync(
tool_use_id="check-account",
name="account_retrieve",
arguments={},
read_timeout_seconds=timedelta(seconds=30),
)
)
print("Magic Hour API authorization confirmed.")
if project_file.exists():
accepted = json.loads(project_file.read_text())
else:
project_file.touch(exist_ok=False)
accepted = metadata(
client.call_tool_sync(
tool_use_id="create-image",
name="ai_image_generator_create_image",
arguments={
"name": "Strands product draft",
"model": "flux-2-klein",
"resolution": "640px",
"aspect_ratio": "1:1",
"image_count": 1,
"style": {
"prompt": "Studio product photograph of an original unbranded cobalt-blue reusable water bottle on a pale stone pedestal, sunrise desert backdrop, no people, text, or logos.",
"tool": "general",
},
},
read_timeout_seconds=timedelta(seconds=45),
)
)
temporary_file = project_file.with_suffix(".tmp")
temporary_file.write_text(
json.dumps({"id": accepted["id"], "credits_charged": accepted["credits_charged"]})
)
temporary_file.replace(project_file)
print("Saved project:", accepted["id"])
print("Credits charged:", accepted["credits_charged"])
project = metadata(
client.call_tool_sync(
tool_use_id="wait-for-image",
name="wait_for_image_project",
arguments={
"id": accepted["id"],
"timeout_seconds": 90,
"include_inline_downloads": False,
},
read_timeout_seconds=timedelta(seconds=120),
)
)
if project.get("id") != accepted["id"] or project.get("status") != "complete":
raise RuntimeError("Image is not complete; resume the saved project, not creation.")
result = client.call_tool_sync(
tool_use_id="download-image",
name="fetch_image_download",
arguments={"download_url": project["downloads"][0]["url"], "max_bytes": 8388608},
read_timeout_seconds=timedelta(seconds=30),
)
if result.get("status") != "success" or result.get("isError"):
raise RuntimeError("Download failed; retain the saved project ID.")
image = next(block["image"] for block in result["content"] if "image" in block)
output = Path("magic-hour-image." + image["format"])
output.write_bytes(image["source"]["bytes"])
print("Saved image:", output)
