import asyncio
import base64
import os
from pathlib import Path
from agents import set_tracing_disabled
from agents.mcp import MCPServerStreamableHttp
set_tracing_disabled(True)
async def main():
async with MCPServerStreamableHttp(
name="Magic Hour",
params={
"url": "https://mcp.magichour.ai/",
"timeout": 30,
"sse_read_timeout": 60,
"headers": {
"Authorization": "Bearer " + os.environ["MAGIC_HOUR_API_KEY"]
},
},
client_session_timeout_seconds=60,
max_retry_attempts=0,
) as server:
tools = await server.list_tools()
print(f"Discovered {len(tools)} tools.")
account = await server.call_tool("account_retrieve", {})
if account.is_error:
raise RuntimeError("Magic Hour authorization failed.")
print("Magic Hour API authorization confirmed.")
result = await server.call_tool(
"wait_for_image_project",
{
"id": os.environ["MAGIC_HOUR_IMAGE_PROJECT_ID"],
"timeout_seconds": 30,
"include_inline_downloads": True,
"max_inline_downloads": 1,
"max_bytes_per_download": 1048576,
},
)
if result.is_error:
raise RuntimeError("Recovery failed; keep the saved project ID.")
project = result.structured_content or {}
if project.get("status") != "complete":
raise RuntimeError("Image project is not complete; keep the saved ID and inspect its status.")
image = next((item for item in result.content if item.type == "image"), None)
if image is None:
raise RuntimeError("Completed image was not embedded; retrieve its exact download URL.")
Path("magic-hour-image.bin").write_bytes(base64.b64decode(image.data, validate=True))
print("Saved magic-hour-image.bin; inspect its format before renaming.")
asyncio.run(main())
