get the rate card for models on amazon bedrock

bedrock_pricing.py

"""Fetch and display on-demand pricing for Claude models on Amazon Bedrock.

Uses the Bedrock management API to list available Anthropic Claude models
and retrieve their per-token pricing from agreement offers.

Note: you'll need boto3 for this
"""

import argparse
import json
import sys

import boto3
from botocore.exceptions import ClientError


def list_claude_models(session, region):
    """Return Claude model summaries from the Bedrock catalog."""
    client = session.client("bedrock", region_name=region)
    resp = client.list_foundation_models(byProvider="Anthropic")
    return resp.get("modelSummaries", [])


def get_model_pricing(session, region, model_id):
    """Fetch pricing for a single model via ListFoundationModelAgreementOffers.

    Returns (pricing_dict, source_string).
    """
    client = session.client("bedrock", region_name=region)
    resp = client.list_foundation_model_agreement_offers(
        modelId=model_id,
        offerType="ALL",
    )

    pricing = {}
    source = None
    for offer in resp.get("offers", []):
        if source is None:
            source = offer.get("offerType", "Bedrock Agreement Offers API")
        term = offer.get("termDetails", {}).get("usageBasedPricingTerm", {})
        for entry in term.get("rateCard", []):
            dim = entry.get("dimension", "").lower()
            price = float(entry.get("price", "0"))
            unit = entry.get("unit", "")
            desc = entry.get("description", "")
            pricing[dim] = {"price": price, "unit": unit, "description": desc}

    return pricing, source or "Bedrock Agreement Offers API"


def classify_dimensions(pricing):
    """Normalize rate-card dimensions into standard categories."""
    result = {}
    for dim, info in pricing.items():
        price = info["price"]
        desc = info.get("description", "").lower()
        key = dim + " " + desc

        if "cache" in key and "read" in key:
            result["cache_read"] = price
        elif "cache" in key and "write" in key:
            result["cache_write"] = price
        elif "output" in key:
            result["output"] = price
        elif "input" in key:
            result["input"] = price

    return result


def matches_filter(model_id, model_name, families):
    """Check if a model matches any of the filter substrings."""
    text = f"{model_id} {model_name}".lower()
    return any(f.lower() in text for f in families)


def fmt(price):
    """Format a price as a dollar string."""
    if price is None:
        return "-"
    if price < 0.01:
        return f"${price:.4f}"
    return f"${price:.2f}"


def print_table(rows):
    """Print the pricing table."""
    if not rows:
        print("No models found matching the filter.")
        return

    hdr = (
        f"  {'Model':<22} {'Bedrock ID':<48}"
        f" {'Input/MTok':>11} {'Output/MTok':>12} {'Cache Read':>11} {'Cache Write':>12}"
        f"  {'Source'}"
    )
    print(f"\n{hdr}")
    print("  " + "-" * 145)

    for name, model_id, status, prices, source in rows:
        tag = " (legacy)" if status == "LEGACY" else ""
        name_col = f"{name}{tag}"
        if prices:
            print(
                f"  {name_col:<22} {model_id:<48}"
                f" {fmt(prices.get('input')):>11}"
                f" {fmt(prices.get('output')):>12}"
                f" {fmt(prices.get('cache_read')):>11}"
                f" {fmt(prices.get('cache_write')):>12}"
                f"  {source}"
            )
        else:
            print(f"  {name_col:<22} {model_id:<48}  {'(pricing not available)':<60} {source}")

    print(f"\n  Prices are USD per million tokens.")
    print()


def main():
    parser = argparse.ArgumentParser(
        description="Display on-demand pricing for Claude models on Amazon Bedrock.",
    )
    parser.add_argument(
        "--family",
        nargs="*",
        metavar="SUBSTR",
        help=(
            "Filter by substring on model ID or name (e.g. opus-4-5 sonnet-4-5). "
            "Default: Claude 4 & 4.5 families."
        ),
    )
    parser.add_argument("--all", action="store_true", help="Show all Claude models")
    parser.add_argument("--profile", help="AWS CLI profile name")
    parser.add_argument(
        "--region",
        default="us-east-1",
        help="AWS region (default: us-east-1)",
    )
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--raw", action="store_true", help="Dump raw rate cards and exit")

    args = parser.parse_args()
    session = boto3.Session(profile_name=args.profile)

    # List available Claude models
    print("Querying Bedrock model catalog ...", file=sys.stderr)
    try:
        catalog = list_claude_models(session, args.region)
    except ClientError as e:
        print(f"Error listing models: {e}", file=sys.stderr)
        sys.exit(1)

    # Deduplicate (some models appear with context-window suffixes like :0:200k)
    seen = set()
    models = []
    for m in catalog:
        base_id = m["modelId"].split(":")[0]
        if base_id in seen:
            continue
        seen.add(base_id)
        models.append(m)

    # Apply filter
    if args.family:
        models = [m for m in models if matches_filter(m["modelId"], m.get("modelName", ""), args.family)]
    elif not args.all:
        default_families = ["opus-4", "sonnet-4", "haiku-4"]
        models = [m for m in models if matches_filter(m["modelId"], m.get("modelName", ""), default_families)]

    # Fetch pricing for each model
    rows = []
    raw_data = []
    for m in models:
        model_id = m["modelId"]
        name = m.get("modelName", "")
        status = m.get("modelLifecycle", {}).get("status", "")

        print(f"  Fetching pricing for {name} ...", file=sys.stderr)
        try:
            raw_pricing, source = get_model_pricing(session, args.region, model_id)
            prices = classify_dimensions(raw_pricing)
        except ClientError as e:
            code = e.response.get("Error", {}).get("Code", "")
            source = "-"
            if code in ("ResourceNotFoundException", "ValidationException"):
                prices = None
                raw_pricing = {}
            else:
                print(f"Error fetching pricing for {model_id}: {e}", file=sys.stderr)
                prices = None
                raw_pricing = {}

        rows.append((name, model_id, status, prices, source))
        if args.raw:
            raw_data.append({"model": name, "model_id": model_id, "source": source, "rate_card": raw_pricing})

    # Sort: active first, then by name
    rows.sort(key=lambda r: (0 if r[2] == "ACTIVE" else 1, r[0]))

    if args.raw:
        print(json.dumps(raw_data, indent=2, default=str))
    elif args.json:
        output = []
        for name, model_id, status, prices, source in rows:
            entry = {"model": name, "bedrock_id": model_id, "status": status, "source": source}
            if prices:
                entry["pricing_per_mtok"] = prices
            output.append(entry)
        print(json.dumps(output, indent=2))
    else:
        print_table(rows)


if __name__ == "__main__":
    main()