Claude API の Code Execution ツールに月次の売上CSV集計を任せる — Files API 連携から課金の落とし穴まで

最小構成で動かす — 応答ブロックの読み方から

最初の一歩は、ファイルなしの最小構成で応答の形を掴むことです。ここを飛ばして Files API 連携から入ると、応答の解析で必ず迷子になります。

import anthropic
 
client = anthropic.Anthropic()  # ANTHROPIC_API_KEY は環境変数で渡す
 
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    messages=[{
        "role": "user",
        "content": "1月から5月の売上 [412000, 389000, 455000, 401000, 478000] 円の平均と標準偏差を計算してください",
    }],
    tools=[{"type": "code_execution_20250825", "name": "code_execution"}],
)
 
# 応答は複数のコンテンツブロックの並びで返ってくる
for block in response.content:
    if block.type == "server_tool_use":
        # Claude がサンドボックスで実行したコマンドやコード
        print("実行内容:", block.input)
    elif block.type == "bash_code_execution_tool_result":
        result = block.content
        print("stdout:", result.stdout)
        print("return_code:", result.return_code)
    elif block.type == "text":
        print("Claude の説明:", block.text)

実行すると、server_tool_use ブロックに Claude が書いた計算コードが、続く bash_code_execution_tool_result ブロックの stdout に平均 427,000 円と標準偏差の計算結果が入り、最後の text ブロックで日本語の説明が締めくくられます。

ここで押さえておきたいのは、通常のツール呼び出しと違って実行が全てサーバー側で完結することです。自分で tool_result を組み立てて返す必要はなく、server_tool_use と *_tool_result のペアが応答の中に並びます。現行版ではこの結果ブロックが bash_code_execution_tool_result（シェル実行）と text_editor_code_execution_tool_result（ファイルの閲覧・作成・編集）の2系統に分かれているので、解析側は両方を想定しておきます。stderr と return_code を確認する癖をつけておくと、集計が失敗していたことに翌日まで気づかない、という事故を防げます。

Files API で売上 CSV をコンテナに渡す

応答の形が掴めたら本題です。月次レポート一式をアップロードし、container_upload ブロックでコンテナに載せます。

import anthropic
from pathlib import Path
 
client = anthropic.Anthropic()
 
# 月次レポート一式（App Store Connect / AdMob / メディエーション3社）
report_files = [
    "asc_financial_2026_05.csv",
    "admob_2026_05.csv",
    "unity_ads_2026_05.csv",
    "liftoff_2026_05.csv",
    "inmobi_2026_05.csv",
]
 
uploaded = []
for path in report_files:
    f = client.beta.files.upload(file=Path(path).open("rb"))
    uploaded.append({"type": "container_upload", "file_id": f.id})
 
AGGREGATION_RULES = """あなたはアプリ収益の月次集計を行います。次のルールを厳守してください。
 
- 通貨は全て JPY に統一する。USD 建ての行は exchange_rate 列があればそれを使い、なければ 1USD = 155JPY で換算する
- 集計期間は 2026-05-01 から 2026-05-31 の暦日に揃える。範囲外の行は除外する
- アプリの識別は bundle_id / package_name / app_id のいずれかの列を正とし、表記ゆれの対応表を最初に作る
- 集計結果はネットワーク別×アプリ別に $OUTPUT_DIR/summary_2026_05.csv へ書き出す
- ネットワーク別×アプリ別の積み上げ棒グラフを $OUTPUT_DIR/revenue_2026_05.png に保存する
- 推測で数値を補完しない。判断に迷った行は $OUTPUT_DIR/excluded_rows.csv に理由付きで書き出す
- 上記3つは同じ一回のコマンドの中で、$OUTPUT_DIR の直下へ平置きで書き出す
- 標準出力へ出すのは件数と合計だけにする。データフレームの全行を print しない
"""
 
response = client.beta.messages.create(
    model="claude-sonnet-4-6",
    betas=["files-api-2025-04-14"],
    max_tokens=8192,
    messages=[{
        "role": "user",
        "content": [{"type": "text", "text": AGGREGATION_RULES}] + uploaded,
    }],
    tools=[{"type": "code_execution_20250825", "name": "code_execution"}],
)

このコードの肝は、アップロード処理よりも AGGREGATION_RULES の方にあります。換算レート・期間の定義・除外基準といった会計上の約束事は、サンドボックスがどれだけ賢くても人間が決めて明文化するべき領域です。特に効いているのが最後の一行で、「迷った行は excluded_rows.csv に理由付きで書き出す」と指示しておくと、集計から落ちた行を後から必ず追跡できます。この指示を入れる前は、欠損のある行が黙って平均値で補完されかけたことがあり、以来どの集計依頼にも除外ログの出力を必ず入れるようにしています。

為替レートをプロンプトで渡しているのは、先に書いた「コンテナはネットに出られない」制約のためです。最初の月はここを見落として、Claude がレート取得の API を叩こうとして失敗する出力を眺めることになりました。外から与えるべき情報と、中で計算できる情報の線引きは、最初に紙に書き出しておくと迷いません。

アップロードの前に、手元で3点だけ検査する

ここまで書いておいて何ですが、私は client.beta.files.upload() を呼ぶ前に、必ずローカルで軽い検査を挟むようにしています。理由は後述する課金の仕様で、ファイルを添付した時点で実行時間の計上が始まるためです。列名が変わっていることに気づかないまま投げれば、サンドボックス側で失敗し、直してもう一度投げ直すことになります。この往復が、まるごと課金単位の無駄になります。

検査するのは3点だけです。必須列を解決できるか、通貨が混ざっていないか、締め期間の外にある行がどれだけあるか。

"""アップロード前に手元で回す3点検査（列名解決 / 通貨 / 期間）"""
import csv, re
from datetime import date, datetime
 
FIELD_PATTERNS = {
    "app":      re.compile(r"(bundle|package|app[ _]?(id|name)|vendor)", re.I),
    "revenue":  re.compile(r"(revenue|earning|share|proceeds)", re.I),
    "date":     re.compile(r"(date|day)", re.I),
    "currency": re.compile(r"(currency|^cur$)", re.I),
}
DATE_FORMATS = ("%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y", "%Y/%m/%d")
 
 
def parse_date(value):
    for fmt in DATE_FORMATS:
        try:
            return datetime.strptime(value.strip(), fmt).date()
        except ValueError:
            continue
    return None
 
 
def preflight(path, period_start, period_end):
    with open(path, newline="", encoding="utf-8") as fp:
        reader = csv.DictReader(fp)
        header = reader.fieldnames or []
        # 列名そのものではなく「意味」で引き当てる
        mapping = {}
        for field, pattern in FIELD_PATTERNS.items():
            hit = next((c for c in header if pattern.search(c)), None)
            if hit:
                mapping[field] = hit
        missing = [f for f in ("app", "revenue", "date") if f not in mapping]
 
        currencies, out_of_range, unparsable = set(), 0, 0
        for row in reader:
            if "currency" in mapping:
                currencies.add((row[mapping["currency"]] or "").strip().upper())
            parsed = parse_date(row[mapping["date"]]) if "date" in mapping else None
            if parsed is None:
                unparsable += 1
            elif not (period_start <= parsed <= period_end):
                out_of_range += 1
 
    return {
        "file": path, "mapping": mapping, "missing": missing,
        "currencies": sorted(c for c in currencies if c),
        "out_of_range": out_of_range, "unparsable": unparsable,
    }

この検査を、レポート形式のゆらぎを再現した CSV 5本（3,000行・合計 140,600 バイト）に対して手元の Python 3.10 で走らせたところ、5回平均 0.070 秒（最小 0.069 / 最大 0.070）でした。アップロード1回分の待ち時間にすら届きません。

素の状態で回すと、締め期間外の行が3ファイルで検出されるだけで通過します。ここで意図的に「レポート形式が変わった」状況を作り、revenue 列を net_payout に、日付を 2026-05-14 から 14.05.2026 に変えて再実行すると、出力はこうなりました。

unity_ads_2026_05.csv: 必須列を解決できず: ['revenue', 'date'] / 日付を解釈できない行: 600

終了コードは 1 です。600行すべてが日付として解釈できない状態、つまりそのままアップロードしていれば、集計対象がまるごと1ネットワーク分抜け落ちていたということになります。しかもこの種の欠落は合計金額が「少し小さい」だけの形で出てくるので、目視では気づきにくい類のものです。

正規表現で意味を引き当てる方式にしたのは、列名を固定で書くのをやめたかったからです。net_payout のようにパターンからも外れる改名は検出できませんが、そのときは「解決できず」として止まってくれます。静かに間違うより、うるさく止まる方を選びます。 集計処理に関しては、この一点だけを設計方針にしています。

生成物を回収できるのは $OUTPUT_DIR の直下だけです

先ほどの集計ルールで、書き出し先にわざわざ $OUTPUT_DIR を付けていたのには理由があります。応答に file_id として返ってくるのは、そのコマンドに割り当てられた $OUTPUT_DIR の直下に置かれたファイルだけです。 ほかの場所へ書いたものはコンテナの中には残りますが、応答には現れません。

しかも $OUTPUT_DIR は、bash の実行1回ごとに新しい空のディレクトリが割り当てられます。集計と作図を別々のコマンドへ分けたうえで、片方だけが $OUTPUT_DIR に書いていれば、もう片方は手元に届かないことになります。

これに気づいたのは、コンテナの中で ls を走らせた出力には summary が確かに並んでいるのに、ダウンロードの一覧には最後まで現れない、という噛み合わなさを追っていたときでした。ファイルが作られていないのではなく、作られた場所が回収の対象外だったわけです。中で整理したくなる気持ちは抑えて、外へ出したいものは直下に平置きする——集計ルールの末尾に一行足したのは、そういう理由です。

書き出し先を明示したうえで、応答に含まれる file_id を拾って Files API からダウンロードします。公式例は bash 実行結果だけを走査する書き方ですが、私は結果ブロック全般を緩く走査する形にしています。

def extract_generated_file_ids(response) -> list[str]:
    """応答内の全ツール結果ブロックから生成ファイルの file_id を集める"""
    file_ids = []
    for block in response.content:
        if not block.type.endswith("_tool_result"):
            continue
        content = getattr(block, "content", None)
        if content is None:
            continue
        # 実行結果の content リストに出力ファイルの参照が入る
        for item in getattr(content, "content", None) or []:
            file_id = getattr(item, "file_id", None)
            if file_id:
                file_ids.append(file_id)
    return file_ids
 
 
for file_id in extract_generated_file_ids(response):
    meta = client.beta.files.retrieve_metadata(file_id)
    client.beta.files.download(file_id).write_to_file(meta.filename)
    print(f"ダウンロード完了: {meta.filename}")

ダウンロードしたファイルを、コンテナ内のものとバイト単位で突き合わせようとして一度つまずいたことがあります。生成された画像には C2PA の Content Credentials が埋め込まれるため、PNG のサイズもチェックサムもコンテナ内の状態とは一致しません。差は数キロバイトです。署名が付くのは画像・音声・動画で、テキストや PDF、オフィス文書には付きません。ダウンロードの検証を入れるなら、突き合わせる相手は CSV の側にしておくのが無難です。

getattr ベースの防御的な書き方にしているのは、ツール結果の型がバージョンアップで増減してきた経緯があるからです。型を固定して isinstance で書くより、file_id を持つものを拾うという意図だけをコードに残す方が、この種の進化中の API には合うというのが実装してみての感触です。

ダウンロードしたファイルのうち、私が毎月必ず開くのは excluded_rows.csv です。除外された行の理由を眺めていると、「Liftoff のレポートにテスト配信の行が混ざっていた」「日付列のタイムゾーン表記が変わった」といった変化に気づけます。集計の答えよりも、答えから外れたものの方に運用の情報量があるというのは、月締めを繰り返してきたうえでの正直な実感です。

コンテナ再利用で「追い質問」のできる締め作業にする

月次の締めで一番ありがたかったのは、集計が一往復で終わらなくてよくなったことです。応答に含まれるコンテナ ID を次のリクエストに渡すと、同じサンドボックスを作業ファイルごと再利用できます。

container_id = response.container.id
 
followup = client.beta.messages.create(
    container=container_id,  # 同じサンドボックスを作業ファイルごと再利用する
    model="claude-sonnet-4-6",
    betas=["files-api-2025-04-14"],
    max_tokens=4096,
    messages=[{
        "role": "user",
        "content": "summary_2026_05.csv を読み直し、前月比の下落が大きいアプリ×ネットワークの組を3つ抜き出してください",
    }],
    tools=[{"type": "code_execution_20250825", "name": "code_execution"}],
)

会話としては新規でも、コンテナ上の summary_2026_05.csv はそのまま残っているので、アップロードし直さずに追い質問ができます。「この下落は単価と表示回数のどちらが原因か」「このネットワークだけ日別で見せてほしい」と、締めの数字を眺めながらの対話が成立します。

注意点は、コンテナは作成から30日で失効することです。container_id をデータベースに保存して翌月も使い回す設計にすると、その ID を指したリクエストはエラーで返ります。復旧そのものは単純で、container パラメータを外して投げ直せば新しいコンテナが割り当てられます。私は「締め作業のその日のうちだけ再利用し、月が変わったら必ず作り直す」と割り切りました。

ここで紛らわしいのが、応答に入っている container.expires_at です。この値は30日の期限を表すものではなく、もっと短い、リクエストのたびに動く値です。expires_at を根拠に「まだ使える／もう切れた」を判断すると誤ります。 5分ほど操作がないとコンテナはチェックポイントされますが、それは失効ではなく、30日以内であれば ID を送るだけで元の作業ファイルごと戻ってきます。判断の根拠は作成日の側に置いてください。なお新しい code_execution_20260120 版ならファイルに加えて REPL の変数状態も維持されますが、月次締めの用途では 20250825 のファイル持ち越しで十分というのが私の感触です。

課金の仕組みを先に理解しておく — 5分最小と無料枠

Code Execution ツールはトークンとは別建てで、実行時間に対して課金される仕組みです。執筆時点の公式仕様を整理すると次のようになります。

• 実行時間の課金は 1回あたり最低5分からカウントされます
• 組織ごとに月1,550時間の無料枠があり、超過分は1コンテナ・1時間あたり $0.05 です
• モデルのトークン料金（入出力）は通常どおり別にかかります
• web_search_20260209 または web_fetch_20260209 を同じリクエストに含める場合、コード実行自体の追加課金はありません（この併用にはコード実行側が code_execution_20260120 以降である必要があります）
• 応答の usage.server_tool_use.code_execution_requests で実行回数を確認できます

そして一番見落としやすい仕様がこれです。リクエストにファイルを含めると、ツールが一度も呼ばれなくても実行時間が課金されます。ファイルをコンテナへプリロードする処理が走るためです。プロンプトの言い回しを試行錯誤する段階では、ファイルを外して文面だけを調整し、固まってからファイル付きで投げる、という順番にするだけで無駄な5分課金の積み上がりを避けられます。

実際のコスト感も書いておきます。私の月次締めはコンテナ1本で5〜15分、追い質問を入れても月1時間に届きません。つまり実行時間は無料枠の中に完全に収まり、支払いはトークン分の数十円〜百円程度だけです。比較のために書くと、もし5ファイル・合計8MBほどのレポートを本文に貼ってトークンとして読ませようとすれば、概算で200万トークン規模になり、そもそも標準のコンテキストウィンドウに収まりません。仮に分割して収めたとしても入力トークンだけで数ドルかかります。ファイルはコンテナへ、本文には判断ルールを — この分担が精度の面でもコストの面でも素直でした。

pause_turn をまたいだ実行は、生成ファイルの回収を静かに取りこぼす

5ファイルの突き合わせのような長い処理では、ターンの途中で stop_reason が pause_turn になって応答が返ってくることがあります。これはエラーではなく「まだ続きがある」の合図で、返ってきた応答をそのまま会話に積み直して再リクエストすると、続きから実行されます。

messages = [{
    "role": "user",
    "content": [{"type": "text", "text": AGGREGATION_RULES}] + uploaded,
}]
 
while True:
    resp = client.beta.messages.create(
        model="claude-sonnet-4-6",
        betas=["files-api-2025-04-14"],
        max_tokens=8192,
        messages=messages,
        tools=[{"type": "code_execution_20250825", "name": "code_execution"}],
    )
    if resp.stop_reason != "pause_turn":
        break
    messages.append({"role": "assistant", "content": resp.content})

ここまでは公式ドキュメントどおりです。問題は、この後に前掲の extract_generated_file_ids(resp) を呼ぶ書き方をしていた点でした。

ループを抜けた時点の resp は最後のターンの応答だけです。summary_2026_05.csv を書き出した処理が最初のターンで完了し、グラフの描画が次のターンに回っていた場合、CSV の file_id は上書きされた応答と一緒に消えます。

手元で応答オブジェクトを模して確かめました。1ターン目で summary 相当のファイル、2ターン目で png 相当のファイルが生成される流れを再現し、両方の書き方に通します。

失われるのが CSV の方だった、というのが厄介でした。グラフは手元に届くので処理は成功したように見え、summary が無いことに気づくのは開こうとした瞬間です。

修正は、ループの中で毎回回収して足していくだけです。あわせて、想定外の連続 pause で回り続けないように上限も入れておきます。

MAX_TURNS = 10
 
collected_file_ids: list[str] = []
for turn in range(MAX_TURNS):
    resp = client.beta.messages.create(
        model="claude-sonnet-4-6",
        betas=["files-api-2025-04-14"],
        max_tokens=8192,
        messages=messages,
        tools=[{"type": "code_execution_20250825", "name": "code_execution"}],
    )
    # 中断されるターンでもファイルは生成され得るので、毎回回収する
    collected_file_ids += extract_generated_file_ids(resp)
 
    if resp.stop_reason != "pause_turn":
        break
    messages.append({"role": "assistant", "content": resp.content})
else:
    raise RuntimeError(f"pause_turn が {MAX_TURNS} ターン続きました。集計ルールを分割してください")
 
# 同じファイルが複数ターンで参照されることがあるので順序を保って重複を落とす
collected_file_ids = list(dict.fromkeys(collected_file_ids))

for に else を付けているのは、上限まで回り切ったケースを黙って通さないためです。集計ルールが欲張りすぎているときに pause が連鎖しやすく、そういうときは指示を「集計」と「可視化」の二往復に割るのが結局は速いという結論に落ち着きました。

失敗したときに何が返るか — error_code で再試行を分ける

成功した応答の読み方だけ覚えて運用へ入ると、最初の失敗で手が止まります。ツールが失敗したときは結果ブロックの代わりに bash_code_execution_tool_result_error が返り、その error_code を見れば理由がわかる作りになっています。現行で定義されているのは6種です。

分け方そのものは単純で、待てば直るものだけを再試行し、こちらの書き方が原因のものは止めます。 invalid_tool_input を再試行の側へ混ぜてしまうと、同じリクエストを投げ直しては5分課金だけが積み上がります。

output_file_too_large は、集計用途だと意外に踏みます。原因はたいてい「確認のつもりで print(df) を書かせた」ことで、数千行がそのまま標準出力へ流れます。集計ルールの側に「標準出力へ出すのは件数と合計だけ」と一行足してからは起きなくなりました。ルールを足すたびに集計の質が上がるというより、失敗の種類が一つずつ減っていく感覚に近いのだと感じています。

よくある間違いと落とし穴

運用を始めてから踏んだもの、踏みかけたものをまとめます。

1. 旧バージョンのサンプルコードの混入。 検索上位には Python 専用だった code_execution_20250522 時代の記事がまだ多く、応答の解析コードをそこから写すと現行版の bash_code_execution_tool_result / text_editor_code_execution_tool_result を取りこぼします。ツール型は 20250825 に統一し、応答の解析も現行の2系統で書き直してください。

2. 「ツールを呼ばなければ無料」という思い込み。 前述のとおり、ファイル付きリクエストはプリロードだけで課金が始まります。ファイルを添付したままプロンプトだけ10回調整すれば、実行ゼロでも50分相当が無料枠から消えます。試行錯誤はファイルなしで行うのが安全です。

3. pause_turn を終端とする読み違い。 途中で返ってきた応答を最終結果として読んでしまう事故です。前節のとおり、応答の積み直しと file_id の累積回収をループの形で最初から組み込んでおけば防げます。

4. コンテナ ID の恒久的な保存。 30日で失効します。expires_at はその期限を表さないので、月をまたぐ再利用は設計に入れず、毎月作り直します。

5. サンドボックスの「ネットなし」の失念。 為替レートの取得も pip install もできません。外部から与える情報はプロンプトに明記し、ライブラリはプリインストール一覧で足りるかを先に確認します。集計用途なら pandas・numpy・matplotlib・openpyxl が揃っているので、私のケースでは不足はありませんでした。

6. 生成物を $OUTPUT_DIR の外へ書く。 コンテナの中には残るのに応答へは現れないので、「作られていない」と誤読しがちです。集計ルールに書き出し先を明記し、サブディレクトリへ入れないことを条件に入れておきます。

7. 機微なデータの素通し。 この機能はゼロデータ保持（ZDR)の対象外で、コンテナ上のデータは最大30日保持されます。売上レポート自体に個人情報はほぼ含まれませんが、ユーザーレビューやサポートログを同じ仕組みに流すなら、ユーザーレビューやクラッシュログをClaude APIに渡す前に — 個人情報を可逆マスキングする前処理の設計メモで書いたようなマスキング前処理を必ず挟んでください。アップロードしたファイルの管理全般は Claude API Files API — ドキュメントを永続化してAPIコストを大幅削減も参考になるはずです。

まず AdMob の CSV を1本渡すところから

全部を一度に組む必要はありません。最初の一歩としておすすめなのは、今月の AdMob レポートを1本だけアップロードし、最小構成のコードに container_upload ブロックを1つ足して、「日別の推定収益を折れ線グラフにして png で保存してください」とだけ頼んでみることです。応答ブロックの並び方と file_id の回収まで一巡すれば、あとは集計ルールを1行ずつ育てていくだけで月次の締めまで届きます。2本目のレポートを足すタイミングで、先の3点検査を前段に置いてください。列名のゆらぎが表に出てくるのは、たいていそこからです。

形式の揺れに振り回されてきた月末の作業が、判断ルールを言語化して渡す作業に変わると、締めの時間そのものが運用を見直す時間になります。私自身、いまも excluded_rows.csv を眺めては換算ルールを書き足している途中ですが、この往復は嫌いではありません。お読みいただきありがとうございました。