【Claude CodeとAWSで作るAIエージェントの土台 第8回】守りと費用：401 の監視、WAF の回数制限、確認用はふだん止める

前回は、既存のリポジトリから移す前の検査と、WordPress への投稿の中継を作りました。これで「AI にはトークンだけを渡し、API が役割を判定して WordPress に投稿する」形がひととおり動きます。

あわせて読みたい

【Claude CodeとAWSで作るAIエージェントの土台 第7回】既存のリポジトリから移す：秘密の値の検査と、W… AIエージェントの土台をAWSに作る連載の第7回。既存のリポジトリを移す前の秘密の値の検査と置き場の決め方、WordPressのパスワードをSecrets Managerに置いてAPIが投稿を中継する実装を解説します。

最終回は、動かし続けるための「守り」と「費用」です。インターネットに公開した API には、自社の AI 以外からも呼び出しが届きます。

「API を公開したら、知らない相手から大量に呼ばれたりしない？ 費用も膨らまないか心配…」

結論から言うと、トークン違い（401）を記録して数え、続いたら知らせる。WAF で呼び出しの回数を制限する。確認用の環境はふだん止める。この3つで、気付ける・止められる・払いすぎない状態にします。

API の守り：401 を記録して数える

/healthz（起動確認）以外の呼び出しは、すべてトークンを求めます（第5回）。そのうえで、トークン違い（401）のときに1行の記録を出し、CloudWatch で数えます。

app/main.py に次の変更を加えます（+ が追加、- が削除）。

--- a/app/main.py
+++ b/app/main.py
@@ -12,7 +12,7 @@
 from pathlib import Path

 import httpx
-from fastapi import Depends, FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException, Request
 from fastapi.responses import PlainTextResponse
 from pydantic import BaseModel, Field

@@ -37,6 +37,20 @@
     app.state.tokens = tokens or TokenStore(settings.tokens_table, settings.local_tokens_file)
     app.state.secrets = secrets or Secrets(settings.secret_prefix)
     app.state.http = http or httpx.AsyncClient(follow_redirects=False)
+
+    @app.middleware('http')
+    async def record(request: Request, call_next):
+        response = await call_next(request)
+        if response.status_code == 401:
+            # トークン違いを1行の記録に出す。CloudWatch で数え、続いたら知らせる（トークンの値は書かない）
+            print(json.dumps({'event': 'auth_failed', 'method': request.method, 'path': request.url.path}),
+                  flush=True)
+        elif request.url.path != '/healthz':
+            p = getattr(request.state, 'principal', None)
+            print(json.dumps({'event': 'call', 'role': p.role if p else '-', 'token_id': p.token_id if p else '-',
+                              'method': request.method, 'path': request.url.path,
+                              'status': response.status_code}, ensure_ascii=False), flush=True)
+        return response

     @app.get('/healthz')
     def healthz():

• 401 のときは {"event": "auth_failed", "method": ..., "path": ...} の1行を出します。トークンの値は書きません。記録を見られる人がトークンを手に入れられては意味がないためです
• それ以外の呼び出しは、役割・トークンの ID（ハッシュの先頭12文字）・パス・結果を記録します。誰がいつ何をしたかを、あとから追えます
• ECS のコンテナが標準出力に出した行は、CloudWatch Logs に集まります（第4回でロググループを作りました）

当社の社内基盤では、呼び出しの記録を DynamoDB にも残し、90日で自動で消しています。

tests/test_api.py に、記録にトークンの値が出ないことのテストを足します。

--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -36,3 +36,10 @@
                        json={'title': 't', 'content': 'c', 'status': 'publish'})
     assert res.json()['status'] == 'draft' and res.json()['requested_status'] == 'publish'
     assert json.loads(wp.calls[-1].content)['status'] == 'draft'
+
+
+def test_auth_failure_is_logged_without_the_token(prod, capsys):
+    prod.get('/me', headers=bearer('th_secret_value'))
+    lines = [line for line in capsys.readouterr().out.splitlines() if 'auth_failed' in line]
+    assert json.loads(lines[-1]) == {'event': 'auth_failed', 'method': 'GET', 'path': '/me'}
+    assert 'th_secret_value' not in ''.join(lines)

監視と WAF を CDK に足す

infra/stacks/api_stack.py に次の変更を加えます。長いので折りたたんでいます。

--- a/infra/stacks/api_stack.py
+++ b/infra/stacks/api_stack.py
@@ -6,18 +6,25 @@
 - ECS Fargate：ALB からだけ受け付ける。失敗したら一つ前の版に自動で戻す
 - トークンの表（DynamoDB）：ハッシュだけを置く
 - WordPress のパスワード：Secrets Manager の <prefix>/wp/<サイト> を API だけが読む（CDK では作らない）
+- WAF（設定で）：トークンなしの呼び出しは 5分で100回、全体は 5分で3,000回まで
+- 監視：トークン違い（401）が 5分で20回を超えたら知らせる
 """

 from pathlib import Path

 import aws_cdk as cdk
 from aws_cdk import aws_certificatemanager as acm
+from aws_cdk import aws_cloudwatch as cw
+from aws_cdk import aws_cloudwatch_actions as cw_actions
 from aws_cdk import aws_dynamodb as ddb
 from aws_cdk import aws_ec2 as ec2
 from aws_cdk import aws_ecs as ecs
 from aws_cdk import aws_elasticloadbalancingv2 as elbv2
 from aws_cdk import aws_iam as iam
 from aws_cdk import aws_logs as logs
+from aws_cdk import aws_sns as sns
+from aws_cdk import aws_sns_subscriptions as subs
+from aws_cdk import aws_wafv2 as wafv2
 from constructs import Construct

 CODE_DIR = Path(__file__).resolve().parents[2]  # Dockerfile のある場所
@@ -33,6 +40,9 @@
         stage: str,
         domain_name: str,
         certificate_arn: str,
+        waf_enabled: bool,
+        running: bool,
+        alarm_email: str = "",
         **kwargs,
     ):
         super().__init__(scope, cid, **kwargs)
@@ -90,7 +100,7 @@
             "Service",
             cluster=cluster,
             task_definition=task,
-            desired_count=1,
+            desired_count=1 if running else 0,
             assign_public_ip=True,
             security_groups=[task_sg],
             circuit_breaker=ecs.DeploymentCircuitBreaker(enable=True, rollback=True),
@@ -123,6 +133,83 @@
             default_action=elbv2.ListenerAction.redirect(protocol="HTTPS", port="443", permanent=True),
         )

+        # --- WAF（設定で） ---
+        if waf_enabled:
+
+            def visibility(name):
+                return wafv2.CfnWebACL.VisibilityConfigProperty(
+                    cloud_watch_metrics_enabled=True, metric_name=f"{cid}-{name}", sampled_requests_enabled=True
+                )
+
+            def rate(name, priority, limit, scope_down=None):
+                return wafv2.CfnWebACL.RuleProperty(
+                    name=name,
+                    priority=priority,
+                    action=wafv2.CfnWebACL.RuleActionProperty(block={}),
+                    statement=wafv2.CfnWebACL.StatementProperty(
+                        rate_based_statement=wafv2.CfnWebACL.RateBasedStatementProperty(
+                            limit=limit,
+                            aggregate_key_type="IP",
+                            evaluation_window_sec=300,
+                            scope_down_statement=scope_down,
+                        )
+                    ),
+                    visibility_config=visibility(name),
+                )
+
+            without_token = wafv2.CfnWebACL.StatementProperty(
+                not_statement=wafv2.CfnWebACL.NotStatementProperty(
+                    statement=wafv2.CfnWebACL.StatementProperty(
+                        byte_match_statement=wafv2.CfnWebACL.ByteMatchStatementProperty(
+                            field_to_match=wafv2.CfnWebACL.FieldToMatchProperty(
+                                single_header={"Name": "authorization"}
+                            ),
+                            positional_constraint="STARTS_WITH",
+                            search_string="Bearer th_",
+                            text_transformations=[wafv2.CfnWebACL.TextTransformationProperty(priority=0, type="NONE")],
+                        )
+                    )
+                )
+            )
+            acl = wafv2.CfnWebACL(
+                self,
+                "Waf",
+                scope="REGIONAL",
+                default_action=wafv2.CfnWebACL.DefaultActionProperty(allow={}),
+                visibility_config=visibility("all"),
+                rules=[rate("rate-without-token", 0, 100, without_token), rate("rate-all", 1, 3000)],
+            )
+            wafv2.CfnWebACLAssociation(self, "WafAlb", resource_arn=alb.load_balancer_arn, web_acl_arn=acl.attr_arn)
+
+        # --- 監視：トークン違い（401）が続いたら知らせる ---
+        logs.MetricFilter(
+            self,
+            "AuthFailed",
+            log_group=logs_group,
+            filter_pattern=logs.FilterPattern.string_value("$.event", "=", "auth_failed"),
+            metric_namespace="ai-agent-kiban",
+            metric_name=f"{stage}-auth-failed",
+            metric_value="1",
+        )
+        topic = sns.Topic(self, "Alarms")
+        if alarm_email:
+            topic.add_subscription(subs.EmailSubscription(alarm_email))
+        cw.Alarm(
+            self,
+            "AuthFailedAlarm",
+            alarm_description="トークン違い（401）が続いている",
+            metric=cw.Metric(
+                namespace="ai-agent-kiban",
+                metric_name=f"{stage}-auth-failed",
+                period=cdk.Duration.minutes(5),
+                statistic="Sum",
+            ),
+            threshold=20,
+            evaluation_periods=1,
+            comparison_operator=cw.ComparisonOperator.GREATER_THAN_THRESHOLD,
+            treat_missing_data=cw.TreatMissingData.NOT_BREACHING,
+        ).add_alarm_action(cw_actions.SnsAction(topic))
+
         cdk.CfnOutput(
             self,
             "AlbDnsName",

401 が続いたら知らせる

• CloudWatch Logs のメトリクスフィルター（＝記録の中から条件に合う行を数える設定）で、$.event が auth_failed の行を数えます
• 5分間で20回を超えたら、アラーム（＝条件を超えたら知らせる設定）から SNS（＝通知を送るサービス）でメールを送ります

📰 出典：Filter pattern syntax for metric filters（Amazon CloudWatch Logs User Guide）

自社の AI がトークンの設定を間違えたとき（第5回の「3つがそろっていない」）にも気付けますし、誰かがトークンを当てずっぽうで試しているときにも気付けます。

WAF のルールの決め方（本番だけ）

AWS WAF（＝Web の呼び出しを、ルールに合わせて通す・止めるサービス）を ALB の前に置きます。サンプルのルールは2つです。

• トークンの形をしていない呼び出しは、自社の AI からのものではありません。少なめの回数で止めます
• 全体の上限は多めにしています。当社の社内基盤では、資料をまとめて取得すると数百件を続けて呼ぶため、少なすぎると自社の作業が止まります
• 回数を数える時間の幅は 60・120・300・600秒から選べます。サンプルは300秒（5分）です

📰 出典：Rate-based rule high-level settings（AWS WAF Developer Guide）

当社の社内基盤では、AWS が用意する「評判の悪い IP アドレス」の一覧（Amazon IP reputation list）のルールも置いています。最初は「数えるだけ」にして、Claude Code の通信元が当たらないことを確かめてから「止める」に切り替えました。また、WAF が回数の制限で止めたときも知らせるようにしています（自社の作業が止められていないかに気付くため）。

なお、Claude Code の通信元の IP アドレスは決まっていないので、ALB を IP アドレスで絞ることはできません。だからこそ、トークンと WAF の回数制限で守ります。

確認用はふだん止める

running が false のときは、ECS のサービスの台数を0にします。ALB や表は残すので、running=true で反映し直せば、同じ URL ですぐに使えます。

• ALB は止めません。作り直すと DNS 名が変わり、DNS の CNAME を毎回直すことになるためです
• 止めている確認用では、「API が止まっている」の知らせは出さないようにします（当社の社内基盤の設定。5xx と 401 の知らせは残します）

設定と CDK のテスト

infra/app.py に次の変更を加えます。

--- a/infra/app.py
+++ b/infra/app.py
@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 """AWS の設定（CDK）。1つのアカウントに、確認用か本番の API を1つ作る。

-  cdk deploy -c stage=prod -c domainName=api.example.com -c certificateArn=arn:aws:acm:...
+  cdk deploy -c stage=prod -c domainName=api.example.com -c certificateArn=arn:aws:acm:... -c wafEnabled=true

 - certificateArn：HTTPS の証明書（ACM で、DNS の確認を済ませておく。CDK では作らない）
+- wafEnabled：WAF（呼び出し回数の制限）を置くか。本番は true
+- running：API を動かすか（確認用はふだん false にして費用を抑える）
 """

 import aws_cdk as cdk
@@ -13,6 +15,11 @@
 ctx = app.node.try_get_context


+def flag(name, default):
+    v = ctx(name)
+    return default if v is None else str(v).lower() == "true"
+
+
 stage = ctx("stage") or "staging"
 ApiStack(
     app,
@@ -20,5 +27,8 @@
     stage=stage,
     domain_name=ctx("domainName") or "api.example.com",
     certificate_arn=ctx("certificateArn") or "",
+    waf_enabled=flag("wafEnabled", stage == "prod"),
+    running=flag("running", True),
+    alarm_email=ctx("alarmEmail") or "",
 )
 app.synth()

infra/cdk.json に次の変更を加えます。

--- a/infra/cdk.json
+++ b/infra/cdk.json
@@ -3,6 +3,9 @@
   "context": {
     "stage": "staging",
     "domainName": "api-stg.example.com",
-    "certificateArn": ""
+    "certificateArn": "",
+    "wafEnabled": false,
+    "running": true,
+    "alarmEmail": ""
   }
 }

infra/tests/test_api_stack.py に次の変更を加えます。

--- a/infra/tests/test_api_stack.py
+++ b/infra/tests/test_api_stack.py
@@ -11,7 +11,7 @@


 def build(**kw):
-    args = dict(stage="prod", domain_name="api.example.com", certificate_arn=CERT)
+    args = dict(stage="prod", domain_name="api.example.com", certificate_arn=CERT, waf_enabled=True, running=True)
     args.update(kw)
     app = cdk.App()
     return Template.from_stack(
@@ -42,3 +42,32 @@
     ]
     assert len(statements) == 1
     assert json.dumps(statements[0]["Resource"]).endswith('secret:ai-agent-kiban/prod/*"]]}')
+
+
+def test_waf_and_401_alarm():
+    t = build()
+    t.has_resource_properties(
+        "AWS::WAFv2::WebACL",
+        {
+            "Rules": Match.array_with(
+                [
+                    Match.object_like(
+                        {
+                            "Name": "rate-without-token",
+                            "Statement": {"RateBasedStatement": Match.object_like({"Limit": 100})},
+                        }
+                    ),
+                    Match.object_like(
+                        {"Name": "rate-all", "Statement": {"RateBasedStatement": Match.object_like({"Limit": 3000})}}
+                    ),
+                ]
+            )
+        },
+    )
+    t.has_resource_properties("AWS::CloudWatch::Alarm", {"Threshold": 20, "MetricName": "prod-auth-failed"})
+
+
+def test_staging_without_waf_and_stopped():
+    t = build(stage="staging", waf_enabled=False, running=False)
+    t.resource_count_is("AWS::WAFv2::WebACL", 0)
+    t.has_resource_properties("AWS::ECS::Service", {"DesiredCount": 0})

本番の設定で WAF の2つのルールと 401 のアラームがあること、確認用の設定（WAF なし・止める）では WAF が無く台数が0になることを確かめます。

反映と、止める・動かす

第3回のワークフローは、Environment の変数 WAF_ENABLED と ALARM_EMAIL を CDK に渡しています。本番の Environment では WAF_ENABLED を true に、確認用では false にします。

確認用を止めるとき・動かすときは、running を変えて反映します（手で行う例）。

cd infra
npx aws-cdk@2 deploy -c stage=staging -c domainName=api-stg.example.com \
  -c certificateArn=arn:aws:acm:ap-northeast-1:<アカウント番号>:certificate/<ID> -c running=false

当社の社内基盤では、GitHub Actions の手動実行（「Run workflow」）で確認用を動かす・止めるを切り替え、main に取り込んだときは止まる形にしています。

落とし穴：知らせが誰にも届いていなかった

SNS のメールの知らせは、送り先を設定したあと、AWS から届く確認のメールのリンク（「Confirm subscription」）を押すまで届きません。確認しないまま48時間たつと、送り先の登録は消えます。また、送り先が空のままだと、アラームが動いても誰にも届きません。反映したら、確認のメールを押して、試しにアラームを鳴らして届くことを確かめておきましょう。

📰 出典：Email notifications（Amazon SNS Developer Guide）

費用の目安

月の目安です（2026年10月時点の東京リージョン。1か月を730時間として計算。データ転送などは含みません）。

📰 出典：AWS WAF の料金

• WAF の月額は「置いている時間」で日割りされます。API を止めても、WAF を置いていればかかります。本番は常にインターネットから届くので、置いたままにします
• 確認用は、WAF を置かず、API もふだん0台にすることで、ALB と IP アドレスの分だけになります
• Secrets Manager は値1つあたり月0.4ドル、DynamoDB は呼び出しの量に応じた料金で、このくらいの規模では小さな金額です

動作確認

pip install -e ".[dev]"
ruff check .
pytest -q
cd infra && pytest -q tests && cd ..

テストが、プロジェクトの直下で14件、infra で4件通れば完成です。手元で API を起動し、トークンなしで /me を呼ぶと、起動したターミナルに次の1行が出ます。

{"event": "auth_failed", "method": "GET", "path": "/me"}

正しいトークンで呼ぶと、役割と ID が記録されます（トークンの値は出ません）。

{"event": "call", "role": "blog-sample", "token_id": "<ID>", "method": "GET", "path": "/me", "status": 200}

執筆環境では、AWS への反映（WAF とアラームが実際に動くこと）までは確かめていません。当社の社内基盤は同じ設定で動いています。

つまずきやすい点・セキュリティ上の注意

• 記録にトークンや本文を出さない：記録を見られる人の範囲は、API を使う人より広いことが多いです
• WAF の上限は、自社の使い方を見てから決める：少なすぎると自社の作業が止まります。最初は多めにし、記録を見て絞ります
• 「数えるだけ」から始める：AWS が用意するルールは、自社の通信が当たらないかを確かめてから「止める」に切り替えます
• 本番運用では監視を足す：サンプルは 401 の監視だけです。API が止まった、5xx が続いた、のときの知らせも足してください

発注者向けメモ

API や AI エージェントの仕組みを公開して運用するときは、次を開発会社に確認しましょう。

• 「おかしな呼び出し（認証の失敗が続く、回数が多すぎる）に、どうやって気付きますか。知らせは誰に届きますか」
• 「記録には何が残りますか。パスワードやトークン、個人情報が記録に出ませんか」
• 「WAF を置きますか。どんなルールで、何回までにしますか」
• 「確認用の環境は、使わないときに止められますか。月の費用はいくらの見込みですか」

常に止まらないことが求められる、利用者が多い、記録を長く保管する必要がある、といった条件があると、費用と工数が増えやすくなります。

連載のまとめ

8回にわたって、AI エージェントに仕事を任せるための土台を作ってきました。

共通する考え方は、AI に「お願い」するだけでなく、仕組みで範囲を決める ことです。指示書で進め方を伝え、権限・トークン・WAF で守り、テストでその決まりが崩れていないことを確かめます。AI が書いたコードを人が確認してマージし、本番への反映は人が決める。この分担で、当社は社内の仕事を AI に安心して任せられるようになりました。

この連載の記事一覧

この記事は連載「Claude CodeとAWSで作るAIエージェントの土台」の1回です。連載のほかの回は次のとおりです（連載の一覧ページ）。

• 【Claude CodeとAWSで作るAIエージェントの土台 第1回】なぜリポジトリをやめて、役割ごとの API にしたのか
• 【Claude CodeとAWSで作るAIエージェントの土台 第2回】AWS を4つのアカウントに分け、Google Workspace でログインする
• 【Claude CodeとAWSで作るAIエージェントの土台 第3回】GitHub Actions から AWS へ入る（OIDC）と、sub が番号付きになる落とし穴
• 【Claude CodeとAWSで作るAIエージェントの土台 第4回】ECS Fargate と ALB を NAT なしで小さく始める：証明書・DNS の落とし穴
• 【Claude CodeとAWSで作るAIエージェントの土台 第5回】トークンで役割を判定し、Claude Code の環境に持たせる（Network secrets）
• 【Claude CodeとAWSで作るAIエージェントの土台 第6回】CLAUDE.md は入口だけにする：役割ごとの指示書を API で配る
• 【Claude CodeとAWSで作るAIエージェントの土台 第7回】既存のリポジトリから移す：秘密の値の検査と、WordPress への投稿の中継
• 【Claude CodeとAWSで作るAIエージェントの土台 第8回】守りと費用：401 の監視、WAF の回数制限、確認用はふだん止める（この記事）