import asyncio
import base64
import os
from pathlib import Path
from httpx2 import AsyncClient, Timeout
from llama_index.tools.mcp import BasicMCPClient, McpToolSpec
async def main():
async with AsyncClient(
headers={"Authorization": "Bearer " + os.environ["MAGIC_HOUR_API_KEY"]},
timeout=Timeout(30.0, read=60.0),
) as http:
client = BasicMCPClient(
"https://mcp.magichour.ai/",
http_client=http,
timeout=60,
)
tools = await McpToolSpec(
client=client,
).to_tool_list_async()
print(f"Loaded {len(tools)} tools.")
by_name = {tool.metadata.name: tool for tool in tools}
account = (await by_name["account_retrieve"].acall()).raw_output
if account.is_error:
raise RuntimeError("Magic Hour authorization failed.")
print("Magic Hour API authorization confirmed.")
result = (
await by_name["wait_for_image_project"].acall(
id=os.environ["MAGIC_HOUR_IMAGE_PROJECT_ID"],
timeout_seconds=30,
include_inline_downloads=True,
max_inline_downloads=1,
max_bytes_per_download=8388608,
)
).raw_output
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())
