import asyncio
import os
from pathlib import Path
os.environ["FASTMCP_TELEMETRY_MODE"] = "off"
os.environ["FASTMCP_CHECK_FOR_UPDATES"] = "off"
from pydantic_ai.mcp import MCPToolset
from pydantic_ai.messages import BinaryImage
async def main():
async with MCPToolset(
"https://mcp.magichour.ai/",
headers={"Authorization": "Bearer " + os.environ["MAGIC_HOUR_API_KEY"]},
init_timeout=30,
read_timeout=60,
max_retries=0,
tool_error_behavior="error",
) as toolset:
tools = await toolset.list_tools()
print(f"Discovered {len(tools)} tools.")
await toolset.direct_call_tool("account_retrieve", {})
print("Magic Hour API authorization confirmed.")
project_id = os.environ["MAGIC_HOUR_IMAGE_PROJECT_ID"]
project = await toolset.direct_call_tool(
"wait_for_image_project",
{"id": project_id, "timeout_seconds": 30, "include_inline_downloads": False},
)
if not isinstance(project, dict) or project.get("status") != "complete":
raise RuntimeError("Image project is not complete; keep the saved ID and inspect its status.")
parts = await toolset.direct_call_tool(
"wait_for_image_project",
{
"id": project_id,
"timeout_seconds": 30,
"include_inline_downloads": True,
"max_inline_downloads": 1,
"max_bytes_per_download": 1048576,
},
)
image = next((part for part in parts if isinstance(part, BinaryImage)), None)
if image is None:
raise RuntimeError("Completed image was not embedded; retrieve its exact download URL.")
Path("magic-hour-image.bin").write_bytes(image.data)
print("Saved magic-hour-image.bin; inspect its format before renaming.")
asyncio.run(main())
