はじめに

Claude Code on Bedrock 環境を複数の IAM ユーザー・ロールで共有していると、Cost Explorer ではサービス単位の合計しか分からず、「誰が」「どのモデルを」「どれだけ」使ったのかが見えません。
そこで CUR（Cost and Usage Report）2.0 と Athena、Lambda を組み合わせて、Bedrock コストを個人単位で可視化し、日次で通知し、月次予算を超えた利用者は自動で止める仕組みを作りました。

本記事では、設計・実装と、本番環境へ反映する過程で踏んだ落とし穴を整理します。

事前に以下を確認しておきます。

管理アカウント側で、メンバーアカウントの請求情報アクセスが SCP や Organizations 設定で制限されていないこと
メンバーアカウントで「IAM ユーザー・ロールによる請求情報へのアクセス」を有効化していること

本方式は「AssumeRole のセッション名にメールアドレスが入る」ことを利用して利用者を特定します。IAM Identity Center を使っていない場合や、セッション名が固定値の構成では個人の特定ができません。その場合は後述のフォールバック（ARN のまま集計）で読み替えてください。

コスト可視化の方式選定

Bedrockの料金を算出して見える化するには様々な方式がありますが、できるだけ簡潔に、個人単位でのトレースができること、料金計算をAWS側におまかせ、という基準でCUR2.0を選びました。

アーキテクチャ	概要	メリット	デメリット
CUR(本方式)	CUR 2.0のデータエキスポートから集計	料金計算が不要、IAM Principal単位で追跡可能	反映が遅い
コスト配分タグ	IAM Principalにタグを付与する	タグ単位で参照可能	タグ付与の運用コストが高い、個人単位の追跡には不向き
Bedrock Invocation Log	Bedrock Model invocation loggingを集計	ほぼリアルタイム、個人別・モデル別など集計軸を自由度が高い	単価・キャッシュトークン差分・価格改定への追従を自前で保守
CloudTrail/CloudWatch Logs/Metrics	ログやメトリクスで制御	実装が軽量かつ分解能が高い	単価・キャッシュトークン差分・価格改定への追従を自前で保守
Budgets	Budgets Actions+他方式と併用	標準機能のみで完結し、超過時挙動・解除フロー・監査ログまで含めたガバナンス設計をしやすい	個人単位の細かい制御には他方式との併用が前提

CURはコストは正確ですが、反映が遅いのが難点です。本記事では、CUR だけで「止める」を実装しているため、超過を検知した時点で実際の消費はさらに膨らんでいる点には注意してください。即時性が必要な場合には、CloudWatch や Invocation Logを使うなどしてください。

できること

料金を「見る」だけで終わらせず、「知らせる」、「止める」まで含めました。

見る利用者自身が自分のBedrock利用コストをその場で確認する
知らせる IAMプリンシパル×推論モデル単位のコスト内訳を、日次で当月累積として通知する
止める 月次予算を超えた利用者(IAMロール)に対して、Bedrock呼び出しを停止する

全体像

独立した5つのコンポーネントで構成しました。

(1)データ基盤

CUR2.0 Data Export → S3 → Glue Crawler → Athena View

(2)通知

EventBridge Scheduler → Lambda → Athenaクエリ → Teams Incoming Webhook

(3)ダッシュボード

EventBridge Scheduler → Lambda → S3
CloudWatch Dashboard → Lambda(Custom Widget) → S3を読んでHTML返却

(4)予算ガードレール

EventBridge Scheduler → Lambda → Athenaで当月コスト判定
→ DynamoDBの許可段階を参照 → 超過者にIAM Denyを付与

(5)利用者向けツール

各自がAIエージェントツールを介してコストを参照する仕組みを提供(配布)

以降、コンポーネントごとに実装について見ていきます。

アーキテクチャー

(1)データ基盤

データエクスポート

「請求とコスト管理」-「データエクスポート-「作成」から、AWS Cost and Usage Report(CUR) 2.0 のデータをデクスポートします。

このデータ基盤の土台は、CUR 2.0 の IAM プリンシパル単位でのコスト属性（INCLUDE_IAM_PRINCIPAL_DATA）です。

その他のエクスポートコンテンツ

具体的には、「その他のエクスポートコンテンツ」-「発信者ID(IAMプリンシパル)割り当てデータを含める」オプションをオンにします。このオプションを有効(オン)にすることで、CURの各明細行に line_item_iam_principal 列が含まれ、そのコストを発生させたIAMロールやIAMユーザーが行レベルで分かります。

時間粒度

一日一回、日次の選択です。

列選択

必要な列のみ選択可能です。今回はデフォルト(全列選択)です。

レポートデータ統合

Amazon Athena統合をオンにしてエクスポートデータを作ると、出力先S3バケットにGlueクローラーをデプロイするCloudFormationテンプレートと、テーブル作成用のSQLが自動生成されます。

エクスポート作成直後はCrawlerの初回実行が終わるまでテーブルが存在しなためしばらく待つ必要がある
Glueデータベース名が、命名規則 athenadataexports_固定となってしまう

注意事項

その他のリソースIDを含める(INCLUDE_RESOURCES)やコスト配分データを分割(INCLUDE_SPLIT_COST_ALLOCATION_DATA)などのオプションは全部オフにしています。これらをオンにすると、Athenaのスキャン量課金に影響を及ぼします。コスト可視化の仕組み自体がコストを生む、というのはさすがに避けたかったので、ここは最初にちゃんと絞っておく必要があります。

Athena

CUR2.0のAthena統合によってS3バケットに、Glue CrawlerをデプロイするCfnテンプレ、テーブル作成のためのSQLが格納されます。これらを実行することで環境を整えます。

個人別集計のベースとなるAthena Viewの定義は以下のとおりです。

CREATE OR REPLACE VIEW "vw_bedrock_user_cost" AS 
SELECT
  billing_period
, regexp_extract(line_item_iam_principal, '([^/]+@[^/]+)$') iam_principal_email
, regexp_replace(COALESCE(product['product_name'], 'Unknown'), ' \(Amazon Bedrock Edition\)$', '') product_name
, round(sum(COALESCE(line_item_unblended_cost, 0)), 4) total_cost
, count(*) line_count
, max(date(line_item_usage_start_date)) last_usage_date
FROM
  athenadataexports_bedrock_cost_athena.data
WHERE ((line_item_iam_principal IS NOT NULL) AND (line_item_operation LIKE 'InvokeModel%'))
GROUP BY 1, 2, 3

SELECT句
billing_period
dataテーブルの実パーティション列(yyyy-MM形式)をそのまま使います。

regexp_extract(line_item_iam_principal, '([^/]+@[^/]+)$') iam_principal_email
line_item_iam_principalにはIAMプリンシパルのARNがそのまま入っています。正規表現でメールアドレスを抜き出しています。

regexp_replace(COALESCE(product['product_name'], 'Unknown'), ' (Amazon Bedrock Edition)$', '') product_name
product列はmap型(map)なのでproduct['product_name']でキーアクセスしています。COALESCEは値が無い場合に'Unknown'にフォールバックする処理。regexp_replaceは、Bedrock経由のモデル名に付く"(Amazon Bedrock Edition)"という接尾辞を取り除いて、モデル名を整えています(例: "Claude Sonnet 5 (Amazon Bedrock Edition)" → "Claude Sonnet 5")。

round(sum(COALESCE(line_item_unblended_cost, 0)), 4) total_cost
コストの合計です。COALESCEでコスト列がNULLのケースを0として扱ってから合計し、小数点4桁に丸めています。

count(*) line_count
集計対象になった明細行数(＝何回分の課金レコードがまとまったか)です。

max(date(line_item_usage_start_date)) last_usage_date
そのグループの中で最も新しい利用日です。line_item_usage_start_dateはタイムスタンプなのでdate()で日付部分だけに変換してから最大値を取っています。

WHERE句
WHERE ((line_item_iam_principal IS NOT NULL) AND (line_item_operation LIKE 'InvokeModel%'))
「呼び出し元IAMプリンシパルが特定できる」かつ「Bedrockのモデル呼び出しAPI」に該当する行だけに絞り込む条件です。

GROUP BY 1, 2, 3
billing_period・iam_principal_email・product_nameの組み合わせごとに1行を作り、「ある月に、ある利用者が、あるモデルをいくら使ったか」を表す行の集まりになります。

注意事項

【ハマりポイント】パーティションプルーニング
Athena は実体として S3 上のファイルにクエリを投げる仕組みで、CUR のようなデータは日々ファイルが増えていきます。何も考えずにクエリすると「先月のコストだけ」を知りたい場合でも過去数年分のファイルを読みにいってしまいます。
これを防ぐため、S3 側のファイルは billing_period=2026-09/ のように月ごとのフォルダ（パーティション）に分けて保存されており、WHERE 句でこのフォルダ名の列（パーティションキー）そのものを比較すれば、Athena は該当フォルダだけを読んで済ませられます（パーティションプルーニング）。
今回ハマったのは、billing_period という列名は同じでも、中身が別の日付列を format_datetime で加工した計算結果になっていたケースです。フォルダ名そのものではなく計算結果を比較する形になっていたため、Athena は各ファイルを実際に開いて計算しないと対象かどうか判断できず、スキャン量が想定以上に膨らんでいました。

NGパターン
# billing_period という名前だが中身は計算式で、パーティションキーとは無関係の値になる
SELECT
  format_datetime(bill_billing_period_start_date, 'MMM-yy') AS billing_period,
  ...
FROM data
GROUP BY 1, ...

OKパターン
# ベーステーブルの実パーティション列をそのまま使う
SELECT
  billing_period,
  ...
FROM data
GROUP BY 1, ...

同じ条件でスキャン量を比較した場合、おおよそ6倍の差が出ていました。VIEWって難しい・・・

line_item_product_code を AmazonBedrock でフィルタすると、Amazon純正モデルの呼び出しは拾えてもAnthropic ClaudeのようなAWS Marketplace経由で課金されるサードパーティ基盤モデルが集計対象から漏れてしまいます。

WHERE ((line_item_iam_principal IS NOT NULL) AND (line_item_operation LIKE 'InvokeModel%'))

この2つの条件をANDで組み合わせることで、「Bedrockのモデルを呼び出して発生し、かつ呼び出し元のIAM ARNが分かっている」明細だけが残ることになります。これが個人別のBedrockコスト集計のベースとなります。

(2)通知

Athena Viewからデータを取得します。

import boto3
import time

QUERY = """
SELECT
  iam_principal_email,
  product_name,
  total_cost
FROM vw_bedrock_user_cost
WHERE billing_period = format_datetime(current_date, 'yyyy-MM')
ORDER BY total_cost DESC
"""

_MAX_RETRIES = 30
_POLL_INTERVAL = 10


def run_query(database: str, output_location: str, query: str) -> list:
    client = boto3.client("athena")
    execution_id = _start(client, database, output_location, query)
    _wait(client, execution_id)
    return _fetch_results(client, execution_id)


def _start(client, database: str, output_location: str, query: str) -> str:
    response = client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={"Database": database},
        ResultConfiguration={"OutputLocation": output_location},
    )
    return response["QueryExecutionId"]


def _wait(client, execution_id: str):
    for _ in range(_MAX_RETRIES):
        resp = client.get_query_execution(QueryExecutionId=execution_id)
        state = resp["QueryExecution"]["Status"]["State"]
        if state == "SUCCEEDED":
            return
        if state in ("FAILED", "CANCELLED"):
            reason = resp["QueryExecution"]["Status"].get("StateChangeReason", "unknown")
            raise RuntimeError(f"Athena query {state}: {reason}")
        time.sleep(_POLL_INTERVAL)
    client.stop_query_execution(QueryExecutionId=execution_id)
    raise TimeoutError(f"Athena query timed out after {_MAX_RETRIES * _POLL_INTERVAL}s")


def _fetch_results(client, execution_id: str) -> list:
    rows = []
    headers = None
    paginator = client.get_paginator("get_query_results")
    for page in paginator.paginate(QueryExecutionId=execution_id):
        page_rows = page["ResultSet"]["Rows"]
        if headers is None:
            headers = [col["VarCharValue"] for col in page_rows[0]["Data"]]
            page_rows = page_rows[1:]
        for row in page_rows:
            values = [col.get("VarCharValue", "") for col in row["Data"]]
            rows.append(dict(zip(headers, values)))
    return rows

Teams Incoming Webhookの準備

Teamsのチャンネル名を右クリック→「コネクタ」→「Incoming Webhook」→「追加」からWebhook URLを発行します。このURLの送信先がPower Automateのワークフロー形式に切り替わっている場合、正常受理時にHTTP 202を返します。status == 200だけで成功判定すると、実際には送信できているのに例外を投げてしまう点は注意しておきましょう。

Secrets ManagerへのURL登録

Teams で発行したWebhook URLはSecrets Managerに登録しておきます。

Lambda + EventBridge Schedulerの実装

EventBridge Schedulerで毎日決まった時刻にLambda関数を起動し、当月分をAthenaでクエリしてAdaptive Cardを組み立て、Webhookへ送信します。

import json
import boto3
import urllib.request
import urllib.error


def send_report(report: dict, secret_name: str):
    webhook_url = _get_webhook_url(secret_name)
    card = _build_adaptive_card(report)
    _post_to_teams(webhook_url, card)


def send_no_usage_report(billing_period: str, secret_name: str):
    webhook_url = _get_webhook_url(secret_name)
    card = {
        "type": "message",
        "attachments": [{
            "contentType": "application/vnd.microsoft.card.adaptive",
            "content": {
                "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                "type": "AdaptiveCard",
                "version": "1.5",
                "body": [{
                    "type": "TextBlock",
                    "text": f"Bedrock 利用コスト — {billing_period} は利用なし",
                    "weight": "Bolder",
                }],
            },
        }],
    }
    _post_to_teams(webhook_url, card)


def _get_webhook_url(secret_name: str) -> str:
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId=secret_name)
    secret_string = response["SecretString"]
    try:
        parsed = json.loads(secret_string)
    except (json.JSONDecodeError, TypeError):
        return secret_string
    if isinstance(parsed, dict) and "url" in parsed:
        return parsed["url"]
    return secret_string


def _build_adaptive_card(report: dict) -> dict:
    user_rows = []
    for user in report["users"]:
        user_rows.append({
            "type": "TableRow",
            "cells": [
                {"type": "TableCell", "items": [{"type": "TextBlock", "text": user["email"], "wrap": True}]},
                {"type": "TableCell", "items": [{"type": "TextBlock", "text": f"${user['cost']:.2f}"}]},
                {"type": "TableCell", "items": [{"type": "TextBlock", "text": user["models"], "wrap": True, "size": "Small"}]},
            ],
        })

    return {
        "type": "message",
        "attachments": [{
            "contentType": "application/vnd.microsoft.card.adaptive",
            "content": {
                "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                "type": "AdaptiveCard",
                "version": "1.5",
                "body": [
                    {"type": "TextBlock", "text": f"Bedrock 利用コスト月次レポート — {report['billing_period']}", "weight": "Bolder", "size": "Medium"},
                    {"type": "TextBlock", "text": f"当月累積合計: **${report['total_cost']:.2f}**"},
                    {"type": "TextBlock", "text": "利用者別内訳", "weight": "Bolder", "spacing": "Medium"},
                    {
                        "type": "Table",
                        "columns": [{"width": 2}, {"width": 1}, {"width": 2}],
                        "rows": [
                            {
                                "type": "TableRow",
                                "style": "accent",
                                "cells": [
                                    {"type": "TableCell", "items": [{"type": "TextBlock", "text": "利用者", "weight": "Bolder"}]},
                                    {"type": "TableCell", "items": [{"type": "TextBlock", "text": "コスト", "weight": "Bolder"}]},
                                    {"type": "TableCell", "items": [{"type": "TextBlock", "text": "モデル", "weight": "Bolder"}]},
                                ],
                            },
                            *user_rows,
                        ],
                    },
                ],
            },
        }],
    }


def _post_to_teams(webhook_url: str, payload: dict):
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        webhook_url,
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                if not (200 <= resp.status < 300):
                    raise urllib.error.URLError(f"HTTP {resp.status}")
                return
        except urllib.error.URLError as e:
            if attempt == 2:
                raise RuntimeError(f"Teams Webhook 送信失敗: {e}") from e

Teamsでの通知イメージです。

(3)ダッシュボード

日次集計とS3キャッシュ

Lambda関数をEventBridge Schedulerで毎日起動し、Athena Viewで集計してS3にJSONキャッシュとして書き込みます。直近nヶ月分への絞り込みは、Python側でリストをフィルタするのではなくSQLのWHERE句で行うほうがよいです。WHERE句なしで全期間を取得してからPython側で絞る実装をしてしまうと、Athenaに対して毎日全期間フルスキャンになりコストが積み上がります。

CloudWatch Custom Widgetの登録

CloudWatch DashboardにCustom Widgetを追加し、endpointにwidget用Lambdaの関数ARNを指定します。widget側のLambda実行ロールにはAthena/Glueへの権限を一切持たせず、S3のキャッシュキー1つへのs3:GetObjectのみに絞り、widgetはAthenaに直接アクセスしない設計にしています。

CloudWatch Custom Widgetは、CloudWatchサービス自体がLambdaを呼び出すのではなく、ダッシュボードを閲覧しているIAMプリンシパル自身が対象Lambdaへのlambda:InvokeFunction権限を持つ必要があります。閲覧させたい各ユーザー・ロールのIAMポリシーに個別でlambda:InvokeFunctionを追加します。

(4)予算ガードレール

しきい値管理

個人ごとに3段階の可変コストしきい値を持たせます。しきい値と現在の許可段階はDynamoDBで管理します。しきい値変更や、段階の変更については管理者用のコマンドから実行します。

試験運用

DRY_RUNパラメータを用意して、試験運用モードを提供しています。デフォルトはtrue(試験運用)です。明示的にfalse(本番運用)に切り替えたときのみ、予算超過時のIAMロールへのDeny付与を行います。

停止処理

超過したユーザーの IAM ロールに、Bedrock 推論系 API を明示的に拒否するインラインポリシーを動的に付与します。IAM のポリシー評価では明示的な Deny が常に Allow より優先されるため、対象ロールが他にどんな許可ポリシーを持っていても、このポリシーが付いている間は推論 API が確実にブロックされます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DenyBedrockInferenceOnBudgetExceeded",
      "Effect": "Deny",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream",
        "bedrock:Converse",
        "bedrock:ConverseStream"
      ],
      "Resource": "*"
    }
  ]
}

(5)利用者向けツール

Claude Code向けスラッシュコマンド(Claude Codeに依存しないbashスクリプトを含む)と、Claude Codeのステータスにコストを表示するスクリプトを配布しています。

自分の数字がその場で見えると、コスト意識は芽生えるはずです(多分)。

改善ポイント

今回、Athena Viewに対して行レベルセキュリティを実装していません(ちょっとサボりました)。さすがにほかの人のデータを見たとて、というところもあり、そこは性善説にお任せしています。厳密に他人のデータを隠す必要がある場合は、AWS Lake Formationの行レベルセキュリティが必要となります。
対象は IAM ロールのみ。ユーザー・アクセスキー経由のアクセスは対象外です。
CUR の反映遅延により、リアルタイムでの利用停止はできません。
クロスリージョン推論プロファイルを使う場合、明細の見え方が変わるのかどうかは要検証です。

あと、管理アカウントに CUR を集約したマルチアカウント横断の個人別集計などは、さらに大きい組織の可視化と制御としてやってみたいと考えています。

おわりに

コストの「誰が」「いくら」をCUR2.0とAthenaで見える化するところまでは比較的素直に組むことができると思います。

今回、難しいなと感じたのは、見える化した数字をどう運用に落とし込むかという点、それからパーティション設計・エクスポート定義・View定義という複数のレイヤーにまたがるデータ構造の整合性を保持することでした。すぐできるかなと思ったのですが、結構はまりどころはあったので、実務レベルまでもっていくのは結構骨が折れました。

今回の可視化＆制御がどこかの組織のお役に立てば幸いです。