Fal.ai MCP Server

Instructions

Implementation Reference

• src/fal_mcp_server/handlers/utility_handlers.py:255-352 (handler)

  The handler function that implements the core logic for the 'get_usage' tool. It processes input arguments, resolves model IDs to endpoint IDs, fetches usage data from the model registry, handles errors, and formats the response as markdown text.

  async def handle_get_usage(
      arguments: Dict[str, Any],
      registry: ModelRegistry,
  ) -> List[TextContent]:
      """Handle the get_usage tool."""
      # Parse dates
      today = datetime.now().date()
      start_str = arguments.get("start") or (today - timedelta(days=7)).isoformat()
      end_str = arguments.get("end") or today.isoformat()
  
      # Resolve endpoint filters if provided
      model_inputs = arguments.get("models", [])
      endpoint_ids = []
      failed_models = []
      if model_inputs:
          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 usage data
      try:
          usage_data = await registry.get_usage(
              start=start_str, end=end_str, endpoint_ids=endpoint_ids or None
          )
      except httpx.HTTPStatusError as e:
          logger.error(
              "Usage API returned HTTP %d: %s",
              e.response.status_code,
              e,
          )
          if e.response.status_code == 403:
              return [
                  TextContent(
                      type="text",
                      text="❌ Access denied. Your API key doesn't have permission to view usage data. Contact your workspace admin.",
                  )
              ]
          return [
              TextContent(
                  type="text",
                  text=f"❌ Usage API error (HTTP {e.response.status_code})",
              )
          ]
      except httpx.TimeoutException:
          return [
              TextContent(
                  type="text",
                  text="❌ Usage request timed out. Please try again.",
              )
          ]
      except httpx.ConnectError as e:
          logger.error("Cannot connect to usage API: %s", e)
          return [
              TextContent(
                  type="text",
                  text="❌ Cannot connect to Fal.ai API. Check your network connection.",
              )
          ]
  
      # Format output
      total_cost = usage_data.get("total_cost", 0)
      currency = usage_data.get("currency", "USD")
      breakdown = usage_data.get("breakdown", [])
  
      if currency == "USD":
          total_str = f"${total_cost:.2f}"
      else:
          total_str = f"{total_cost:.2f} {currency}"
  
      lines = [
          f"## Usage Report: {start_str} to {end_str}\n",
          f"**Total Cost**: {total_str}\n",
      ]
  
      if breakdown:
          lines.append("### Breakdown by Model\n")
          for item in breakdown:
              endpoint_id = item.get("endpoint_id", "Unknown")
              quantity = item.get("quantity", 0)
              cost = item.get("cost", 0)
              if currency == "USD":
                  cost_str = f"${cost:.2f}"
              else:
                  cost_str = f"{cost:.2f} {currency}"
              lines.append(f"- **{endpoint_id}**: {quantity} requests, {cost_str}")
  
      return [TextContent(type="text", text="\n".join(lines))]

• src/fal_mcp_server/tools/utility_tools.py:85-107 (schema)

  The JSON schema defining the input parameters for the 'get_usage' tool, including optional start/end dates and model filters.

  Tool(
      name="get_usage",
      description="Get usage and spending history for your Fal.ai workspace. Shows quantity, cost, and breakdown by model. Requires admin API key.",
      inputSchema={
          "type": "object",
          "properties": {
              "start": {
                  "type": "string",
                  "description": "Start date (YYYY-MM-DD format). Defaults to 7 days ago.",
              },
              "end": {
                  "type": "string",
                  "description": "End date (YYYY-MM-DD format). Defaults to today.",
              },
              "models": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Filter by specific model IDs/aliases (optional)",
              },
          },
          "required": [],
      },
  ),

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

  Registration of the 'get_usage' handler in the TOOL_HANDLERS dictionary, which maps tool names to their handler functions in the MCP 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,
      # 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/model_registry.py:657-705 (helper)

  Helper method in ModelRegistry that performs the actual HTTP request to the Fal.ai API to retrieve usage data.

  async def get_usage(
      self,
      start: Optional[str] = None,
      end: Optional[str] = None,
      endpoint_ids: Optional[List[str]] = None,
  ) -> Dict[str, Any]:
      """
      Fetch usage and spending history.
  
      Args:
          start: Start date (YYYY-MM-DD format)
          end: End date (YYYY-MM-DD format)
          endpoint_ids: Optional list of endpoint IDs to filter by
  
      Returns:
          Dict with "time_series" and "summary" usage data
  
      Raises:
          httpx.HTTPStatusError: If API request fails (e.g., 401 for non-admin key)
      """
      client = await self._get_http_client()
  
      # Build query params
      params: Dict[str, Any] = {"expand": "summary"}
      if start:
          params["start"] = start
      if end:
          params["end"] = end
  
      # Add endpoint_id filters if specified
      if endpoint_ids:
          # For multiple endpoint IDs, we need to make the request with repeated params
          # httpx supports this with a list of tuples
          param_tuples: List[Tuple[str, Union[str, int, float, bool, None]]] = [
              ("expand", "summary")
          ]
          if start:
              param_tuples.append(("start", start))
          if end:
              param_tuples.append(("end", end))
          for eid in endpoint_ids:
              param_tuples.append(("endpoint_id", eid))
          response = await client.get("/models/usage", params=param_tuples)
      else:
          response = await client.get("/models/usage", params=params)
  
      response.raise_for_status()
      result: Dict[str, Any] = response.json()
      return result