第15章アンビエントエージェントをCDKで作ろう — 学習メモ¶

書籍「Amazon Bedrock AgentCore実践入門」第15章のサンプルコード（このフォルダ）を読み解いた個人学習メモ。 AgentCore 全体像は ../../lectures/agentcore_basics/STUDY_NOTES.md 参照。実行検証は伴わない。コードに現れた API 名だけ断定し、読めない挙動は「〜と推測」で明示する。

一言で¶

経費精算エージェント＝ S3アップロードをトリガーに動く「アンビエント（環境常駐型）」エージェント。ユーザーがチャットで話しかけるのではなく、イベント（領収書アップロード／承認リンククリック）が2回、非同期にRuntimeを叩く構成をAWS CDK(TypeScript)でIaC化した章。

第13章がRuntime・Memory・Identity・Browser・Observabilityの5機能を1リクエスト内で束ねたのに対し、本章はMemory/Gateway/Identityを使わず、Runtime＋非同期タスク（add_async_task/complete_async_task）という1本の機能を、イベント駆動アーキテクチャの中にどう組み込むかに焦点がある。

全体像¶

flowchart TB
    subgraph Flow1["フロー1: 経費申請"]
        S3up["S3<br>receipts/{user_id}/へアップロード"]
        Invoker["AgentInvokerFunction<br>Lambda(agent_invoker.py)"]
        RT1["AgentCore Runtime<br>expense_agent<br>process_expense()"]
        Async["非同期タスク<br>add_async_task→worker()→complete_async_task"]
        Tools["解析→分類→承認者選定→<br>承認依頼メール送信"]
        DDB["DynamoDB<br>expense-agent-approvals"]
        SNS1["SNS<br>承認者ごとのTopic"]
        Approver["承認者<br>(課長/部長)"]
    end
    subgraph Flow2["フロー2: 承認"]
        Callback["ApprovalCallbackFunction<br>Function URL(approval_callback.py)"]
        RT2["AgentCore Runtime<br>expense_agent<br>process_approval()"]
        Conf["Confluence<br>write_to_confluence"]
        SNS2["SNS<br>申請者への結果通知"]
        Submitter["申請者"]
    end

    S3up -->|"ObjectCreated イベント"| Invoker
    Invoker -->|"invoke_agent_runtime"| RT1
    RT1 --> Async --> Tools
    Tools --> DDB
    Tools --> SNS1 --> Approver
    Approver -->|"承認/却下リンクをクリック"| Callback
    Callback --> DDB
    Callback -->|"list_agent_runtimes→<br>invoke_agent_runtime"| RT2
    RT2 --> Conf
    Callback --> SNS2 --> Submitter

非同期タスクの中身（Lambdaのタイムアウトより先にRuntimeが応答を返し、処理はバックグラウンドで継続する）:

sequenceDiagram
    participant Invoker as AgentInvokerFunction
    participant RT as Runtime(handle_request)
    participant Worker as worker()スレッド
    participant Agent as create_agent()

    Invoker->>RT: invoke_agent_runtime(payload)
    RT->>RT: process_expense(request)
    RT->>RT: task_id = app.add_async_task(...)
    RT->>Worker: threading.Thread(target=worker).start()
    RT-->>Invoker: {"accepted": true, "expense_id": ...} を即座に返す
    Note over Worker,Agent: Lambda呼び出しはここで完了済み
    Worker->>Agent: create_agent()(prompt)
    Agent->>Agent: process_receipt_image→search_classification_info→<br>get_approver_by_amount→send_approval_request
    Worker->>RT: app.complete_async_task(task_id)

使用ライブラリ・原理¶

ライブラリ / API	役割
`strands` (`strands-agents==1.38.0`)	`Agent`・`@tool`。`BedrockModel(model_id=...)`でモデルを明示指定（第13章の文字列指定と違いオブジェクトを渡す）
`strands.Agent(..., structured_output_model=T)`	Pydantic `BaseModel`（`ReceiptInfo`/`ClassificationInfo`）を渡すと、LLM出力を構造化データとして`result.structured_output`に得られる。マルチモーダル解析（画像+テキスト入力）と経費分類のLLM推論フォールバックの両方で使用
`bedrock_agentcore.BedrockAgentCoreApp`	Runtime機能。ただし本章のentrypointは`async def`ではなく同期の`def handle_request(request)`
`app.add_async_task(name, metadata)` / `app.complete_async_task(task_id)`	非同期タスクAPI。呼び出し元（Lambda）へは即座に応答を返しつつ、Runtimeコンテナ側には「まだ仕事が終わっていない」ことを伝え続ける仕組み（詳細挙動はコードから読めないため「タスク未完了をRuntimeに伝え、コンテナの早期終了を防ぐと推測」）
`threading.Thread(target=worker, daemon=True)`	素朴なバックグラウンド実行。第13章の`asyncio`ベースとは対照的に、本章は同期関数の中でスレッドを立てて非同期性を作る
`boto3` (`s3`/`sns`/`dynamodb`/`bedrock-agentcore`/`bedrock-agentcore-control`)	全ツール・Lambdaの実行基盤。`bedrock-agentcore`は`invoke_agent_runtime`、`bedrock-agentcore-control`は`list_agent_runtimes`（Runtime名→ARN逆引き）
AWS CDK (TypeScript) + `@aws-cdk/aws-bedrock-agentcore-alpha`	`agentcore.Runtime`コンストラクトでECRイメージからRuntimeを直接CDK定義。本章はAgentCore CLIを使わずCDKのみでIaC化する点が第13章と対照的
`@cdklabs/deploy-time-build` `ContainerImageBuild`	デプロイ時にARM64のDockerイメージをビルドしてECRにpush。ローカルのDocker環境やCI用ビルドパイプラインが不要
`urllib.request`（標準ライブラリ）	Confluence REST APIへのPOSTをBasic認証で素朴に実装（外部HTTPライブラリ不使用）

ファイル別の役割¶

ファイル	役割
`src/agent.py`	Runtimeのエントリポイント。`handle_request`が`action`の有無で`process_expense`/`process_approval`に分岐。5つの`@tool`（`process_receipt_image`/`search_classification_info`/`get_approver_by_amount`/`send_approval_request`/`write_to_confluence`）を定義
`src/lambda/agent_invoker/agent_invoker.py`	S3の`ObjectCreated`イベントを受け、S3キーから`user_id`を抽出しユーザー情報をS3から引いて`invoke_agent_runtime`を呼ぶ
`src/lambda/approval_callback/approval_callback.py`	承認者がメール内リンクをクリックした際のFunction URLハンドラー。DynamoDB更新→`list_agent_runtimes`でARN逆引き→再度Runtime呼び出し→申請者へSNS通知
`cdk/lib/expense-agent-stack.ts`	S3・DynamoDB・SNS（ユーザーごとのTopic）・2つのLambda・IAMロール・AgentCore Runtimeを1つのCDK Stackとして定義
`cdk/bin/app.ts`	CDK Appのエントリポイント。`ExpenseAgentStack`をリージョン`us-east-1`（デフォルト）でインスタンス化
`docker/Dockerfile`	ARM64ベースのRuntimeコンテナ。`uv pip install .`で依存解決し`python -m src.agent`で起動（第13章と異なりADOT自動計装は無し）
`pyproject.toml`	依存パッケージ定義（`strands-agents==1.38.0`/`bedrock-agentcore==1.6.4`/`boto3==1.42.96`）
`data/users.json`	ユーザーマスタ（user_id/name/email/role）。S3にデプロイされ、Lambdaとエージェントの両方から読まれる
`data/classification_rules.json`	社内の経費分類ルール（ベンダー名キーワード→カテゴリ）。マッチしなければLLM推論にフォールバック
`.env.example`	CDKデプロイ時に読み込む環境変数のテンプレ（Confluence接続情報・Bedrockモデル ID）

中心コードの読み解き¶

src/agent.py:320-330 — 非同期タスクの起動から即時応答までの核心部分:

    # 非同期タスクとして実行
    task_id = app.add_async_task(                       # ①
        "expense_processing", {"expense_id": expense_id})
    def worker():                                        # ②
        try:
            create_agent()(prompt)                       # ③
        finally:
            app.complete_async_task(task_id)              # ④
    threading.Thread(target=worker, daemon=True).start()  # ⑤

    return {"accepted": True, "expense_id": expense_id}   # ⑥

行	やってること	なぜ
① `app.add_async_task(...)`	非同期タスクを登録し`task_id`を得る	Runtime側に「まだ処理中のタスクがある」ことを伝える。応答を先に返しても、このタスクが完了するまでRuntimeが実行中とみなす仕組みと推測
② `def worker():`	バックグラウンドで実行する関数を定義	領収書解析〜承認依頼メール送信までの一連のツール呼び出しは数秒〜数十秒かかりうるため、Lambda呼び出し元をブロックしたくない
③ `create_agent()(prompt)`	Strands Agentを生成し、経費処理プロンプトを渡して実行	5つの`@tool`（画像解析→分類→承認者選定→メール送信）を1回のエージェント呼び出しで連鎖実行させる
④ `app.complete_async_task(task_id)`	`finally`節で必ず実行	エージェント実行が例外で終わっても完了通知は必ず送る。タスクが「未完了のまま放置」される事態を防ぐ
⑤ `threading.Thread(..., daemon=True).start()`	ワーカーを別スレッドで起動し即座に制御を返す	呼び出し元の`handle_request`はこの行の直後で`return`できる。「非同期」を`asyncio`ではなく素朴な`threading`で実現している点が本章の特徴
⑥ `return {"accepted": True, ...}`	エージェント処理の完了を待たずに即時応答	Lambda（`agent_invoker.py`）はこのレスポンスを受け取った時点で自分の実行を終えられる。「アンビエントエージェント」＝ユーザーの応答待ちを介さず環境変化に反応して裏で動く、という本章のコンセプトそのもの

学んだこと（要点）¶

「アンビエントエージェント」＝チャットではなくイベントで駆動するエージェント。本章のRuntimeは会話履歴もセッションも持たず、handle_requestが受け取るrequestの中身（actionキーの有無）だけで「新規申請」か「承認結果」かを判定する、ステートレスなイベントハンドラとして設計されている。
AgentCoreの非同期タスクAPIは、Lambdaのように実行時間に制約のある呼び出し元と、時間のかかるエージェント処理を切り離すための仕組み。add_async_task/complete_async_taskのペアで「今はまだ仕事中」を宣言・解除する、という単純な契約になっている。
CDKだけでRuntimeまで定義できる（agentcore.Runtimeコンストラクト）。第13章のagentcore create→agentcore.json→CLIデプロイという「AgentCore CLI中心」のワークフローとは別に、生のCDKアプリにAgentCore L2/L3コンストラクトを混ぜるという選択肢がある。IaCを一元管理したいチームにはこちらが向く。
ContainerImageBuildはデプロイ時ビルドという発想。イメージのビルド自体をcdk deploy実行時にCodeBuild等へ委譲する（@cdklabs/deploy-time-build）ため、ローカルにDockerが無くても、あるいはCI環境を分けなくてもCDKだけでコンテナデプロイが完結する。
承認コールバックはRuntime名→ARNの逆引きが必要。承認Lambda（approval_callback.py）はデプロイ時に静的なRuntimeのARNを知らない（Function作成時点ではRuntimeがまだ無い/循環参照になるため？）ので、list_agent_runtimes()で名前からARNを都度引く設計になっている。
ユーザーごとにSNS Topicを動的生成する設計（CDK側でemailAddresses.forEach）。承認者と申請者への通知経路を、メールアドレスごとに独立したTopicとSubscriptionで表現しており、通知の追加・削除がユーザーマスタの変更だけで完結する。

落とし穴・デプロイ前提¶

.env（CONFLUENCE_URL/CONFLUENCE_EMAIL/CONFLUENCE_API_TOKEN/CONFLUENCE_SPACE_KEY/BEDROCK_MODEL_ID）の置換が必須。.env.exampleはプレースホルダのみで、実際の値は各自のConfluence環境に合わせて用意する。
data/users.jsonのメールアドレスはプレースホルダ（your-email+1@gmail.com等）。実際に動かす際は自分のメールアドレス（+1/+2/+3のGmailエイリアス等）に置き換える前提。承認者ごとのSNS Topicはこのメールアドレスから自動生成される。
SNSの初回サブスクリプションはメールでの確認（confirm）が必要。aws sns list-subscriptions-by-topicで状態確認、未確認ならaws sns subscribe→受信メールのリンク、またはCLIでaws sns confirm-subscription --token <トークン>を使う。
docker/DockerfileはARM64固定（ecrAssets.Platform.LINUX_ARM64とCDK側で明示）。x86イメージを別途用意している場合は動かない。
デプロイ手順はnpx cdk bootstrap→npx cdk deploy（AgentCore CLIは使わない）。cdk destroy実行時、S3バケットとDynamoDBテーブルはRemovalPolicy.DESTROY＋autoDeleteObjects: trueが設定されており、中身ごと削除される（本番運用では要注意の設定）。
動作確認はaws s3 cpで領収書画像をアップロードするだけ。data/receipt_rule_under100k.png等4パターン（社内ルール有無 × 金額閾値10万円の上下）がサンプルとして用意されている。
Memory/Gateway/Identityは本章では未使用。Observabilityも専用の計装コードは無く、IAM側でCloudWatch Logsへの書き込み権限（logs:PutLogEvents等）を渡しているだけ（ADOT等の自動計装は入っていない）。
bedrock-agentcore==1.6.4固定。書籍執筆時点のバージョンで、agent.pyのadd_async_task/complete_async_taskのシグネチャがこのバージョンに依存する。

記事参照¶

書籍第15章「【ハンズオン】アンビエントエージェントをCDKで作ろう」。
関連: ../../lectures/agentcore_basics/STUDY_NOTES.md（AgentCore全体像・第13章との機能比較表「4. 8機能が実アプリでどう組み合わさるか」）。
関連章: 第13章（../chapter13/STUDY_NOTES.md）は同じRuntime機能をasyncioベースの対話型フローで使っており、本章のthreadingベース非同期処理と対比すると設計の違いが掴みやすい。

作成: 2026-07-17 ／最終更新: 2026-07-17

第15章 アンビエントエージェントをCDKで作ろう — 学習メモ¶