Vitals API Reference

# Values that you need to provide
client_id = "<your client id>"
client_secret = "<your client secret>"
package_name = "<your app package name>"

BASE_URL = "https://developer.amazon.com/api/appstore"

scope = "adx_reporting::appstore:marketer"
data = {
    "grant_type": "client_credentials",
    "client_id": client_id,
    "client_secret": client_secret,
    "scope": scope
}
auth_response = requests.post("https://api.amazon.com/auth/o2/token", data=data)
auth_token = auth_response.json()["access_token"]

headers = {
    "Authorization": f"Bearer {auth_token}",
    "Content-Type": "application/json"
}

freshness_url = f"{BASE_URL}/vitals/apps/{package_name}/crashMetricSet"
freshness_response = requests.get(freshness_url, headers=headers)
freshness = freshness_response.json()

for info in freshness["freshnessInfo"]:
    print(f"{info['aggregationPeriod']}: {info['latestDataTimestamp']}")

from datetime import datetime, timezone, timedelta


def utc_date_components(local_dt=None):
    """Convert a datetime to UTC year/month/day for the API.

    A naive local_dt is interpreted as system local time.
    Pass a tz-aware datetime to avoid ambiguity.
    """
    if local_dt is None:
        utc_dt = datetime.now(timezone.utc)
    else:
        if local_dt.tzinfo is None:
            local_dt = local_dt.astimezone()  # attach system local tz explicitly
        utc_dt = local_dt.astimezone(timezone.utc)
    return {"year": utc_dt.year, "month": utc_dt.month, "day": utc_dt.day}


# Example: "yesterday" in UTC (regardless of your local timezone)
yesterday_utc = datetime.now(timezone.utc) - timedelta(days=1)
start = utc_date_components(yesterday_utc)

metric_set = "crashMetricSet"   # or "anrMetricSet", "lmkMetricSet"
query_url = f"{BASE_URL}/vitals/apps/{package_name}/{metric_set}:query"

body = {
    # Required: time range and granularity
    "timelineSpec": {
        "aggregationPeriod": "DAILY", # DAILY / HOURLY
        "startTime": {"year": 2026, "month": 4, "day": 01},
        "endTime":   {"year": 2026, "month": 4, "day": 01},
    },

    # Optional: metrics from the metricset being queried
    "metrics": ["crashRate", "crashCount", "distinctDevices"],

    # Optional: break down by dimension (max 5)
    "dimensions": ["versionCode", "countryCode"],

    # Optional: AIP-160 filter expression
    "filter": {"countryCode": ["US", "CA"], "deviceType": ["AMAZON_FIRE_TV"]},

    # Optional: paging (default 1000, max 100000)
    "pageSize": 1000,
    "pageToken": "<from previous response>",
}

response = requests.post(query_url, headers=headers, json=body)
result = response.json()

all_rows = []
body.pop("pageToken", None)
while True:
    result = requests.post(query_url, headers=headers, json=body).json()
    all_rows.extend(result["rows"])
    token = result.get("nextPageToken")
    if not token:
        break
    body["pageToken"] = token

issues_url = f"{BASE_URL}/vitals/apps/{package_name}/issuesMetricSet:query"
body = {
  "startTime": { "year": 2025, "month": 6, "day": 1 },
  "endTime": { "year": 2025, "month": 6, "day": 15 },
  "reportType": "CRASH",
  "dimensions": ["deviceType", "osVersion"],
  "filter": { "countryCode": ["US", "CA"] },
  "pageSize": 25,
  "pageToken": null
}
response = requests.post(issues_url, headers=headers, json=body)
result = response.json()

# Response shape (different from rate metric sets):
# {
#   "rows": [
#     {
#       "crashDescriptor": "67f35eb2...",
#       "reportText": "java.lang.IllegalStateException: ...",
#       "metrics": [
#         {"name": "errorEventCount", "value": "78"},
#         {"name": "affectedDeviceCount", "value": "45"}
#       ]
#     }
#   ],
#   "nextPageToken": "..."
# }

import time

MAX_RETRIES = 3
RETRIABLE_CODES = {429, 500, 502, 503}

def query_with_retry(client, package, metric_set, body):
    for attempt in range(MAX_RETRIES):
        resp = client.query(package, metric_set, body)

        if resp.status_code == 200:
            return resp.json()

        if resp.status_code not in RETRIABLE_CODES:
            break  # Non-retriable error

        wait = (2 ** attempt) + 0.5  # 1.5s, 2.5s, 4.5s
        print(f"Retry {attempt + 1}/{MAX_RETRIES} after {wait}s (HTTP {resp.status_code})")
        time.sleep(wait)

    # Final attempt failed
    error = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
    raise RuntimeError(f"Failed after {MAX_RETRIES} retries: {resp.status_code} {error.get('message', '')}")