text_to_image_mystic_sync

Instructions

{
  "data": {
    "generated": [
      "https://openapi-generator.tech",
      "https://openapi-generator.tech"
    ],
    "task_id": "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
    "status": "CREATED"
  }
}

{
  "message": "message"
}

{
  "message": "message"
}

{
  "message": "Internal Server Error"
}

{
  "message": "Service Unavailable. Please try again later."
}

Implementation Reference

• src/infrastructure/mystic_service.py:16-103 (handler)

  Core implementation of the text-to-image generation: POSTs payload to /v1/ai/mystic to create task, polls /v1/ai/mystic/{task_id} until COMPLETED, with retries, progress reporting, and timeout handling.

  async def generate_image_with_polling(
      self,
      payload: dict[str, Any],
      ctx: Context,
      poll_interval: float = 5.0,
      timeout: float = 120.0,
  ) -> dict[str, Any]:
      start_time = time.time()
      await ctx.info("🎨 Starting image generation with Mystic...")
  
      # Create a fresh client with current configuration for this request
      client = HttpClientFactory.create_dynamic_client()
      
      try:
          # 1. Make POST request to create the task
          await ctx.debug(f"Sending POST request with payload: {payload}")
          response = await client.post("/v1/ai/mystic", json=payload)
          response.raise_for_status()
          data = response.json()
          ctx.info(f"Response: {data}")
          task_id = data.get("data", {}).get("task_id")
          if not task_id:
              await ctx.error("❌ No task_id received from Mystic")
              raise Exception("No task_id received from Mystic")
  
          await ctx.info(f"✅ Task created with ID: {task_id}")
  
          # 2. Poll until it's COMPLETED or fails
          status_url = f"/v1/ai/mystic/{task_id}"
          elapsed = 0.0
          poll_count = 0
  
          while elapsed < timeout:
              await asyncio.sleep(poll_interval)
              elapsed = time.time() - start_time
              poll_count += 1
  
              await ctx.debug(f"🔍 Query #{poll_count} - Elapsed time: {elapsed:.1f}s")
  
              # Retry GET up to 3 times if it fails
              status_data = None
  
              for attempt in range(3):  # 3 attempts maximum
                  try:
                      status_resp = await client.get(status_url)
                      status_resp.raise_for_status()
                      status_data = status_resp.json()
                      break  # If it works, exit the retry loop
                  except Exception as e:
                      if attempt < 2:  # If not the last attempt
                          await ctx.warning(
                              f"⚠️ Error in status query (attempt {attempt + 1}/3): {type(e).__name__}: {str(e)}"
                          )
                          await asyncio.sleep(1)  # Wait 1s before retrying
                      else:
                          await ctx.error(
                              f"❌ Error after 3 attempts querying status: {type(e).__name__}: {str(e)}"
                          )
                          raise Exception(
                              f"Error querying status after 3 attempts: {type(e).__name__}: {str(e)}"
                          )
  
              if status_data is None:
                  continue  # This shouldn't happen, but for safety
  
              mystic_status = status_data.get("data", {}).get("status")
  
              await ctx.report_progress(progress=poll_count, total=timeout, message=mystic_status)
              await ctx.info(
                  f"📊 Current status: {mystic_status} (query #{poll_count}, {elapsed:.1f}s elapsed)"
              )
  
              if mystic_status == "COMPLETED":
                  await ctx.report_progress(progress=100.0, total=100.0)
                  await ctx.info(f"🎉 Image generated successfully in {elapsed:.1f} seconds!")
                  result_data = status_data["data"]
                  return result_data if isinstance(result_data, dict) else {}
              if mystic_status in ("FAILED", "CANCELLED"):  # other error states
                  await ctx.error(f"❌ Mystic task failed: {mystic_status} after {elapsed:.1f}s")
                  raise Exception(f"Mystic task failed: {mystic_status}")
  
          await ctx.warning(f"⏰ Timeout after {elapsed:.1f}s waiting for Mystic to finish")
          raise TimeoutError("Timeout waiting for Mystic to complete the task")
          
      finally:
          # Always close the client to free resources
          await client.aclose()

• src/application/server_factory.py:59-75 (registration)

  Transforms the original OpenAPI POST /v1/ai/mystic tool into a synchronous 'text_to_image_mystic_sync' tool using the polling transform_fn.

  def mcp_component_fn(route: Any, component: Any) -> Any:
      # Only for the Mystic POST mystic tool
      if (
          route.path == "/v1/ai/mystic"
          and route.method == "POST"
          and isinstance(component, OpenAPITool)
      ):
          # Create a transformed tool
          transformed = Tool.from_tool(
              component,
              name="text_to_image_mystic_sync",
              transform_fn=mystic_transform_fn,
          )
          component.disable()
          transformed_tools.append(transformed)
          return transformed

• src/application/server_factory.py:52-58 (handler)

  The direct handler (transform_fn) for the 'text_to_image_mystic_sync' tool, which delegates to MysticService.generate_image_with_polling.

  async def mystic_transform_fn(**kwargs: Any) -> dict[str, Any]:
      # Get the active context using the dependency function
      ctx = get_context()
      # kwargs are the POST arguments
      return await mystic_service.generate_image_with_polling(kwargs, ctx, timeout=300.0)

• src/application/server_factory.py:93-94 (registration)

  Adds the transformed 'text_to_image_mystic_sync' tool to the FastMCP server instance.

  for tool in transformed_tools:
      server.add_tool(tool)