Anthropic API のベータヘッダを一元管理し、機能の引退でバッチを止めない設計

from enum import Enum
 
class Capability(Enum):
    PROMPT_CACHE = "prompt_cache"
    LONG_CONTEXT = "long_context"
    CONTEXT_EDIT = "context_edit"
    CODE_EXECUTION = "code_execution"
 
# 現在有効なベータ識別子。GA 済みは None(=ヘッダ不要)。
# 識別子は時期により変わるため、最新は公式の API リリースノートで確認します。
BETA_IDS: dict[Capability, str | None] = {
    Capability.PROMPT_CACHE: None,                       # GA 済み → ヘッダ不要
    Capability.LONG_CONTEXT:  "context-1m-2025-08-07",
    Capability.CONTEXT_EDIT:  "context-management-2025-06-27",
    Capability.CODE_EXECUTION: "code-execution-2025-05-22",
}
 
def beta_header(caps: set[Capability], enabled: set[Capability]) -> dict[str, str]:
    """要求された能力のうち、有効かつヘッダが必要なものだけを連結する。"""
    ids = sorted(
        BETA_IDS[c] for c in caps & enabled if BETA_IDS.get(c) is not None
    )
    return {"anthropic-beta": ",".join(ids)} if ids else {}

# After: 欲しい能力を宣言するだけ。識別子はレジストリが知っている。
headers = beta_header(
    caps={Capability.LONG_CONTEXT, Capability.PROMPT_CACHE},
    enabled=ENABLED,        # プリフライトで確定した有効集合(後述)
)
resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=4096,
    messages=messages,
    extra_headers=headers,
)

import anthropic, logging
 
client = anthropic.Anthropic()
log = logging.getLogger("beta_preflight")
 
def preflight(candidates: set[Capability]) -> set[Capability]:
    """各能力を個別に検証し、API が受理したものだけを返す。"""
    enabled: set[Capability] = set()
    for c in candidates:
        if BETA_IDS.get(c) is None:   # GA 済みは検証不要
            enabled.add(c)
            continue
        try:
            client.messages.count_tokens(
                model="claude-opus-4-8",
                messages=[{"role": "user", "content": "preflight"}],
                extra_headers={"anthropic-beta": BETA_IDS[c]},
            )
            enabled.add(c)
        except anthropic.BadRequestError as e:
            # 未知・引退済みベータは 400。その能力だけ落とし、処理は続行。
            if "beta" in str(e).lower():
                log.warning("capability disabled: %s (%s)", c.value, e)
            else:
                raise   # ベータと無関係な 400 は握りつぶさない
    return enabled
 
ENABLED = preflight({
    Capability.LONG_CONTEXT,
    Capability.CONTEXT_EDIT,
    Capability.PROMPT_CACHE,
})

def generate(messages, caps: set[Capability]):
    headers = beta_header(caps, ENABLED)
    try:
        return client.messages.create(
            model="claude-opus-4-8", max_tokens=4096,
            messages=messages, extra_headers=headers,
        )
    except anthropic.BadRequestError as e:
        if "beta" not in str(e).lower():
            raise
        # 稼働中の無効化を検知。能力を落として再構成する。
        log.warning("runtime beta rejection, degrading: %s", e)
        if Capability.LONG_CONTEXT in caps:
            messages = chunk_and_summarize(messages)   # 第2段階: 機能代替
        return client.messages.create(                 # 第1段階: ヘッダを外して再送
            model="claude-opus-4-8", max_tokens=4096,
            messages=messages, extra_headers=beta_header(
                caps - {Capability.LONG_CONTEXT}, ENABLED),
        )

# レジストリは出力先を選ばない。betas 引数へ流す版。
def beta_list(caps: set[Capability], enabled: set[Capability]) -> list[str]:
    return sorted(BETA_IDS[c] for c in caps & enabled if BETA_IDS.get(c))
 
resp = client.beta.messages.create(
    model="claude-opus-4-8", max_tokens=4096, messages=messages,
    betas=beta_list({Capability.CONTEXT_EDIT}, ENABLED),
)

import json, datetime
 
def log_capability_state(enabled: set[Capability], reason: str = "preflight"):
    record = {
        "ts": datetime.datetime.now(datetime.UTC).isoformat(),
        "reason": reason,
        "enabled": sorted(c.value for c in enabled),
        "disabled": sorted(c.value for c in set(Capability) - enabled),
    }
    log.info("capability_state %s", json.dumps(record, ensure_ascii=False))