Claude API Usage & Cost API — APIコストをプログラムで監視・最適化する方法

# 環境変数にAdmin APIキーを設定
export ADMIN_API_KEY="sk-ant-admin-your-key-here"

# 直近7日間の日次利用量を取得
curl "https://api.anthropic.com/v1/organizations/usage_report/messages?\
starting_at=2026-03-17T00:00:00Z&\
ending_at=2026-03-24T00:00:00Z&\
bucket_width=1d" \
  --header "anthropic-version: 2023-06-01" \
  --header "x-api-key: $ADMIN_API_KEY"

# モデル別の日次トークン消費量
curl "https://api.anthropic.com/v1/organizations/usage_report/messages?\
starting_at=2026-03-01T00:00:00Z&\
ending_at=2026-03-24T00:00:00Z&\
group_by[]=model&\
bucket_width=1d" \
  --header "anthropic-version: 2023-06-01" \
  --header "x-api-key: $ADMIN_API_KEY"

# Claude Opus 4.6のバッチ処理利用量を1時間単位で取得
curl "https://api.anthropic.com/v1/organizations/usage_report/messages?\
starting_at=2026-03-24T00:00:00Z&\
ending_at=2026-03-24T23:59:59Z&\
models[]=claude-opus-4-6&\
service_tiers[]=batch&\
context_window[]=0-200k&\
bucket_width=1h" \
  --header "anthropic-version: 2023-06-01" \
  --header "x-api-key: $ADMIN_API_KEY"

# 地域別・モデル別の利用量を取得
curl "https://api.anthropic.com/v1/organizations/usage_report/messages?\
starting_at=2026-03-01T00:00:00Z&\
ending_at=2026-03-24T00:00:00Z&\
group_by[]=inference_geo&\
group_by[]=model&\
bucket_width=1d" \
  --header "anthropic-version: 2023-06-01" \
  --header "x-api-key: $ADMIN_API_KEY"

# ワークスペース別・サービス別の月間コスト
curl "https://api.anthropic.com/v1/organizations/cost_report?\
starting_at=2026-03-01T00:00:00Z&\
ending_at=2026-03-31T00:00:00Z&\
group_by[]=workspace_id&\
group_by[]=description" \
  --header "anthropic-version: 2023-06-01" \
  --header "x-api-key: $ADMIN_API_KEY"

import requests
import json
from datetime import datetime, timedelta, timezone
 
# Admin APIキーを設定
ADMIN_API_KEY = "sk-ant-admin-your-key-here"
BASE_URL = "https://api.anthropic.com/v1/organizations"
HEADERS = {
    "anthropic-version": "2023-06-01",
    "x-api-key": ADMIN_API_KEY
}
 
def get_usage_report(days=7, group_by="model"):
    """直近N日間の利用量レポートを取得"""
    now = datetime.now(timezone.utc)
    start = now - timedelta(days=days)
 
    params = {
        "starting_at": start.strftime("%Y-%m-%dT00:00:00Z"),
        "ending_at": now.strftime("%Y-%m-%dT23:59:59Z"),
        "group_by[]": group_by,
        "bucket_width": "1d"
    }
 
    response = requests.get(
        f"{BASE_URL}/usage_report/messages",
        headers=HEADERS,
        params=params
    )
    response.raise_for_status()
    return response.json()
 
def get_cost_report(days=30):
    """直近N日間のコストレポートを取得"""
    now = datetime.now(timezone.utc)
    start = now - timedelta(days=days)
 
    params = {
        "starting_at": start.strftime("%Y-%m-%dT00:00:00Z"),
        "ending_at": now.strftime("%Y-%m-%dT23:59:59Z"),
        "group_by[]": ["workspace_id", "description"],
        "bucket_width": "1d"
    }
 
    response = requests.get(
        f"{BASE_URL}/cost_report",
        headers=HEADERS,
        params=params
    )
    response.raise_for_status()
    return response.json()
 
def print_usage_summary(data):
    """利用量サマリーを表示"""
    print("=" * 60)
    print("📊 Claude API 利用量サマリー")
    print("=" * 60)
 
    for bucket in data.get("data", []):
        date = bucket.get("bucket_start", "N/A")
        model = bucket.get("model", "all")
        input_tokens = bucket.get("input_tokens", 0)
        output_tokens = bucket.get("output_tokens", 0)
        cached_tokens = bucket.get("cache_read_input_tokens", 0)
 
        print(f"\n日付: {date[:10]} | モデル: {model}")
        print(f"  入力トークン: {input_tokens:,}")
        print(f"  出力トークン: {output_tokens:,}")
        print(f"  キャッシュ読取: {cached_tokens:,}")
 
# 実行例
# usage = get_usage_report(days=7, group_by="model")
# print_usage_summary(usage)
 
# 期待される出力:
# ============================================================
# 📊 Claude API 利用量サマリー
# ============================================================
#
# 日付: 2026-03-23 | モデル: claude-sonnet-4-6
#   入力トークン: 1,245,678
#   出力トークン: 523,456
#   キャッシュ読取: 890,123

def get_all_pages(endpoint, params):
    """ページネーションを処理して全データを取得"""
    all_data = []
 
    while True:
        response = requests.get(
            f"{BASE_URL}/{endpoint}",
            headers=HEADERS,
            params=params
        )
        response.raise_for_status()
        result = response.json()
 
        all_data.extend(result.get("data", []))
 
        # 次のページがあるかチェック
        if result.get("has_more", False):
            params["page"] = result["next_page"]
        else:
            break
 
    return all_data
 
# 使用例: 1ヶ月分のデータを全ページ取得
# all_usage = get_all_pages("usage_report/messages", {
#     "starting_at": "2026-03-01T00:00:00Z",
#     "ending_at": "2026-03-31T00:00:00Z",
#     "bucket_width": "1d",
#     "limit": 7
# })
# print(f"取得したバケット数: {len(all_usage)}")
 
# 期待される出力:
# 取得したバケット数: 24

def analyze_cache_efficiency(usage_data):
    """キャッシュ効率を分析して最適化の余地を見つける"""
    total_input = 0
    total_cached = 0
 
    for bucket in usage_data.get("data", []):
        total_input += bucket.get("input_tokens", 0)
        total_cached += bucket.get("cache_read_input_tokens", 0)
 
    if total_input > 0:
        cache_rate = (total_cached / total_input) * 100
        print(f"キャッシュヒット率: {cache_rate:.1f}%")
 
        if cache_rate < 30:
            print("⚠️ キャッシュ率が低めです。")
            print("  → システムプロンプトやツール定義の")
            print("    キャッシュ設定を見直しましょう。")
        elif cache_rate > 70:
            print("✅ キャッシュが効率的に機能しています。")
    else:
        print("データがありません。")
 
# 期待される出力:
# キャッシュヒット率: 45.2%
# ⚠️ キャッシュ率が低めです。
#   → システムプロンプトやツール定義の
#     キャッシュ設定を見直しましょう。