Below is a command-line tool to fetch the costs of LLM (Large Language Model) deployments in Azure for a set of subscriptions within your directory. The tool can display the total costs, per subscription, and per resource group. Subscriptions can be specified in a configuration file or discovered automatically. It simplifies cost tracking and provides insights into the expenses associated with Azure LLM services. ## Prerequisites - Python (3.11 or higher); dependencies: `requests` (and `pyyaml` if you want to configure subscriptions to track) - Azure account with permissions to access subscription cost data - Azure Cost Management API enabled for your subscriptions - Azure CLI `az` installed (for token generation) ## Setup Instructions ### Optional: Set Azure Access Token By default, the script will fetch the access token on its own using the `az` CLI tool. To generate an Azure Access Token using the Azure CLI: ```bash az account get-access-token --resource https://management.azure.com ``` Copy the `accessToken` value from the output. Set the Access Token as an Environment Variable: - On macOS/Linux: ```bash export AZURE_ACCESS_TOKEN=<your-access-token> ``` - On Windows (Command Prompt): ```cmd set AZURE_ACCESS_TOKEN=<your-access-token> ``` - On Windows (PowerShell): ```powershell $env:AZURE_ACCESS_TOKEN="<your-access-token>" ``` To make this easy, create a script `set_azure_token.sh` with this content: ```shell #!/bin/bash # Get the access token using Azure CLI TOKEN=$(az account get-access-token --resource https://management.azure.com --query accessToken -o tsv) if [ -z "$TOKEN" ]; then echo "Error: Failed to retrieve access token. Make sure you are logged in with 'az login'." exit 1 fi export AZURE_ACCESS_TOKEN="$TOKEN" ``` And then set the token using `source set_azure_token.sh` ### Optional: Configure Subscriptions Update the `config.yaml` file with your Azure subscription details. The file includes a list of subscriptions to track: ```yaml subscriptions: - id: 86c605ce-2d8f-41b6-9102-2d057feb1b5e name: Dealogic Data Science - id: 971b9197-043b-4ec4-ac45-276598220406 name: IONA Data Engineering ``` Add or modify subscription entries as needed for your environment. To use those subscriptions, use the command-line flag `--configure` ## Using the tool ```bash python track_costs.py --help ``` To capture the full cost report in a markdown file: ```bash python track_costs.py > report.md ``` # Code ```python #!/usr/bin/env python3 import os import sys import logging import requests import argparse import time import subprocess import json from datetime import datetime, timedelta logging.basicConfig( level=logging.INFO, format="%(levelname)s: %(message)s", stream=sys.stderr ) def get_azure_token(): """Get Azure access token from environment or Azure CLI.""" token = os.getenv("AZURE_ACCESS_TOKEN") if token: return token try: result = subprocess.run( [ "az", "account", "get-access-token", "--resource", "https://management.azure.com/", "--query", "accessToken", "-o", "tsv", ], capture_output=True, text=True, check=True, ) return result.stdout.strip() except subprocess.CalledProcessError: return None except FileNotFoundError: return None def load_subscriptions(config_file): """Load subscriptions from the configuration file.""" try: import yaml with open(config_file, "r") as file: return yaml.safe_load(file).get("subscriptions", []) except ImportError: logging.error("Error: 'pyyaml' module is not installed.") logging.info("Install pyyaml using: pip install pyyaml") sys.exit(1) def discover_subscriptions(access_token): """Discover all subscriptions the user has access to.""" url = "https://management.azure.com/subscriptions?api-version=2022-12-01" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } try: response = requests.get(url, headers=headers) response.raise_for_status() data = response.json() subscriptions = [] for sub in data.get("value", []): subscriptions.append( {"id": sub.get("subscriptionId"), "name": sub.get("displayName")} ) return subscriptions except requests.exceptions.RequestException as e: logging.error(f"Error discovering subscriptions: {e}") return [] def get_date_range(timeframe): """Calculate start and end dates for a given timeframe.""" today = datetime.now() if timeframe == "this-week": start_of_week = today - timedelta(days=today.weekday()) return start_of_week.replace(hour=0, minute=0, second=0, microsecond=0), today elif timeframe == "7days": end_date = today - timedelta(days=1) start_date = end_date - timedelta(days=6) return start_date, end_date elif timeframe == "month-to-date": return today.replace(day=1, hour=0, minute=0, second=0, microsecond=0), today elif timeframe == "last-month": first_day_current_month = today.replace(day=1) last_day_previous_month = first_day_current_month - timedelta(days=1) return last_day_previous_month.replace(day=1), last_day_previous_month elif timeframe == "30days": end_date = today - timedelta(days=1) start_date = end_date - timedelta(days=29) return start_date, end_date return None, None def fetch_openai_deployments(resource_id, access_token): """Fetch the list of model deployments for an Azure OpenAI resource.""" url = ( f"https://management.azure.com{resource_id}/deployments?api-version=2023-05-01" ) headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } try: response = requests.get(url, headers=headers) response.raise_for_status() data = response.json() deployments = [] for deployment in data.get("value", []): deployment_name = deployment.get("name", "Unknown") model_name = ( deployment.get("properties", {}).get("model", {}).get("name", "Unknown") ) deployments.append({"name": deployment_name, "model": model_name}) return deployments except requests.exceptions.RequestException: return [] def fetch_llm_costs( subscription_id, subscription_name, access_token, timeframe="month-to-date" ): """Fetch the costs of LLM deployments for a given subscription.""" url = f"https://management.azure.com/subscriptions/{subscription_id}/providers/Microsoft.CostManagement/query?api-version=2023-03-01" headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", } if timeframe == "this-week": timeframe_config = {"timeframe": "WeekToDate"} elif timeframe in ["7days", "30days", "last-month"]: start_date, end_date = get_date_range(timeframe) timeframe_config = { "timeframe": "Custom", "timePeriod": { "from": start_date.strftime("%Y-%m-%dT00:00:00Z"), # type: ignore "to": end_date.strftime("%Y-%m-%dT23:59:59Z"), # type: ignore }, } else: timeframe_config = {"timeframe": "MonthToDate"} payload = { "type": "ActualCost", **timeframe_config, "dataset": { "granularity": "None", "aggregation": {"totalCost": {"name": "Cost", "function": "Sum"}}, "grouping": [ {"type": "Dimension", "name": "ServiceName"}, {"type": "Dimension", "name": "ResourceId"}, ], }, } try: response = requests.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() time.sleep(1.0) # Rate limiting to avoid 429 errors rows = data.get("properties", {}).get("rows", []) resources = {} for row in rows: cost = row[0] service_name = row[1] if len(row) > 1 else "Unknown" resource_id = row[2] if len(row) > 2 else "" if "Foundry Models" in service_name and cost > 0: resource_name = "Unknown" if resource_id: parts = resource_id.split("/") if len(parts) > 0: resource_name = parts[-1] deployments = fetch_openai_deployments(resource_id, access_token) if resource_name not in resources: resources[resource_name] = { "cost": 0.0, "deployments": deployments, "resource_id": resource_id, } resources[resource_name]["cost"] += cost return {"subscription": subscription_name, "resources": resources} except requests.exceptions.RequestException as e: logging.error(f"Error fetching costs for subscription {subscription_name}: {e}") return {"subscription": subscription_name, "resources": None} def main(): access_token = get_azure_token() if not access_token: logging.error("Error: Unable to retrieve Azure access token.") logging.error("Please ensure you are logged in to Azure CLI: az login") return 1 parser = argparse.ArgumentParser( description="Fetch costs of LLM deployments in Azure subscriptions" ) parser.add_argument( "--timeframe", choices=["this-week", "7days", "month-to-date", "last-month", "30days"], default="7days", help="Time period for cost analysis: 'this-week' (current week), '7days' (last 7 days, default), 'month-to-date', 'last-month' (previous complete month), '30days' (last 30 days)", ) parser.add_argument( "--configure", nargs="?", const="config.yaml", metavar="CONFIG_FILE", help="Use a configuration file instead of discovering subscriptions (try config.yaml if no file specified)", ) args = parser.parse_args() if args.configure: config_file = args.configure logging.info(f"Loading subscriptions from {config_file}...") subscriptions = load_subscriptions(config_file) else: logging.info("Discovering subscriptions...") subscriptions = discover_subscriptions(access_token) if subscriptions: logging.info(f"Found {len(subscriptions)} subscriptions:") for sub in subscriptions: logging.info(f" - {sub['name']} ({sub['id']})") else: logging.error("No subscriptions found or error during discovery") return if not subscriptions: logging.error("No subscriptions found.") return start_date, end_date = get_date_range(args.timeframe) if start_date and end_date: period_label = ( f"{start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')}" ) else: period_label = "Unknown timeframe" print(f"# Azure AI Foundry model costs\n**Date range: from {period_label}**") total_cost = 0.0 for idx, subscription in enumerate(subscriptions, 1): subscription_id = subscription["id"] subscription_name = subscription["name"] logging.info( f"[{idx}/{len(subscriptions)}] Fetching costs for {subscription_name}..." ) cost_data = fetch_llm_costs( subscription_id, subscription_name, access_token, args.timeframe ) if cost_data["resources"] is not None: resources = cost_data["resources"] if resources: subscription_total = sum(res["cost"] for res in resources.values()) print(f"\n## {cost_data['subscription']}: ${subscription_total:.2f}") for resource_name, resource_info in sorted(resources.items()): print(f"### {resource_name}: ${resource_info['cost']:.2f}") if resource_info["deployments"]: for deployment in resource_info["deployments"]: print(f"- {deployment['name']} ({deployment['model']})") else: print("(Unable to fetch deployment details)") total_cost += subscription_total else: logging.info(f"No Foundry model costs for {cost_data['subscription']}") else: print(f"\n## {cost_data['subscription']}: ERR\n(Error fetching costs)") print(f"\n----\n**Total Foundry model costs: ${total_cost:.2f}**") if __name__ == "__main__": main() ```