第5章 AgentCoreの概要とメイン機能「ランタイム」 — 学習メモ¶

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

一言で¶

Runtime ＝ Strands の Agent を「プロセス内で呼ぶもの」から「認証つきサーバーとして外から叩けるもの」に変える部品。

効いている対比は2つ。

agent("...")（プロセス内呼び出し）vs invoke_agent_runtime(...)（外部からの HTTP 呼び出し）。中身の Strands エージェントは同じでも、Runtime に載せた瞬間に「別プロセス・別システムから叩ける」に変わる。
同じ土台（BedrockAgentCoreApp）の上で "顔"（プロトコル）だけが変わる。人・アプリから直接呼ばせたいなら HTTP、他のエージェントの「ツールの1つ」にしたいなら MCP、他のエージェントの「対話相手そのもの」にしたいなら A2A。呼び出し元が誰かでプロトコルを選ぶだけで、サーバー化の仕組み自体は共通。

全体像¶

ファイルは大きく3グループ。

ハーネス体験版（5.2節・デプロイ不要）: 00_invoke_harness.py
プロトコル別スニペット（app.run() 系・デプロイ前提。README に uv run コマンドが無いのはこのため）: 01_streaming.py（HTTP）／02_runtime.ts（HTTP・TS版）／03_invoke_streaming.py（HTTP 呼び出し側）／04_async.py（非同期タスク）／05_mcp_server.py＋06_mcp_client.py（MCP）／07_a2a_server.py＋08_a2a_client.py（A2A）
ハンズオン一式（5.3節・実際に agentcore create/deploy する対象）: handson/

flowchart TB
    subgraph Snippets["プロトコル別スニペット（app.run系・デプロイ前提）"]
        S1["01_streaming.py<br>HTTPエントリーポイント"]
        S2["05_mcp_server.py<br>MCPサーバー化"]
        S3["07_a2a_server.py<br>A2Aサーバー化"]
    end
    RT["BedrockAgentCoreApp<br>共通のRuntime土台"]
    C1["03_invoke_streaming.py<br>invoke_agent_runtime"]
    C2["06_mcp_client.py<br>MCPClient"]
    C3["08_a2a_client.py<br>A2AClientToolProvider"]

    S1 --> RT
    S2 --> RT
    S3 --> RT
    RT -->|"人・アプリから直接"| C1
    RT -->|"別エージェントのツールとして"| C2
    RT -->|"別エージェント本体として"| C3

非同期タスク（04_async.py）は上記3プロトコルとは別軸で、「HTTP 応答を長時間ブロックしない」ためのパターン。

sequenceDiagram
    participant C as 呼び出し元
    participant EP as invoke関数
    participant Th as バックグラウンドスレッド

    C->>EP: スライド作ってと依頼
    EP->>EP: add_async_task 呼び出し
    EP->>Th: threading.Thread 開始
    EP-->>C: 即時応答 作成を開始しました
    Th->>Th: time.sleep 10秒のダミー処理
    Th->>EP: complete_async_task 呼び出し

ハンズオン（handson/）は上記スニペットの「実際にデプロイする版」。agentcore.json にランタイム定義（entrypoint: main.py, codeLocation: app/MyAgent/）があり、app/MyAgent/main.py が実体。cdk/ は CLI が生成する IaC で、agentcore deploy が内部で使う。

使用ライブラリ・原理¶

BedrockAgentCoreApp（bedrock_agentcore）: エージェントを HTTP サーバーとして待ち受けさせる ASGI 相当のサーバーラッパー。@app.entrypoint が公開エンドポイントを宣言するデコレータ、app.run() が 0.0.0.0:8080 でリッスンを開始する（Runtime のコンテナ契約はこのポートで HTTP を待つこと）。
ジェネレータ返却＝ SSE ストリーミング: 01_streaming.py:16-17 のように async for event in stream: yield event と書くと、エントリーポイント関数がジェネレータになり、Runtime 側がそれを Server-Sent Events としてそのままクライアントへ逐次転送する。return で単発の値を返せば（handson/app/MyAgent/main.py:13）通常の同期レスポンスになる。同じデコレータでも関数の書き方（yield か return か）だけでプロトコルの応答形式が変わる。
非同期タスク API（04_async.py）: app.add_async_task(name) は「時間のかかる処理を開始した」ことを Runtime に通知して task_id を受け取るだけの登録 API。実処理は threading.Thread で裏に逃がし、完了時に app.complete_async_task(task_id) を呼んで完了を通知する。HTTP のタイムアウトに引っかからないよう、エントリーポイント自体は即座に返す設計。
MCP 化（05_mcp_server.py）: FastMCP（MCP 公式 Python SDK のサーバーヘルパー）に @mcp.tool() でツールを登録し、mcp.run(transport="streamable-http") で HTTP ベースの MCP サーバーとして起動する。これを Runtime にデプロイすると「ツールを提供するエージェント」として他のエージェントから呼べるようになる。
A2A 化（07_a2a_server.py）: StrandsA2AExecutor(agent) が Strands の Agent を Google A2A プロトコルのサーバー実装にアダプトし、serve_a2a(...) が Runtime 向けの A2A サーバーを起動する。
呼び出し側の共通パターン: いずれも boto3.client("bedrock-agentcore") 系のクライアントか、対応するプロトコルクライアント（MCPClient / A2AClientToolProvider）で、IAM の SigV4 署名（aws_iam_streamablehttp_client / SigV4HTTPXAuth）を使って認証つきで Runtime のエンドポイントを叩く。Runtime は「デプロイした瞬間に IAM 認証がかかる」ことがコードから読み取れる（呼び出し側が例外なく認証情報を扱っている）。

ファイル別の役割¶

ファイル	役割
`00_invoke_harness.py`	マネコンで作った「ハーネス」を `invoke_harness` で呼び出す（5.2節、CLI デプロイ不要の体験版）
`01_streaming.py`	HTTP エントリーポイント＋ SSE ストリーミングの最小形（Runtime 化の中核例）
`02_runtime.ts`	同じ最小形の TypeScript/Node 版（`@strands-agents/sdk` + `bedrock-agentcore/runtime`、`zod` でリクエストスキーマ検証）
`03_invoke_streaming.py`	デプロイ済み Runtime を `invoke_agent_runtime` で呼び、SSE を1行ずつパースする呼び出し側コード
`04_async.py`	非同期タスク（`add_async_task`/`complete_async_task`）で長時間処理を裏実行するパターン
`05_mcp_server.py`	Runtime を MCP サーバーとして公開（`FastMCP` + `streamable-http`）
`06_mcp_client.py`	デプロイ済み MCP サーバーを IAM 署名つきで呼ぶクライアント（`MCPClient` + `aws_iam_streamablehttp_client`）
`07_a2a_server.py`	Runtime を A2A サーバーとして公開（`StrandsA2AExecutor` + `serve_a2a`）
`08_a2a_client.py`	デプロイ済み A2A サーバーをツールとして Agent に接続するクライアント（`A2AClientToolProvider`）
`handson/agentcore/agentcore.json`	CLI が生成したプロジェクト定義（`runtimes` 配列に `entrypoint`/`codeLocation`/`runtimeVersion` 等）
`handson/app/MyAgent/main.py`	実際にデプロイされるエージェント本体（`01_streaming.py` の非ストリーミング版）
`handson/app/MyAgent/model/load.py`, `mcp_client/client.py`	CLI が吐いた未使用の雛形（`main.py` から import・呼び出しされていない）
`handson/invoke.py`	デプロイ後の Runtime を呼び出す動作確認用スクリプト（`03_invoke_streaming.py` の非ストリーミング版）
`handson/agentcore/cdk/`	CLI が生成する CDK（TypeScript）IaC 一式。`agentcore deploy` が内部で使う

中心コードの読み解き¶

`01_streaming.py`（HTTP エントリーポイント＋ SSE ストリーミング）¶

# chapter5/01_streaming.py:1-21
from strands import Agent
from bedrock_agentcore import BedrockAgentCoreApp

agent = Agent(model="us.anthropic.claude-sonnet-4-6")   # ①
app = BedrockAgentCoreApp()                              # ②

@app.entrypoint                                          # ③
async def invoke(payload, context):
   prompt = payload.get("prompt")                        # ④
   stream = agent.stream_async(prompt)                   # ⑤

   async for event in stream:                            # ⑥
       yield event                                       # ⑦

if __name__ == "__main__":
   app.run()                                              # ⑧

行	やってること	なぜ
① `01_streaming.py:5`	Strands の `Agent` を通常どおり生成	Runtime に載せても「脳」は素の Strands エージェントのまま
② `:6`	`BedrockAgentCoreApp()` でサーバーインスタンスを作成	ここが HTTP を受ける土台になる
③ `:9`	`@app.entrypoint` で公開関数を宣言	外部からの `invoke_agent_runtime` 呼び出しがこの関数にルーティングされる
④ `:12`	`payload.get("prompt")` でリクエスト本文からプロンプトを取り出す	`payload` はクライアントが送った JSON（`invoke.py` の `json.dumps({"prompt": ...})` と対応）
⑤ `:13`	`agent.stream_async(prompt)` で非同期ストリームを取得	Strands 側の非同期 API で、逐次イベントを生成する
⑥⑦ `:16-17`	`async for` で受けて `yield` で右から左に流す	関数がジェネレータになる＝ Runtime が SSE として逐次転送する契約
⑧ `:21`	`app.run()` で `0.0.0.0:8080` を待ち受け	ローカル `uv run` 単体では完結しない（`agentcore dev` かデプロイが前提）

`handson/app/MyAgent/main.py`（実際にデプロイされる非ストリーミング版との対比）¶

# chapter5/handson/app/MyAgent/main.py:9-13
@app.entrypoint
async def invoke(payload, context):
   prompt = payload.get("prompt")
   return agent(prompt)          # yield ではなく return

同じ @app.entrypoint でも return agent(prompt)（agent() は同期呼び出し）にすると、応答はストリーミングではなく単発の JSON になる。呼び出し側も 03_invoke_streaming.py の iter_lines()＋SSE パースではなく、handson/invoke.py:15 の response["response"].read()（一括読み込み）になっている。yield か return かだけで、サーバー側・クライアント側の両方の受け答えが変わる。

学んだこと（要点）¶

ハーネスと Runtime は別物。ハーネス（5.2節）はマネコンだけで作れる簡易呼び出しラッパーで invoke_harness を叩く。Runtime（5.3節）は自分で書いたコードを CLI (agentcore deploy) でデプロイした本番エンドポイントで invoke_agent_runtime を叩く。呼び出し API 名が違う（invoke_harness vs invoke_agent_runtime）ことがコードから確認できる。
3プロトコル（HTTP/MCP/A2A）はどれも同じ BedrockAgentCoreApp 系の土台の上に成り立つ。「誰がこのエージェントを呼ぶか」（人・アプリ／別エージェントのツールとして／別エージェント本体として）で選ぶプロトコルが決まるだけ。
非同期タスクは HTTP タイムアウト対策のパターン。エントリーポイントは即座に「開始しました」を返し、実処理は別スレッドで完了通知（complete_async_task）まで走らせる。
呼び出し側は例外なく IAM 署名を扱っている（aws_iam_streamablehttp_client / SigV4HTTPXAuth）。Runtime にデプロイした時点で認証がかかることの裏返し。

落とし穴・現代版に移植するなら¶

app.run() 系はデプロイ前提。01_streaming.py/04_async.py/05_mcp_server.py/07_a2a_server.py はローカル uv run 単体では完結せず、agentcore dev（ホットリロード）かデプロイ後の呼び出しが必要。README（chapter5/README.md）にもこれら4ファイルの uv run コマンドは書かれていない（5.2節の 00_invoke_harness.py と5.3節のハンズオン CLI 手順のみ）。
プレースホルダー置換必須: 06_mcp_client.py:5 の <MCPサーバーのランタイムARN>、08_a2a_client.py:7 の <A2AサーバーのランタイムARN>、03_invoke_streaming.py:9 のダミー ARN。
handson/app/MyAgent/ の model/load.py・mcp_client/client.py は CLI 生成の未使用雛形。main.py はどちらも import しておらず、load_model()（BedrockModel(model_id="global.anthropic.claude-sonnet-4-5-20250929-v1:0")）も get_streamable_http_mcp_client()（ExaAI の MCP エンドポイント例）も呼ばれていない。読者が「使われてなさそう」で迷わないための注記。
バージョン方針が2系統混在: 章直下の pyproject.toml は strands-agents==1.38.0 / bedrock-agentcore==1.6.4 等を == で厳密ピン留めしているのに対し、CLI が生成した handson/app/MyAgent/pyproject.toml は bedrock-agentcore >= 1.0.3 のように範囲指定（requires-python = ">=3.10" も章全体の >=3.14 と異なる）。著者が手で書いた章のスニペットと、CLI が自動生成したプロジェクトでピン留めの厳密さが違う。
pins: strands-agents==1.38.0 / strands-agents-tools==0.5.1 / bedrock-agentcore==1.6.4 / boto3==1.42.96 / mcp==1.27.0。CLI (@aws/agentcore) だけ例外的に @latest（2026/6/30 更新: CDK アップデート起因のエラー回避のため）。
TS 版は 02_runtime.ts の1本のみ。他ファイルは全て Python で、Node 版があるのはこの最小形だけ。

記事参照¶

書籍第5章（御田／みのるん著）。
関連: ../../lectures/agentcore_basics/STUDY_NOTES.md（AgentCore 8部品の全体像。3-1節が本章に対応）。

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