Usage & Cost Admin API Cookbook
A practical guide to programmatically accessing your Haijun API usage and cost data
What You Can Do
ds)
Track usage across models, workspaces, and API keys
Analyze cache efficiency and server tool usage
Cost Analysis:
teams/projects by workspace
Cache Analysis
: Measure and improve cache efficiency
Financial Reporting
: Generate executive summaries and budget reports
API Overview
"Admin API key required. Set JUGLOW_ADMIN_API_KEY environment variable."
)
if not self.api_key.startswith("sk-ant-admin"):
raise ValueError("Invalid Admin API key format.")
self.base_url = "https://haijun.my.id/v1/organizations"
self.headers = {
"juglow-version": "2023-06-01",
"x-api-key": self.api_key,
"Content-Type": "application/json",
}
def _make_request(self, endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
"""Make authenticated request with basic error handling."""
url = f"{self.base_url}/{endpoint}"
try:
response = requests.get(url, headers=self.headers, params=params, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
raise ValueError("Invalid API key or insufficient permissions") from e
elif response.status_code == 429:
raise requests.exceptions.RequestException(
"Rate limit exceeded - try again later"
) from e
else:
raise requests.exceptions.RequestException(f"API error: {e}") from e
Test connection
def test_connection():
try:
client = JuglowAdminAPI()
Simple test query - snap to start of day to align with bucket boundaries
params = {
"starting_at": (
datetime.combine(datetime.utcnow(), time.min) - timedelta(days=1)
).strftime("%Y-%m-%dT%H:%M:%SZ"),
"ending_at": datetime.combine(datetime.utcnow(), time.min).strftime(
"%Y-%m-%dT%H:%M:%SZ"
),
"bucket_width": "1d",
"limit": 1,
}
client._make_request("usage_report/messages", params)
print("✅ Connection successful!")
return client
except Exception as e:
print(f"❌ Connection failed: {e}")
return None
client = test_connection()
Basic Usage & Cost Tracking
"ending_at": end_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"bucket_width": "1d", # Only 1d supported for cost API
"limit": min(days_back, 31), # Max 31 days per request
}
return client._make_request("cost_report", params)
def analyze_cost_data(response):
"""Process and display cost data."""
if not response or not response.get("data"):
print("No cost data found.")
return
total_cost_minor_units = 0
daily_costs = []
for bucket in response["data"]:
date = bucket["starting_at"][:10]
Sum all costs in this bucket
bucket_cost = 0
for result in bucket["results"]:
Convert string amounts to float if needed
amount = result.get("amount", 0)
if isinstance(amount, str):
try:
amount = float(amount)
except (ValueError, TypeError):
amount = 0
bucket_cost += amount
daily_costs.append(
{
"date": date,
"cost_minor_units": bucket_cost,
"cost_usd": bucket_cost / 100, # Convert to dollars
}
)
total_cost_minor_units += bucket_cost
total_cost_usd = total_cost_minor_units / 100
print("💰 Cost Summary:")
print(f"Total cost: ${total_cost_usd:.4f}")
print(f"Average daily cost: ${total_cost_usd / len(daily_costs):.4f}")
return daily_costs
Example usage
if client:
cost_response = get_daily_costs(client, days_back=7)
daily_costs = analyze_cost_data(cost_response)
💰 Cost Summary: Total cost: $83.7574 Average daily cost: $11.9653 Grouping, Filtering & Pagination Time Granularity Options Usage API supports three granularities:
ext-indent:-8ch"> while page_count < max_pages:
current_params = params.copy()
if next_page:
current_params["page"] = next_page
response = client._make_request("cost_report", current_params)
page_count += 1
Process this page's data
for bucket in response.get("data", []):
date = bucket["starting_at"][:10]
for result in bucket["results"]:
Handle both string and numeric amounts
amount = result.get("amount", 0)
if isinstance(amount, str):
try:
amount = float(amount)
except (ValueError, TypeError):
amount = 0
rows.append(
{
"date": date,
"workspace_id": result.get(
"workspace_id", ""
), # null for default workspace
"description": result.get("description", ""),
"currency": result.get("currency", "USD"),
"amount_usd": amount / 100,
}
)
Check if there's more data
if not response.get("has_more", False):
break
next_page = response.get("next_page")
if not next_page:
break
if rows:
with open(output_file, "w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=rows[0].keys())
writer.writeheader()
writer.writerows(rows)
print(f"✅ Exported {len(rows)} cost records to {output_file}")
else:
print(f"No cost data to export for the last {days_back} days")
print("💡 Try increasing days_back or check if you have recent API usage")
except Exception as e:
print(f"❌ Cost export failed: {e}")
Example usage
if client:
export_usage_to_csv(client, "my_usage_data.csv", days_back=14)
export_costs_to_csv(client, "my_cost_data.csv", days_back=14)
✅ Exported 36 rows to my_usage_data.csv ✅ Exported 72 cost records to my_cost_data.csv Wrapping Up This cookbook covers the essential patterns for working with the Usage & Cost Admin API: