Fal.ai MCP Server

Instructions

Implementation Reference

• src/fal_mcp_server/handlers/utility_handlers.py:162-252 (handler)

  The core handler function for the 'get_pricing' tool. It validates input models, resolves them to endpoint IDs using the ModelRegistry, fetches pricing data from the API, handles errors, and formats the pricing information as text content.

  async def handle_get_pricing(
      arguments: Dict[str, Any],
      registry: ModelRegistry,
  ) -> List[TextContent]:
      """Handle the get_pricing tool."""
      model_inputs = arguments.get("models", [])
      if not model_inputs:
          return [
              TextContent(
                  type="text",
                  text="❌ No models specified. Provide a list of model IDs or aliases.",
              )
          ]
  
      # Resolve all model inputs to endpoint IDs
      endpoint_ids = []
      failed_models = []
      for model_input in model_inputs:
          try:
              endpoint_id = await registry.resolve_model_id(model_input)
              endpoint_ids.append(endpoint_id)
          except ValueError:
              failed_models.append(model_input)
  
      if failed_models:
          return [
              TextContent(
                  type="text",
                  text=f"❌ Unknown model(s): {', '.join(failed_models)}. Use list_models to see available options.",
              )
          ]
  
      # Fetch pricing from API
      try:
          pricing_data = await registry.get_pricing(endpoint_ids)
      except httpx.HTTPStatusError as e:
          logger.error(
              "Pricing API returned HTTP %d for %s: %s",
              e.response.status_code,
              endpoint_ids,
              e,
          )
          return [
              TextContent(
                  type="text",
                  text=f"❌ Pricing API error (HTTP {e.response.status_code})",
              )
          ]
      except httpx.TimeoutException:
          logger.error("Pricing API timeout for %s", endpoint_ids)
          return [
              TextContent(
                  type="text",
                  text="❌ Pricing request timed out. Please try again.",
              )
          ]
      except httpx.ConnectError as e:
          logger.error("Cannot connect to pricing API: %s", e)
          return [
              TextContent(
                  type="text",
                  text="❌ Cannot connect to Fal.ai API. Check your network connection.",
              )
          ]
  
      prices = pricing_data.get("prices", [])
      if not prices:
          return [
              TextContent(
                  type="text",
                  text="No pricing information available for the specified models.",
              )
          ]
  
      # Format output
      lines = ["💰 **Pricing Information**\n"]
      for price_info in prices:
          endpoint_id = price_info.get("endpoint_id", "Unknown")
          unit_price = price_info.get("unit_price", 0)
          unit = price_info.get("unit", "request")
          currency = price_info.get("currency", "USD")
  
          # Format price with currency symbol
          if currency == "USD":
              price_str = f"${unit_price:.4f}".rstrip("0").rstrip(".")
          else:
              price_str = f"{unit_price:.4f} {currency}".rstrip("0").rstrip(".")
  
          lines.append(f"- **{endpoint_id}**: {price_str} per {unit}")
  
      return [TextContent(type="text", text="\n".join(lines))]

• src/fal_mcp_server/tools/utility_tools.py:68-84 (schema)

  The Tool schema definition for 'get_pricing', including name, description, and input schema requiring a list of model IDs or aliases.

  Tool(
      name="get_pricing",
      description="Get pricing information for Fal.ai models. Returns cost per unit (image/video/second) in USD. Use this to check costs before generating content.",
      inputSchema={
          "type": "object",
          "properties": {
              "models": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Model IDs or aliases to get pricing for (e.g., ['flux_schnell', 'fal-ai/kling-video'])",
                  "minItems": 1,
                  "maxItems": 50,
              },
          },
          "required": ["models"],
      },
  ),

• src/fal_mcp_server/server.py:61-85 (registration)

  Registration of the 'get_pricing' handler in the TOOL_HANDLERS dictionary used by the stdio server to map tool names to their handler functions.

  TOOL_HANDLERS = {
      # Utility tools (no queue needed)
      "list_models": handle_list_models,
      "recommend_model": handle_recommend_model,
      "get_pricing": handle_get_pricing,
      "get_usage": handle_get_usage,
      "upload_file": handle_upload_file,
      # Image generation tools
      "generate_image": handle_generate_image,
      "generate_image_structured": handle_generate_image_structured,
      "generate_image_from_image": handle_generate_image_from_image,
      # Image editing tools
      "remove_background": handle_remove_background,
      "upscale_image": handle_upscale_image,
      "edit_image": handle_edit_image,
      "inpaint_image": handle_inpaint_image,
      "resize_image": handle_resize_image,
      "compose_images": handle_compose_images,
      # Video tools
      "generate_video": handle_generate_video,
      "generate_video_from_image": handle_generate_video_from_image,
      "generate_video_from_video": handle_generate_video_from_video,
      # Audio tools
      "generate_music": handle_generate_music,
  }

• src/fal_mcp_server/server_http.py:61-67 (registration)

  Registration of the 'get_pricing' handler in the TOOL_HANDLERS dictionary used by the HTTP/SSE server.

  TOOL_HANDLERS = {
      # Utility tools (no queue needed)
      "list_models": handle_list_models,
      "recommend_model": handle_recommend_model,
      "get_pricing": handle_get_pricing,
      "get_usage": handle_get_usage,
      "upload_file": handle_upload_file,