max_tokens is set above the model's own output cap

"""Compare each configured max_tokens against the model's own published cap.

Read only. GET requests and nothing else: give this a workspace API key. No
payload is ever sent, no tokens are counted, and /v1/messages is never called.
The repair is printed, because choosing an output budget is a judgement about
your product and not a side effect of an audit.
"""
import argparse
import json
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("anthropic_max_tokens_cap")

API = "https://api.anthropic.com/v1"
VERSION = "2023-06-01"

# The Batch API raises the output ceiling on the 1M-context models, and only
# behind this header. A batch path that does not send it is capped exactly as a
# synchronous one, which is why the header is an input to the check rather than
# an assumption.
BATCH_300K_BETA = "output-300k-2026-03-24"
BATCH_MAX_TOKENS = 300_000
LONG_CONTEXT_WINDOW = 1_000_000

FINDINGS = ("above-cap", "below-minimum", "cap-unknown", "model-not-found")


def parse_path(spec):
    """Read a NAME=MODEL:MAX_TOKENS argument. Pure. (name, entry) or None."""
    text = str(spec or "").strip()
    if "=" not in text:
        return None
    name, rest = text.split("=", 1)
    if ":" not in rest:
        return None
    model, value = rest.rsplit(":", 1)
    try:
        configured = int(value)
    except (TypeError, ValueError):
        return None
    name, model = name.strip(), model.strip()
    if not name or not model:
        return None
    return (name, {"model": model, "max_tokens": configured, "endpoint": "messages"})


def sync_cap(model_obj):
    """The model object's own max_tokens field. Pure. None if absent.

    This is the source of truth. The published table lags a release and a
    constant in your source lags the table, so a missing field is reported as
    missing rather than filled in from either.
    """
    if not isinstance(model_obj, dict):
        return None
    value = model_obj.get("max_tokens")
    return value if isinstance(value, int) and value > 0 else None


def window_of(model_obj):
    """max_input_tokens off a model object. Pure. Used only to size the batch
    ceiling, which applies to the 1M-context models."""
    if not isinstance(model_obj, dict):
        return None
    value = model_obj.get("max_input_tokens")
    return value if isinstance(value, int) and value > 0 else None


def effective_cap(model_obj, endpoint="messages", betas=()):
    """The legal ceiling for max_tokens on one model at one endpoint. Pure.

    Two inputs, because the ceiling belongs to the pair and not to the model.
    A batch path with the output-300k header on a 1M-context model gets the
    higher number; the same path without the header does not, and neither does
    a 200k-context model that has it.
    """
    cap = sync_cap(model_obj)
    if cap is None:
        return (None, "the model object carried no max_tokens field")
    if str(endpoint) == "batches" and BATCH_300K_BETA in set(betas or ()):
        window = window_of(model_obj)
        if window is not None and window >= LONG_CONTEXT_WINDOW:
            return (BATCH_MAX_TOKENS, "the Batch API with " + BATCH_300K_BETA)
        return (cap, "the model object; the 300K batch ceiling needs a "
                     "1M context model")
    return (cap, "the model object")


def verdict(configured, cap):
    """Classify one configured value against one cap. Pure. (state, detail)."""
    configured = int(configured or 0)
    if configured < 1:
        return ("below-minimum",
                "max_tokens is %d, and the minimum accepted value is 1"
                % configured)
    if cap is None:
        return ("cap-unknown",
                "max_tokens is %d and no ceiling could be read for this model "
                "and endpoint" % configured)
    if configured > cap:
        return ("above-cap",
                "max_tokens is %d against a cap of %d, which is a 400 "
                "invalid_request_error on every call, %d over"
                % (configured, cap, configured - cap))
    if configured == cap:
        return ("at-cap",
                "max_tokens is %d, exactly the cap, so any move to a smaller "
                "model breaks this path" % configured)
    return ("within-cap",
            "max_tokens is %d of a %d cap (%.0f%%)"
            % (configured, cap, configured / float(cap) * 100))


def tier_spans(rows):
    """One configured value reused across models with different ceilings. Pure.

    rows: [(name, model_id, configured, cap)]. Returns [(value, [model ids])].

    The number appears once in the source, so nothing at any call site says
    that its effective ceiling is the smallest cap among the models using it.
    That is the finding even on the day every path still passes, because the
    next model swap is the one that turns it into a 400.
    """
    by_value = {}
    for name, model, configured, cap in rows or []:
        by_value.setdefault(int(configured or 0), []).append((name, model, cap))
    out = []
    for value in sorted(by_value):
        entries = by_value[value]
        models = sorted({m for _n, m, _c in entries})
        if len(models) < 2:
            continue
        out.append((value, models))
    return out


def get_model(session, model_id):
    """One GET per distinct model id. A 404 here belongs to a different note."""
    r = session.get(API + "/models/" + str(model_id), timeout=30)
    if r.status_code == 404:
        return None
    if r.status_code in (401, 403):
        raise SystemExit("%d from Anthropic: ANTHROPIC_API_KEY has to be a "
                         "workspace key" % r.status_code)
    r.raise_for_status()
    return r.json()


def load_config(path):
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--config", help="JSON file of call paths: "
                                     '{"name": {"model": ..., "max_tokens": ..., '
                                     '"endpoint": "messages|batches", "betas": []}}')
    ap.add_argument("--path", action="append", default=[], metavar="NAME=MODEL:MAX",
                    help="one call path in shorthand, repeatable")
    ap.add_argument("--show-all", action="store_true",
                    help="also print paths comfortably under their cap")
    args = ap.parse_args()

    key = os.environ.get("ANTHROPIC_API_KEY")
    if not key:
        log.error("set ANTHROPIC_API_KEY to a workspace key")
        return 2

    paths = dict(load_config(args.config)) if args.config else {}
    for spec in args.path:
        parsed = parse_path(spec)
        if parsed is None:
            log.error("cannot read --path %r, expected NAME=MODEL:MAX_TOKENS", spec)
            return 2
        paths[parsed[0]] = parsed[1]
    if not paths:
        log.error("give --config FILE or at least one --path NAME=MODEL:MAX_TOKENS")
        return 2

    session = requests.Session()
    session.headers.update({"x-api-key": key, "anthropic-version": VERSION})

    models = {}
    rows = []
    bad = 0

    for name in sorted(paths):
        entry = paths[name] or {}
        model_id = str(entry.get("model") or "")
        configured = entry.get("max_tokens")
        endpoint = entry.get("endpoint") or "messages"
        betas = entry.get("betas") or []

        if model_id not in models:
            models[model_id] = get_model(session, model_id)
        model_obj = models[model_id]
        if model_obj is None:
            bad += 1
            log.warning("%-14s %-16s %-28s the model id is not in the live list "
                        "at all, which is a retirement or a typo rather than a "
                        "max_tokens problem", "model-not-found", name, model_id)
            continue

        cap, source = effective_cap(model_obj, endpoint, betas)
        state, detail = verdict(configured, cap)
        rows.append((name, model_id, int(configured or 0), cap))

        line = "%-14s %-16s %-28s %s" % (state, name, model_id, detail)
        if state in FINDINGS:
            bad += 1
            log.warning(line)
            log.warning("  ceiling read from %s", source)
        elif state == "at-cap":
            log.warning(line)
        elif args.show_all:
            log.info(line)

    for value, shared in tier_spans(rows):
        caps = [cap for _n, _m, configured, cap in rows
                if configured == value and cap is not None]
        note = "shared value %d is configured on %d model(s): %s" % (
            value, len(shared), ", ".join(shared))
        if caps and min(caps) < value:
            bad += 1
            log.warning("%-14s %s, and the smallest cap among them is %d",
                        "spans-tiers", note, min(caps))
        else:
            log.info("  %s, so the effective ceiling is the smallest of their "
                     "caps whether or not anything says so", note)

    if bad:
        log.warning("  repair: set each path's max_tokens from the cap the "
                    "Models API reports for its own model, not from a shared "
                    "constant and not from the docs table, which lags. Note "
                    "that maxing it out trades a 400 for truncated answers and "
                    "long non-streaming requests. Printed, not applied.")

    log.info("%d path(s) checked, %d finding(s)", len(paths), bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Compare each configured max_tokens against the model's own published cap.
 *
 * Read only. GET requests and nothing else: give this a workspace API key. No
 * payload is ever sent, no tokens are counted, and /v1/messages is never
 * called. The repair is printed.
 */
import { readFile } from 'node:fs/promises';

const API = 'https://api.anthropic.com/v1';
const VERSION = '2023-06-01';

const BATCH_300K_BETA = 'output-300k-2026-03-24';
const BATCH_MAX_TOKENS = 300000;
const LONG_CONTEXT_WINDOW = 1000000;

const FINDINGS = new Set(['above-cap', 'below-minimum', 'cap-unknown', 'model-not-found']);

/** Read a NAME=MODEL:MAX_TOKENS argument. Pure. [name, entry] or null. */
export function parsePath(spec) {
  const text = String(spec ?? '').trim();
  const eq = text.indexOf('=');
  if (eq < 0) return null;
  const name = text.slice(0, eq).trim();
  const rest = text.slice(eq + 1);
  const colon = rest.lastIndexOf(':');
  if (colon < 0) return null;
  const model = rest.slice(0, colon).trim();
  const value = rest.slice(colon + 1).trim();
  if (!name || !model || !/^-?[0-9]+$/.test(value)) return null;
  return [name, { model, max_tokens: Number(value), endpoint: 'messages' }];
}

/** The model object's own max_tokens field. Pure. null if absent. */
export function syncCap(modelObj) {
  if (!modelObj || typeof modelObj !== 'object') return null;
  const value = modelObj.max_tokens;
  return Number.isInteger(value) && value > 0 ? value : null;
}

/** max_input_tokens off a model object. Pure. Sizes the batch ceiling only. */
export function windowOf(modelObj) {
  if (!modelObj || typeof modelObj !== 'object') return null;
  const value = modelObj.max_input_tokens;
  return Number.isInteger(value) && value > 0 ? value : null;
}

/**
 * The legal ceiling for max_tokens on one model at one endpoint. Pure.
 * The ceiling belongs to the pair: a batch path with the output-300k header on
 * a 1M-context model gets the higher number and nothing else does.
 */
export function effectiveCap(modelObj, endpoint = 'messages', betas = []) {
  const cap = syncCap(modelObj);
  if (cap === null) return [null, 'the model object carried no max_tokens field'];
  if (String(endpoint) === 'batches' && new Set(betas ?? []).has(BATCH_300K_BETA)) {
    const window = windowOf(modelObj);
    if (window !== null && window >= LONG_CONTEXT_WINDOW) {
      return [BATCH_MAX_TOKENS, `the Batch API with ${BATCH_300K_BETA}`];
    }
    return [cap, 'the model object; the 300K batch ceiling needs a 1M context model'];
  }
  return [cap, 'the model object'];
}

/** Classify one configured value against one cap. Pure. [state, detail]. */
export function verdict(configured, cap) {
  const value = Math.trunc(configured || 0);
  if (value < 1) {
    return ['below-minimum',
      `max_tokens is ${value}, and the minimum accepted value is 1`];
  }
  if (cap === null || cap === undefined) {
    return ['cap-unknown',
      `max_tokens is ${value} and no ceiling could be read for this model and endpoint`];
  }
  if (value > cap) {
    return ['above-cap',
      `max_tokens is ${value} against a cap of ${cap}, which is a 400 ` +
      `invalid_request_error on every call, ${value - cap} over`];
  }
  if (value === cap) {
    return ['at-cap',
      `max_tokens is ${value}, exactly the cap, so any move to a smaller model ` +
      'breaks this path'];
  }
  return ['within-cap',
    `max_tokens is ${value} of a ${cap} cap (${(value / cap * 100).toFixed(0)}%)`];
}

/**
 * One configured value reused across models with different ceilings. Pure.
 * rows: [[name, modelId, configured, cap]]. Returns [[value, [modelIds]]].
 */
export function tierSpans(rows) {
  const byValue = new Map();
  for (const [name, model, configured, cap] of rows ?? []) {
    const value = Math.trunc(configured || 0);
    if (!byValue.has(value)) byValue.set(value, []);
    byValue.get(value).push([name, model, cap]);
  }
  const out = [];
  for (const value of [...byValue.keys()].sort((a, b) => a - b)) {
    const models = [...new Set(byValue.get(value).map(([, m]) => m))].sort();
    if (models.length < 2) continue;
    out.push([value, models]);
  }
  return out;
}

async function getModel(key, modelId) {
  const res = await fetch(`${API}/models/${modelId}`, {
    headers: { 'x-api-key': key, 'anthropic-version': VERSION },
  });
  if (res.status === 404) return null;
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Anthropic: ANTHROPIC_API_KEY has to be a workspace key`);
  }
  if (!res.ok) throw new Error(`${res.status} from /models/${modelId}`);
  return res.json();
}

async function main() {
  const key = process.env.ANTHROPIC_API_KEY;
  if (!key) {
    console.error('set ANTHROPIC_API_KEY to a workspace key');
    process.exitCode = 2;
    return;
  }
  const paths = {};
  if (process.env.CONFIG) Object.assign(paths, JSON.parse(await readFile(process.env.CONFIG, 'utf8')));
  for (const spec of process.argv.slice(2).filter((a) => !a.startsWith('--'))) {
    const parsed = parsePath(spec);
    if (!parsed) {
      console.error(`cannot read '${spec}', expected NAME=MODEL:MAX_TOKENS`);
      process.exitCode = 2;
      return;
    }
    paths[parsed[0]] = parsed[1];
  }
  if (Object.keys(paths).length === 0) {
    console.error('set CONFIG to a JSON file, or pass NAME=MODEL:MAX_TOKENS arguments');
    process.exitCode = 2;
    return;
  }
  const showAll = process.env.SHOW_ALL === '1';

  const models = new Map();
  const rows = [];
  let bad = 0;

  for (const name of Object.keys(paths).sort()) {
    const entry = paths[name] ?? {};
    const modelId = String(entry.model ?? '');
    const endpoint = entry.endpoint ?? 'messages';
    const betas = entry.betas ?? [];

    if (!models.has(modelId)) models.set(modelId, await getModel(key, modelId));
    const modelObj = models.get(modelId);
    if (modelObj === null) {
      bad += 1;
      console.warn(`${'model-not-found'.padEnd(14)} ${name.padEnd(16)} ` +
                   `${modelId.padEnd(28)} the model id is not in the live list at ` +
                   'all, which is a retirement or a typo rather than a max_tokens problem');
      continue;
    }

    const [cap, source] = effectiveCap(modelObj, endpoint, betas);
    const [state, detail] = verdict(entry.max_tokens, cap);
    rows.push([name, modelId, Math.trunc(entry.max_tokens || 0), cap]);

    const line = `${state.padEnd(14)} ${name.padEnd(16)} ${modelId.padEnd(28)} ${detail}`;
    if (FINDINGS.has(state)) {
      bad += 1;
      console.warn(line);
      console.warn(`  ceiling read from ${source}`);
    } else if (state === 'at-cap') {
      console.warn(line);
    } else if (showAll) {
      console.log(line);
    }
  }

  for (const [value, shared] of tierSpans(rows)) {
    const caps = rows.filter(([, , configured, cap]) => configured === value && cap !== null)
      .map(([, , , cap]) => cap);
    const note = `shared value ${value} is configured on ${shared.length} model(s): ` +
                 shared.join(', ');
    if (caps.length && Math.min(...caps) < value) {
      bad += 1;
      console.warn(`${'spans-tiers'.padEnd(14)} ${note}, and the smallest cap ` +
                   `among them is ${Math.min(...caps)}`);
    } else {
      console.log(`  ${note}, so the effective ceiling is the smallest of their ` +
                  'caps whether or not anything says so');
    }
  }

  if (bad) {
    console.warn('  repair: set each path\'s max_tokens from the cap the Models API ' +
                 'reports for its own model, not from a shared constant and not from ' +
                 'the docs table, which lags. Note that maxing it out trades a 400 for ' +
                 'truncated answers and long non-streaming requests. Printed, not applied.');
  }

  console.log(`${Object.keys(paths).length} path(s) checked, ${bad} finding(s)`);
  process.exitCode = bad ? 1 : 0;
}

if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from anthropic_max_tokens_cap import (effective_cap, parse_path, sync_cap,
                                       tier_spans, verdict, window_of)

SONNET = {"id": "claude-sonnet-5", "max_tokens": 128_000,
          "max_input_tokens": 1_000_000}
HAIKU = {"id": "claude-haiku-4-5-20251001", "max_tokens": 64_000,
         "max_input_tokens": 200_000}


def test_the_same_value_is_legal_on_one_model_and_a_400_on_the_other():
    # The whole note. One shared constant, two tiers, one of them rejected on
    # every call from the first one.
    assert verdict(128_000, effective_cap(SONNET)[0])[0] == "at-cap"
    state, detail = verdict(128_000, effective_cap(HAIKU)[0])
    assert state == "above-cap"
    assert "against a cap of 64000" in detail
    assert "64000 over" in detail
    assert "400" in detail


def test_the_batch_ceiling_needs_the_endpoint_and_the_header_and_the_window():
    # Three inputs, and dropping any one of them gives the wrong ceiling.
    cap, source = effective_cap(SONNET, "batches", ["output-300k-2026-03-24"])
    assert (cap, "output-300k-2026-03-24" in source) == (300_000, True)
    # Same model, same header, synchronous endpoint: the model object wins.
    assert effective_cap(SONNET, "messages", ["output-300k-2026-03-24"])[0] == 128_000
    # Same model, batch endpoint, header not sent: the model object again.
    assert effective_cap(SONNET, "batches", [])[0] == 128_000
    # Header sent on a 200k-context model: it does not qualify.
    cap, source = effective_cap(HAIKU, "batches", ["output-300k-2026-03-24"])
    assert cap == 64_000
    assert "1M context model" in source


def test_a_model_object_with_no_cap_is_not_an_unlimited_one():
    assert sync_cap({"id": "claude-sonnet-5"}) is None
    assert sync_cap({"max_tokens": 0}) is None
    assert sync_cap({"max_tokens": "128000"}) is None
    assert sync_cap(None) is None
    assert window_of(HAIKU) == 200_000
    assert window_of({}) is None
    state, detail = verdict(128_000, effective_cap({"id": "x"})[0])
    assert state == "cap-unknown"
    assert "no ceiling could be read" in detail


def test_the_floor_is_one_and_it_is_a_different_finding():
    assert verdict(0, 128_000)[0] == "below-minimum"
    assert verdict(-1, 128_000)[0] == "below-minimum"
    assert verdict(1, 128_000)[0] == "within-cap"


def test_a_value_sitting_exactly_on_the_ceiling_is_its_own_warning():
    state, detail = verdict(64_000, 64_000)
    assert state == "at-cap"
    assert "any move to a smaller model breaks this path" in detail
    assert verdict(16_000, 64_000) == (
        "within-cap", "max_tokens is 16000 of a 64000 cap (25%)")


def test_one_number_shared_across_two_tiers_is_reported_before_it_breaks():
    rows = [("reports", "claude-opus-5", 64_000, 128_000),
            ("classifier", "claude-haiku-4-5-20251001", 64_000, 64_000),
            ("summaries", "claude-sonnet-5", 8_000, 128_000)]
    # 64000 passes on both today, and it is still the number the next model
    # swap turns into a 400, so it is named.
    assert tier_spans(rows) == [(64_000, ["claude-haiku-4-5-20251001",
                                          "claude-opus-5"])]
    # A value used by one model only is not a span.
    assert tier_spans(rows[2:]) == []
    assert tier_spans([]) == []
    assert tier_spans(None) == []


def test_the_shorthand_argument_parses_model_ids_that_contain_no_colon():
    assert parse_path("classifier=claude-haiku-4-5-20251001:64000") == (
        "classifier", {"model": "claude-haiku-4-5-20251001",
                       "max_tokens": 64000, "endpoint": "messages"})
    assert parse_path("reports=claude-opus-5:128000")[1]["max_tokens"] == 128000
    assert parse_path("no-colon=claude-opus-5") is None
    assert parse_path("claude-opus-5:128000") is None
    assert parse_path("reports=claude-opus-5:lots") is None
    assert parse_path("") is None
    assert parse_path(None) is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { effectiveCap, parsePath, syncCap, tierSpans, verdict, windowOf }
  from './anthropic-max-tokens-cap.mjs';

const SONNET = { id: 'claude-sonnet-5', max_tokens: 128000, max_input_tokens: 1000000 };
const HAIKU = { id: 'claude-haiku-4-5-20251001', max_tokens: 64000, max_input_tokens: 200000 };

test('the same value is legal on one model and a 400 on the other', () => {
  assert.equal(verdict(128000, effectiveCap(SONNET)[0])[0], 'at-cap');
  const [state, detail] = verdict(128000, effectiveCap(HAIKU)[0]);
  assert.equal(state, 'above-cap');
  assert.match(detail, /against a cap of 64000/);
  assert.match(detail, /64000 over/);
  assert.match(detail, /400/);
});

test('the batch ceiling needs the endpoint and the header and the window', () => {
  const [cap, source] = effectiveCap(SONNET, 'batches', ['output-300k-2026-03-24']);
  assert.equal(cap, 300000);
  assert.match(source, /output-300k-2026-03-24/);
  assert.equal(effectiveCap(SONNET, 'messages', ['output-300k-2026-03-24'])[0], 128000);
  assert.equal(effectiveCap(SONNET, 'batches', [])[0], 128000);
  const [haikuCap, haikuSource] = effectiveCap(HAIKU, 'batches', ['output-300k-2026-03-24']);
  assert.equal(haikuCap, 64000);
  assert.match(haikuSource, /1M context model/);
});

test('a model object with no cap is not an unlimited one', () => {
  assert.equal(syncCap({ id: 'claude-sonnet-5' }), null);
  assert.equal(syncCap({ max_tokens: 0 }), null);
  assert.equal(syncCap({ max_tokens: '128000' }), null);
  assert.equal(syncCap(null), null);
  assert.equal(windowOf(HAIKU), 200000);
  assert.equal(windowOf({}), null);
  const [state, detail] = verdict(128000, effectiveCap({ id: 'x' })[0]);
  assert.equal(state, 'cap-unknown');
  assert.match(detail, /no ceiling could be read/);
});

test('the floor is one and it is a different finding', () => {
  assert.equal(verdict(0, 128000)[0], 'below-minimum');
  assert.equal(verdict(-1, 128000)[0], 'below-minimum');
  assert.equal(verdict(1, 128000)[0], 'within-cap');
});

test('a value sitting exactly on the ceiling is its own warning', () => {
  const [state, detail] = verdict(64000, 64000);
  assert.equal(state, 'at-cap');
  assert.match(detail, /any move to a smaller model breaks this path/);
  assert.deepEqual(verdict(16000, 64000),
    ['within-cap', 'max_tokens is 16000 of a 64000 cap (25%)']);
});

test('one number shared across two tiers is reported before it breaks', () => {
  const rows = [['reports', 'claude-opus-5', 64000, 128000],
                ['classifier', 'claude-haiku-4-5-20251001', 64000, 64000],
                ['summaries', 'claude-sonnet-5', 8000, 128000]];
  assert.deepEqual(tierSpans(rows),
    [[64000, ['claude-haiku-4-5-20251001', 'claude-opus-5']]]);
  assert.deepEqual(tierSpans(rows.slice(2)), []);
  assert.deepEqual(tierSpans([]), []);
  assert.deepEqual(tierSpans(null), []);
});

test('the shorthand argument parses model ids that contain no colon', () => {
  assert.deepEqual(parsePath('classifier=claude-haiku-4-5-20251001:64000'),
    ['classifier', { model: 'claude-haiku-4-5-20251001', max_tokens: 64000,
                     endpoint: 'messages' }]);
  assert.equal(parsePath('reports=claude-opus-5:128000')[1].max_tokens, 128000);
  assert.equal(parsePath('no-colon=claude-opus-5'), null);
  assert.equal(parsePath('claude-opus-5:128000'), null);
  assert.equal(parsePath('reports=claude-opus-5:lots'), null);
  assert.equal(parsePath(''), null);
  assert.equal(parsePath(null), null);
});