この記事は Zenn に公開した記事の再掲です（原文・最新版: https://zenn.dev/acs_developer/articles/anthropic-batch-custom-id-pattern-400 ）。

何が起きたか

YouTubeのコメントを分類するバッチ処理を、AnthropicのMessage Batches APIへ移した時の話です。結果をコメントへ戻すには、各リクエストの custom_id に何かのキーを入れる必要があります。最初に思いつくのは「元データのIDをそのまま入れる」方法です。

ところがYouTubeのコメントIDには . が入ることがあります（返信コメントのIDは親スレッドIDと返信IDを . でつないだ形です）。custom_id にはこの文字が使えません。

本記事では、当方の環境（anthropic Python SDK 2.3.0）で無効な custom_id を実際に送り、返ってきたエラーを確認しました。そのうえで、当方が採用した「連番キー＋手元の対応表」で突き合わせる実装を紹介します。

実測: どんなIDが弾かれるか

検証用に、max_tokens: 1 のリクエストを1〜4件だけ詰めたバッチを作成しました。custom_id だけを変えて messages.batches.create() を呼んでいます。

import anthropic

c = anthropic.Anthropic()

def req(cid):
    return {"custom_id": cid, "params": {
        "model": "claude-haiku-5-5", "max_tokens": 1,
        "messages": [{"role": "user", "content": "hi"}]}}

cases = {
    "dot":   ["Ugx3abcDEF.8xyz"],   # 「.」を含む外部ID（ダミー値）
    "len65": ["a" * 65],
    "jp":    ["コメント1"],
    "dup":   ["c0", "c0"],
    "empty": [""],
}
for name, ids in cases.items():
    try:
        c.messages.batches.create(requests=[req(i) for i in ids])
        print(name, "ACCEPTED")
    except anthropic.APIStatusError as e:
        print(name, e.status_code, e.body["error"]["message"])

結果（2026-10-30・当方実測。各ケース0.3〜0.4秒で応答）:

dot   400 requests.0.custom_id: String should match pattern '^[a-zA-Z0-9_-]{1,64}$'
len65 400 requests.0.custom_id: String should have at most 64 characters
jp    400 requests.0.custom_id: String should match pattern '^[a-zA-Z0-9_-]{1,64}$'
dup   400 requests: `custom_id`s must be unique within a batch. Duplicate `custom_id` found: `c0`
empty 400 requests.0.custom_id: String should have at least 1 character

5件とも invalid_request_error（HTTP 400）でした。ここから分かる制約は次の3つです。

制約	内容
使える文字	英数字・`_`・`-` のみ（`.`・`:`・`/`・日本語は不可）
長さ	1〜64文字
一意性	同じバッチの中で重複不可

1件の不正IDでバッチ全体が作られない

運用で効いてくるのはこちらです。正常なID2件と、. 入りのID2件を混ぜて投入しました。

c.messages.batches.create(requests=[req("c0"), req("c1"),
                                    req("UgzA.UgzB"), req("UgzC.UgzD")])

400 requests.2.custom_id: String should match pattern '^[a-zA-Z0-9_-]{1,64}$'

不正なものだけが除外されるのではなく、バッチそのものが作成されません（直後の batches.list() にも増えていませんでした）
エラーで報告されるのは最初の1件だけ（requests.2）。requests.3 も同じ理由で不正ですが、メッセージには出ません

作成の段階で弾かれるので、課金は発生しません。その代わり、定時ジョブの中で起きると「その回の処理が丸ごと止まる」形で表に出ます。しかも返信コメントは毎回あるわけではありません。トップレベルのコメントしかない日は通り、返信が1件でも混ざった日だけ落ちます。

なお、SDKの Request 型（TypedDict）はクライアント側で形式をチェックしないため、送信するまでエラーに気づけません。

対処: 連番キーにして、対応表は手元に持つ

当方では、custom_id を「バッチ内の連番」にして、元のIDとの対応はローカルのデータに持たせました。

def build_requests(comments: list[dict], model: str) -> list[dict]:
    reqs = []
    for i, c in enumerate(comments):
        # YouTube のコメントID は "." を含み custom_id の規則 ^[a-zA-Z0-9_-]{1,64}$ に合わない
        c["cid"] = f"c{i}"
        reqs.append({"custom_id": c["cid"], "params": {
            "model": model, "max_tokens": 1500,
            "messages": [{"role": "user", "content": c["text"]}],
        }})
    return reqs

結果は custom_id で突き合わせます。Batchの結果は投入した順番で返るとは限らないため、リストの位置に頼らず、必ずキーで引きます。

out = {}
for res in c.messages.batches.results(batch.id):
    if res.result.type == "succeeded":
        out[res.custom_id] = res.result.message
    else:  # errored / expired / canceled
        print(f"{res.custom_id} {res.result.type}", file=sys.stderr)

for cm in comments:
    msg = out.get(cm["cid"])   # 元のコメントID（cm["id"]）は cm に残っている

他の方法と比べて連番にした理由

方法	長所	気になる点
`.` を `_` に置き換える	結果から元IDを復元しやすい	YouTubeのIDはもともと `_` と `-` を含むので、置き換えた結果が別のIDと同じ文字列にならないと言い切れない。外部IDの文字種が変わればまた落ちる
ハッシュ（sha1の16進40文字など）	長さと文字種が必ず規則に収まる	復元には結局、手元の対応表が要る
連番＋手元の対応表	規則違反と重複が起きない。短い	対応表（今回は `comments` のリスト）を、結果を受け取るまで保持する必要がある

結果を待つ間もプロセスが comments を持ち続ける構成（投入→完了までポーリング→突き合わせ）だったので、対応表を別途保存する必要はありませんでした。プロセスをまたいで結果を回収する構成なら、batch_id と一緒に対応表をファイルへ書いておく必要があります。

移行後の確認

直近2週間の実コメントを新経路で再処理し（台帳へは書かないdry-run）、旧経路が残した台帳と比べました。旧経路と同じ11件を拾い、11件すべてが連番キー経由で元のコメントへ戻りました。分類の一致は、状態が11件中10件（差の1件は新版で対象外にした短い一言）、ヒント種別が11件中9件、要注意フラグが1件中1件です。

もう1つ、件数が少ない時はBatchを使わないようにしています。Batchは多くの場合すぐ終わりますが、完了まで最大24時間かかる場合があります。定時ジョブで数件のために待つのは割に合わないので、8件未満は同期呼び出しに振り分けています。

def run(job, requests, *, batch_min=8, **kw):
    if len(requests) >= batch_min:
        return batch_run(job, requests, **kw)
    return sync_run(job, requests)

同期経路も custom_id をキーにした同じ形の辞書を返します。そのため、呼び出し側はどちらの経路を通ったかを気にせずに済みます。

まとめ

custom_id に使えるのは ^[a-zA-Z0-9_-]{1,64}$ に合う文字列だけで、バッチ内での重複も不可
不正なIDが1件でもあるとバッチ全体が400で作成されず、エラーには最初の1件しか出ない
外部IDをそのまま入れると「特定の形のデータが混ざった日だけ落ちる」障害になる。連番キーと手元の対応表に分けると、外部IDの形式に左右されなくなる
結果は順番ではなく custom_id で突き合わせる

検証の続きや関連ツールは ACS Developer で公開しています。