import asyncio
import json
import os
from google.adk.agents.invocation_context import InvocationContext
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.tools.tool_context import ToolContext
def payload(result):
if result.get("isError") or result.get("error"):
raise RuntimeError("Magic Hour returned an error; keep the saved project ID.")
if isinstance(result.get("structuredContent"), dict):
return result["structuredContent"]
for block in result.get("content", []):
if block.get("type") == "text":
try:
data = json.loads(block.get("text", ""))
except json.JSONDecodeError:
continue
if isinstance(data, dict):
return data
raise RuntimeError("Magic Hour returned no project or account metadata.")
async def check_project(api_key, project_id):
service = InMemorySessionService()
session = await service.create_session(app_name="magic_hour", user_id="local_user")
context = ToolContext(
InvocationContext(
session_service=service, session=session, invocation_id="check_project"
)
)
toolset = McpToolset(
connection_params=StreamableHTTPConnectionParams(
url="https://mcp.magichour.ai/",
headers={"Authorization": "Bearer " + api_key},
timeout=30,
sse_read_timeout=60,
terminate_on_close=True,
)
)
try:
tools = {tool.name: tool for tool in await toolset.get_tools()}
print(f"Discovered {len(tools)} tools.")
payload(await tools["account_retrieve"].run_async(args={}, tool_context=context))
print("Magic Hour API authorization confirmed.")
project = payload(
await tools["image_projects_retrieve_details"].run_async(
args={"id": project_id}, tool_context=context
)
)
if project.get("id") != project_id:
raise RuntimeError("Project response did not match the saved ID.")
print("Image project status:", project["status"])
finally:
try:
await toolset.close()
finally:
await service.delete_session(
app_name=session.app_name, user_id=session.user_id, session_id=session.id
)
if __name__ == "__main__":
asyncio.run(
check_project(
os.environ["MAGIC_HOUR_API_KEY"], os.environ["MAGIC_HOUR_IMAGE_PROJECT_ID"]
)
)
