技術的負債の蓄積しやすい AI エージェントの開発
ここで「技術的負債」とは、ベストプラクティスが日々更新されることで開発時点とリリース時点で実装の洗練度合いに無視できないギャップが生まれる状況を指します。ご存じの通り、生成 AI はこの意味での「技術的負債」が蓄積しやすい領域です。
みなさんは AI エージェントを開発されたことはあるでしょうか? 今から開発するなら、LangChain、LangGrah、CrewAI、また Mastra など新旧様々なフレームワークがあります。登場した順番は必ずしも洗練また主流になることを保証しておらず、GitHub Star を一つの指標とすると後発のフレームワークが急激に普及していることがわかります。
AI エージェントの本番利用が進むにつれて、セキュリティや Observability も関心が高まっているトピックです。機械学習のプロダクトに関わった方は 機械学習システムにおける隠れた技術的負債 について聞いたことがあるかもしれませんが、AI エージェントの動作範囲や権限が広がるにつれ、これらの周辺機能の重要性と必要性が無視できなくなってきます。それにより開発者として AI エージェントのコア機能の実装に集中することが難しくなる「隠れた技術的負債」も増えてくると予測しています。
Gartner は、2028 年までに日常業務における意思決定のうち少なくとも 15% が AI エージェントにより「自律的に」行われるようになると予測しています。5 回に 1 回ぐらいは AI の判断をもとに進めるとするとなかなかの割合です。今後 AI エージェントの利用が普及すると負債の蓄積スピードは無視できない規模になる可能性があります。
実装に依存しない「技術的負債を予防する」ランタイム環境の重要性
こうした状況の中発表された Amazon Bedrock AgentCore は、実装に依存しないランタイム環境と、本番環境で動く AI エージェントの実装に不可欠な認証認可・Observability、メモリ管理といったコンポーネントを提供します。
※執筆時点 Preview での公開となり、記載内容は変更される可能性があります※
あなたの手元に AI エージェントがあれば、それを Docker コンテナにまとめることで「AI エージェントのデプロイ先」である AgentCore Runtime で動かすことが出来ます。実装に使っているフレームワークや言語は問わず、必要な要件は起動用の entrypoint と死活監視用の ping を HTTP で公開していることのみです。そのため、TypeScript の Mastra も動かすことが出来ます (参考 : MastraをBedrock AgentCore Runtimeにデプロイ！) 。この実装の非依存性とシンプルなインタフェースはフレームワークの切り替えを容易にします。実行はもちろんサーバーレスで、しかもモデルの応答待ち時間は課金対象になりません (参考 : Consumption-based pricing)。
さらに、将来の「隠れた技術的負債」になりうる安全性を担保するための監視・認証・アクセス制限、また機能拡張に欠かせない記憶・コード実行・ブラウザ実行といったコンポーネントを提供しています。このコンポーネントは AWS らしい "Building Block" として簡単につけ外しでき、どんなフレームワークで実装していても取り付けができます。これにより AI エージェントの実装に集中し続けることが出来ます。
Amazon Bedrock AgentCore は、実装非依存のランタイムとコンポーネントを提供することで「技術的負債を予防」し、AI エージェントそのものの開発に集中し続けるために役立つサービスと言えます。
Amazon Bedrock AgentCore の全機能を体験する
前置きが長くなったので、早速「ゼロからまるっと」 Amazon Bedrock AgentCore の機能を体験してみましょう！本記事では、皆さんが AI エージェントの開発を実際に行う時に流用しやすい実装を目指して紹介しています。記事執筆時点で公開されている 6 機能の体験ツアーのお品書きは次の通りです。
- 🧮 : AWS の見積りを「計算」するエージェントを作成する : AgentCore Code Interpreter
- 🚀 : クラウド上に AI エージェントをデプロイする : AgentCore Runtime
- 🛡️ : AI エージェントの利用に制限をかけて公開する : AgentCore Gateway
- 👤 : 認証認可により権限を取得し管理する : AgentCore Identity
- 📊 : AI エージェントの動作をモニタリングする : AgentCore Observability
- 🧠 : 見積の内容を「記憶」する : AgentCore Memory
実装は sample-amazon-bedrock-agentcore-onboarding で公開しています。
事前準備として、リポジトリをセットアップします。Python の環境構築に git と uv を使用しているため、未インストール場合は公式を参考にインストールください。また、Amazon Bedrock の Claude モデルの有効化が必要です (実装に使っている Strands Agents が us-west-2 をデフォルトで使うのでそちらと皆さんのデフォルトリージョン (サンプルでは us-east-1) を念のため有効化頂ければと)。
- 1.5 Getting Started - Installing Git
- Installing uv
- Amazon Bedrock のモデルの有効化
-
サンプルのリポジトリを手元で設定するには、任意のディレクトリで下記のコマンドを実行すれば準備は完了です。
git clone https://github.com/icoxfog417/sample-amazon-bedrock-agentcore-onboarding.git
uv sync
では、はじめて行きましょう！
🧮 AWS の見積りを「計算」するエージェントを作成する : AgentCore Code Interpreter
はじめに、「実装非依存」であることを体感いただくために AgentCore Code Interpreter を紹介します。計算するエージェントの例として、"AWS のコスト見積もり" を行うエージェントを作成します。仕様は次の通りです。Cost Estimator Agent が、LLM を使用しユーザーのアーキテクチャ解釈した後 (Bedrock は省略しています)、AWS Pricing MCP Server で価格を取得、AgentCore Code Interpreter で計算を行い結果を返します。このサンプルは、多くの業務で求められる「データを集めて」「計算する」過程を MCP と CodeInterpreter で行っており応用の幅が広いと考えています。
今回、AI エージェントの実装は Strands Agents で行いました。ただ、どのフレームワークでも CodeInterpreter の実装に変化はありません。実際に見ていきましょう。
体験手順
サンプルのディレクトリの構成は次のようになっています。
01_code_interpreter/
├── README.md # This documentation
├── cost_estimator_agent/
│ ├── __init__.py # Package initialization
│ ├── config.py # Configuration and prompts
│ └── cost_estimator_agent.py # Main agent implementation
└── test_cost_estimator_agent.py # Test suite
Step 1: AWS コスト見積りエージェントの実行
エージェントは次のコマンドで実行できます。
cd 01_code_interpreter
uv run python test_cost_estimator_agent.py --architecture "パン屋のWebサイトをHTMLでなるべく安く作りたい"
実行結果のサンプルは以下の通りです、
# パン屋のためのAWS静的ウェブサイト コスト分析
## アーキテクチャ概要
最も費用対効果の高いパン屋のウェブサイトを構築するために、以下のシンプルなアーキテクチャを提案します：
1. **Amazon S3**：静的HTMLファイルのホスティング
2. **Amazon Route 53**：（オプション）ドメイン名管理とDNS
このアーキテクチャは、HTMLで作成した静的ウェブサイトを最小限のコストで運用できるよう設計されています。
## コスト内訳（月額）
### 基本プラン（ドメイン名なし）
...
Step 2: Streaming で試す
--tests streaming のオプションをつけることで Streaming で処理します (見積り結果に影響はないですが。。。)
uv run python test_cost_estimator_agent.py --architecture "パン屋のWebサイトをHTMLでなるべく安く作りたい" --tests streaming
実装の解説
CodeInterpreter は、start で開始して stop で停止する単純明快な実装です。
```python
from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter
def _setup_code_interpreter(self) -> None:
"""Setup AgentCore Code Interpreter for secure calculations"""
try:
logger.info("Setting up AgentCore Code Interpreter...")
self.code_interpreter = CodeInterpreter(self.region)
self.code_interpreter.start()
logger.info("✅ AgentCore Code Interpreter session started successfully")
except Exception as e:
logger.exception(f"❌ Failed to setup Code Interpreter: {e}")
raise
- 環境の作成 :
self.code_interpreter = CodeInterpreter(self.region) - 起動 :
self.code_interpreter.start() - コード実行 :
self.code_interpreter.invok("executeCode", {"language": "python", "code": calculation_code}) - 停止 :
self.code_interpreter.stop()
stop し忘れないよう、@contextmanager で実装しています。
@contextmanager
def _estimation_agent(self) -> Generator[Agent, None, None]:
"""Context manager for cost estimation components"""
try:
# Setup components in order
self._setup_code_interpreter()
aws_pricing_client = self._setup_aws_pricing_client()
# Create agent with both execute_cost_calculation and MCP pricing tools
with aws_pricing_client:
pricing_tools = aws_pricing_client.list_tools_sync()
all_tools = [self.execute_cost_calculation] + pricing_tools
agent = Agent(
model=DEFAULT_MODEL,
tools=all_tools,
system_prompt=SYSTEM_PROMPT
)
yield agent
finally:
# Ensure cleanup happens regardless of success/failure
self.cleanup()
Streaming で処理する際は、{"data": content} の形式で来るのでこれを処理する必要があります。前回の出力と重複しないようチェックを入れています。
async def estimate_costs_stream(self, architecture_description: str) -> AsyncGenerator[dict, None]:
"""Stream cost estimation with proper delta handling"""
try:
with self._estimation_agent() as agent:
# Implement proper delta handling to prevent duplicates
previous_output = ""
async for event in agent.stream_async(prompt, callback_handler=null_callback_handler):
if "data" in event:
current_chunk = str(event["data"])
# Handle delta calculation following Bedrock best practices
if current_chunk.startswith(previous_output):
# Extract only the new part
delta_content = current_chunk[len(previous_output):]
if delta_content:
previous_output = current_chunk
yield {"data": delta_content}
except Exception as e:
yield {"error": True, "data": f"❌ Streaming failed: {e}"}
実装するフレームワークが何であれ、start と stop で停止する環境にコードを入れれば良いことがわかります。つまり、たとえ紹介した Strands Agents で実装していなくても、まして AWS 上でホスティングしていなくても AgentCore Code Interpreter は安全にコードを実行できるセキュアなサンドボックスとして普通に使うことができます。特に、Claude Desktop のようなクライアント環境で動くソフトウェアではローカルのリソースを消費せず端末側の安全性を確実にする手段として適していると思います。Code Interpreter 以外に AgentCore Browser が提供されており、こちらも隔離されたブラウザ環境が利用できローカルのブラウザが不正に操作されるリスクを回避できます。
本セクションのまとめ
- AgentCore は AWS に何かデプロイしなくても使える！ : セキュアなコード実行環境を提供する Code Interpreter、ブラウザ実行環境を提供する Browser はローカル/クラウド関わらず AI エージェントの処理を安全に行うのに役立つ
🚀 : クラウド上に AI エージェントをデプロイする : AgentCore Runtime
あなたがシステム開発をしていれば、Web サイト上に概算見積りをしてくれる AI エージェントを設置することでお客様が自分自身で予算感に見合ったアーキテクチャをセルフサービスで探索できるようになるかもしれません。このように AI エージェントをサービスに組み込む場合、あるいは開発チームでエージェントを共有する場合クラウド環境にデプロイしたくなるでしょう。
AgentCore Runtimeは、AI エージェントを安全かつスケーラブルにホスティングするためのサービスで次の 4 つの特徴があります。
- Serverless : 稼働時間での課金で、「稼働時間」には LLM の応答待ち時間は含まない
- Isolated : 各 AI エージェントのセッションは専用の microVM により隔離されている (詳細 : Use isolated sessions for agents)
- Long Running : 15 分の idle or 8 時間の稼働上限まで処理を継続できる
- Framework / Language Agnostic : AI エージェントの実装言語やフレームワークを問わない
実体はコンテナをホスティングする形式をとっています。コンテナの中身は、ドキュメント からわかる通り FastAPI 等で作成した API サーバーをコンテナに固めていることを想定しています。期待する API が実装されていればどんな方式で実装されていても良く、このため "実装非依存・言語非依存" となっています。コンテナは Amazon Elastic Container Registry (ECR) に登録し、AgentCore Runtime へ紐付けを行います。
体験手順
サンプルのディレクトリの構成は次のようになっています。
02_runtime/
├── README.md # This documentation
├── prepare_agent.py # Agent preparation tool
└── deployment/ # Generated deployment directory
├── invoke.py # Runtime entrypoint
├── requirements.txt # Dependencies
└── cost_estimator_agent/ # Copied source files
デプロイに当たっては、先ほど作成したコスト見積もりのエージェントをコピーしてきて使います(cost_estimator_agent/)。ローカルで開発していたエージェントを持ってきて、Runtime に上げるイメージでサンプルを作成しています。全体の流れは次の通りです。agentcore CLI は bedrock-agentcore-starter-toolkit から使うことが出来ます。
では、実際に試して見ましょう。
Step 1: AI エージェントの準備を行う
cd 02_runtime
uv run prepare_agent.py prepare --source-dir ../01_code_interpreter/cost_estimator_agent
こちらのコマンドで、AI エージェントのフォルダの準備と Runtime を動かすために必要な IAM Role の作成を行っています。Runtime に必要な IAM ロールはドキュメントに記載されている とおりで、今回は CodeInterpreter の実行権限も加えています。
Step 2: 生成された agentcore configure のコマンドを実行
prepare_agent.py を実行すると agentcore のコマンドが作成されるので 3 つ順番に実行します。
※下記の出力例は、アカウント id などをダミーに置き換えているので注意してください。
# Configure the agent runtime
agentcore configure --entrypoint deployment/invoke.py --name cost_estimator_agent --execution-role arn:aws:iam::000000000000:role/AgentCoreRole-cost_estimator_agent --requirements-file deployment/requirements.txt --region us-east-1
# Launch the agent
agentcore launch
# Test your agent
agentcore invoke '{"prompt": "I would like to prepare small EC2 for ssh. How much does it cost?"}'
agentcore launch の実行には、Docker と CPU が ARM アーキテクチャの環境が必要です。これらがない場合、--codebuild のオプションをつけることで AWS CodeBuild 経由でのデプロイが出来ます
最後 invoke で Runtime 上のエージェントを起動していますが、AWS Console からも実行できます。まず Runtime 環境に AWS Console からアクセスします。
DEFALUT を選択し、"Test Endpoint" から invoke で入力したのと同様にテスト入力を行うことで出力が確認できます。
invoke で送る JSON データの形式は entrypoint の実装の影響を受けます。今回は invoke.py が payload.get("prompt") でパラメータを取得しているため { "prompt" : "XXXX"} の形式で送付しています。
実装の解説
02_runtime/deployment/invoke.py が Runtime から起動されるファイルになります。今回は受け取った payload を実装済みのエージェントに渡すだけという非常にシンプルな実装になっています。
# deployment/invoke.py
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from cost_estimator_agent.cost_estimator_agent import AWSCostEstimatorAgent
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@app.entrypoint
def invoke(payload):
user_input = payload.get("prompt")
agent = AWSCostEstimatorAgent()
return agent.estimate_costs(user_input)
if __name__ == "__main__":
app.run()
非同期の場合はこちらです。async になっているところ以外は大差ありません。
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from cost_estimator_agent.cost_estimator_agent import AWSCostEstimatorAgent
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@app.entrypoint
async def invoke(payload):
user_input = payload.get("prompt")
agent = AWSCostEstimatorAgent()
stream = agent.estimate_costs_stream(user_input)
async for event in stream:
yield (event)
if __name__ == "__main__":
app.run()
依存関係は requirements.tx で定義しており、AI エージェントの実装以外に必要なファイルは 2 つだけです。
-
requirements.txtにはuvが必要です。これは AWS Pricing MCP Server をuvxで使用するためです - AWS Pricing MCP Server を利用するには AWS Profile が必要です。AgentCore Runtime 上には
defaultのプロファイルがないので、docker の設定を参考に AWS STS で ACCESS_KEY / SECRET_ACCESS_KEY / SESSION_TOKEN を発行して接続しています(01_code_interpreter/cost_estimator_agent._get_aws_credentialsで実装)
Runtime にデプロイする場合、最小限の実装で同期、非同期いずれにも対応できることがわかりました。
本セクションのまとめ
- AgentCore Runtime により継続的かつ安全なコンテキストが維持できる : いままで AI エージェントをデプロイする先は AWS Fargate / Amazon ECS や AWS Lambda がありましたが、完全サーバーレスの実現や Long Running な実行に課題がありました。 AgentCore Runtime が登場したことで、AI エージェントとのインタラクティブなやり取りで必要になるセキュアかつ継続的な実行環境が利用できるようになりました
🛡️ : AI エージェントの利用に制限をかけて公開する : AgentCore Gateway
AgentCore Gatewayは AI エージェントを安全に公開するための文字通り「玄関口」を提供します。具体的には、サーバーサイドで MCP ツール等にアクセスする際、認証済みである証拠 (Bearer Token) を要求することが出来ます。これにより、LiteLLM などが提供する 1) MCP Server に対する単一のエンドポイント 2) MCP サーバーアクセス時の認証機能を Managed で実装できます。もちろん先行する OSS が高機能・多機能である面はありますが、必要な機能を満たす場合サーバーレスのコスト効率を得ることが出来ます。
AgentCore Gateway には Inbound のチェックに使用する OAuth 認証認可 、通過したリクエストをパスする Outbound のターゲットを設定します。ターゲットとして AWS Lambda、また OpenAPI や Smithy を設定できます。3rd Party の提供する MCP や Agent を認証経由で利用したい場合、Gateway を通じ統合することで Internal / External 共通のエンドポイントを作成できる構成になっています。
体験手順
実際に試して見ましょう。先ほど Runtime にデプロイしたコスト見積もりエージェントを呼び出す AWS Lambda を作成し、それを Gateway の Outbound ターゲットとして登録します。全体の流れは次の通りになります。Gateway に対する Inbound は Cognito を Authorizer として使用し、Cognito 側で OAuth プロセスを実施し得られる token を Gateway 通過時に要求する構成としています。
フォルダの構成は次の通りです。src/app.py が先ほどデプロイした Runtime を呼び出す AWS Lambda で、今回 AWS Serverless Application Model （AWS SAM) で構築しました。
-
03_gatewayを進める際は、AWS SAM のインストール を行い進めてください
03_gateway/
├── README.md # This documentation
├── template.yaml # SAM template for Lambda
├── src/app.py # Lambda function implementation
├── deploy.sh # Lambda deployment script
├── create_gateway.py # Gateway setup with Cognito
├── test_gateway.py # Gateway testing suite
└── gateway_config.json # Generated configuration
Step 1: AWS Lambda をデプロイする
cd 03_gateway
./deploy.sh
デプロイした AWS Lambda の arn は gateway_config.json に記録され後続のコマンドで参照するようにしています。
Step 2: AgentCore Gateway を Cognito を Authorizer として登録し構築
cd 03_gateway
uv run create_gateway.py
Step 3: Gateway 経由でリクエストを送る
Gateway は提供するツール / エージェントの List と呼び出しを MCP の方式でサポートします。今回は Strands Agents で作成した Agent からツール (MCP) として呼び出す場合と直接エンドポイントに jsonrpc 形式でリクエストを送る方式の 2 つが実行できます。
cd 03_gateway
# Test with MCP protocol
uv run python test_gateway.py --tests mcp --architecture "Amazon Translateで100ページの議事録翻訳"
# Test with direct API calls
uv run python test_gateway.py --tests api --architecture "Amazon Translateで100ページの議事録翻訳"
レスポンスが長いと、Gateway Time-out (504) が発生する場合があります。これは後述する応答時間制限のためです。[quick] をつけると MCP サーバーにアクセスせず概算を出力するようプロンプトを調整しているので Time-out が発生する場合試して見てください。
実装の解説
Gateway の重要な要素である Inbound の認証と Outbound の実行主体について解説します。
Inbound の OAuth
03_gateway/create_gateway.py で OAuth を行う Cognito とそれを利用した Gateway を作成しています。実装は bedrock_agentcore_starter_toolkit の GatewayClient で簡単に行うことが出来ます。
from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
...
client = GatewayClient(region_name=boto3.Session().region_name)
...
cognito_result = client.create_oauth_authorizer_with_cognito("AWSCostEstimationResourceServer")
cognito_config = {
"cognito": {
"client_id": cognito_result['client_info']['client_id'],
"client_secret": cognito_result['client_info']['client_secret'],
"token_endpoint": cognito_result['client_info']['token_endpoint'],
"scope": cognito_result['client_info']['scope'],
"user_pool_id": cognito_result['client_info']['user_pool_id']
}
}
...
gateway = client.create_mcp_gateway(
name="AWSCostEstimationGateway",
role_arn=None,
authorizer_config=cognito_result["authorizer_config"],
enable_semantic_search=False
)
-
create_oauth_authorizer_with_cognitoで作成される Cognito はclient_credentialsが指定されており、本来は M2M 、システム間同士の接続に利用されます。そのため、一般的な画面が遷移して認証する形式の OAuth (code) を使用する場合は使えません
Outbound の AWS Lambda
AWS SAM で AWS Lambda を実装しており、03_gateway/src/app.py で invoke_agent_runtime を実行しています。
# Initialize AgentCore client
client = boto3.client('bedrock-agentcore')
# Prepare the payload for cost estimation
payload = {
"prompt": architecture_description
}
# Generate session ID for this request
session_id = str(uuid.uuid4())
logger.info(f"Invoking AgentCore Runtime with session: {session_id}")
# Invoke the runtime
# Explicitly set the traceId to avoid `Value at 'traceId' failed to satisfy constraint: Member must have length less than or equal to 128\` error
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode('utf-8'),
traceId=session_id,
)
- 本記事執筆時点の注意点として、
traceIdを明示的に指定しないとValue at 'traceId' failed to satisfy constraint: Member must have length less than or equal to 128が発生する場合があります - AWS Lambda の実行ロールには
bedrock-agentcore:InvokeAgentRuntimeの許可が必要です。詳細は 03_gateway/template.yamlを参照ください
AWS Lambda を作成したら、Gateway への登録を行います。03_gateway/create_gateway.py にて実装をしています。
tool_schema = [
{
"name": "aws_cost_estimation",
"description": "Estimate AWS costs for a given architecture description",
"inputSchema": {
"type": "object",
"properties": {
"architecture_description": {
"type": "string",
"description": "Description of the AWS architecture to estimate costs for"
}
},
"required": ["architecture_description"]
}
}
]
# Create lambda target with required credentialProviderConfigurations
# Note: toolkit's create_mcp_gateway_target doesn't handle custom target_payload + credentials
# Reference: https://github.com/aws/bedrock-agentcore-starter-toolkit/pull/57
target_name = f"AWSCostEstimationLambdaTarget"
create_request = {
"gatewayIdentifier": gateway["gatewayId"],
"name": target_name,
"targetConfiguration": {
"mcp": {
"lambda": {
"lambdaArn": lambda_arn,
"toolSchema": {
"inlinePayload": tool_schema
}
}
}
},
"credentialProviderConfigurations": [{"credentialProviderType": "GATEWAY_IAM_ROLE"}]
}
logger.info("Creating Lambda target with custom schema and credentials...")
logger.info(f"Request: {create_request}")
# Use boto3 client directly since toolkit method doesn't support this combination
bedrock_client = client.session.client('bedrock-agentcore-control')
target_response = bedrock_client.create_gateway_target(**create_request)
- 本当は
bedrock-agentcore-starter-toolkitのcreate_mcp_gateway_targetを使うと楽なのですが、執筆時点では credential を指定できない不具合があり対応待ちとなっています。
Gateway へのアクセス
Agent から Gateway にアクセスします。Gateway には登録されている Outbound のツールの一覧を提供する Listing の機能があり、今回は登録済みのはずの aws_cost_estimation がないかチェックしてから実行しています。
def test_with_mcp_client(gateway_url, token):
"""Test the Gateway using MCP client via Strands Agents"""
def create_streamable_http_transport():
return streamablehttp_client(
gateway_url,
headers={"Authorization": f"Bearer {token}"}
)
# Create MCP client with authentication
mcp_client = MCPClient(create_streamable_http_transport)
with mcp_client:
# List available tools
tools = get_full_tools_list(mcp_client)
tool_names = [tool.tool_name for tool in tools]
# Find and invoke cost estimation tool
for tool in tools:
if 'aws_cost_estimation' in tool.tool_name:
result = mcp_client.call_tool_sync(
tool_name=tool.tool_name,
arguments={"architecture_description": "A web application with ALB + 2x EC2 t3.medium"}
)
return result.content
肝心の Gateway を通過するための "Authorization": f"Bearer {token}" は get_oauth_token で取得しています。GatewayClient の get_access_token_for_cognito を使用することで得られます。
def get_oauth_token(config):
"""Get OAuth token from Cognito using bedrock_agentcore_starter_toolkit"""
logger.info("Getting OAuth token from Cognito...")
# Create GatewayClient and use its method to get access token
gateway_client = GatewayClient()
# Prepare client_info in the format expected by the method
client_info = {
'client_id': config['cognito']['client_id'],
'client_secret': config['cognito']['client_secret'],
'scope': config['cognito']['scope'],
'token_endpoint': config['cognito']['token_endpoint']
}
token = gateway_client.get_access_token_for_cognito(client_info)
logger.info("Successfully obtained OAuth token")
return token
Gateway を使用することで、AWS Lambda は必要ですが Amazon API Gateway なしに Agent や MCP を安全に公開できます。Gateway はタイムアウト上限が 55 秒で (Timeout for a gateway invocation)、API Gateway のデフォルト 30 秒ほどより長く、なにより Streamable HTTP に対応している点が最大のメリットとなります。ただ、価格は API Gateway の \$3.5/100 万 (REST API) に比べて \$5/100 万 (API Invocations) となり若干割高です (※価格は 2025/7/27 時点のもので、AgentCore は Preview 中である点にご注意ください)。
本セクションのまとめ
- AgentCore Gateway により安全に Agent/MCP を公開するエンドポイントが作成できる : Inbound に OAuth の認証認可を事前に要求するアクセス制限をかけることでセキュアに、かつ Outboud に AWS Lambda や OpenAPI 等を設定することで API Gateway 等なしに Streamable HTTP で接続できるエンドポイントを公開することが出来ます
👤 : 認証認可により権限を取得し管理する : AgentCore Identity
AgentCore Identityはいわば "AI Agent のための Amazon Cognito" です。AgentCore Identity はユーザーもしくは AI エージェント自身による認証認可プロセスと頻繁な再認証を抑止する認証結果のセキュアな保管機能を提供します。具体的には、OAuth の認証認可プロセスと取得した Token を安全に保管する仕組みです (OAuth 以外に、API キーの発行・補完をサポートしています : Creating an API key credential provider)。Token の暗号化やリフレッシュをサポートしており (Features of AgentCore Identity)、セキュアな運用を効率化できます。
先ほど構築した Gateway は OAuth 認証で得られる Token をアクセスに使用していましたが、Gateway 自体は OAuth 認証のプロセスをカバーしていないため get_oauth_token で token を取得してからアクセスしていました。AgentCore Identity で認証認可プロセスを実施し Gateway に必要な token を取得できます。具体的には次のようなフローになります。
体験手順
AgentCore Identiy を使用し、先ほど構築した Gateway を通過してみます。ディレクトリの構成は以下のようになります。
04_identity/
├── README.md # This documentation
├── setup_credential_provider.py # OAuth2 provider setup
├── agent_with_identity.py # Main agent implementation
└── test_identity_agent.py # Test suite
Step 1: OAuth2 の Credential Provider の作成
Gateway 作成時に Authorizer として作成した Cognito は OAuth をサポートしているため、こちらを token を得るための "Credential Provider" として AgentCore Identity を作成します。先ほど作成した Gateway と Cognito の情報は 03_gateway/gateway_config.json から読み取りますので、事前に 03_gateway の実行が必要です。
cd 04_identity
uv run python setup_credential_provider.py
Step 2: Identity のテスト
レスポンスが正常に帰ってくれば、AgentCore Identity で取得した token で Gateway を通過しています！
cd 04_identity
uv run python test_identity_agent.py
実装の解説
AgentCore Identity に必要な Credential Provider は
create_oauth2_credential_provider で作成します。Google や Slack、GitHub といった 3rd Party の認証認可先を登録したり、AWS では Cognito も設定できます。Cognito を登録する際は、client / secret 、また認証先の情報を得るための discovery_url をまとめた oauth2_config を渡して作成します。
04_identity/setup_credential_provider.py
ROVIDER_NAME = "agentcore-identity-for-gateway"
def setup_oauth2_credential_provider(provider_name: str = PROVIDER_NAME):
# Load gateway configuration from existing setup
config_path = Path("../03_gateway/gateway_config.json")
if not config_path.exists():
logger.error(f"Gateway configuration not found at {config_path}")
logger.error("Please deploy the gateway first using 03_gateway setup")
return False
try:
with open(config_path, 'r') as f:
gateway_config = json.load(f)
logger.info("✅ Loaded existing gateway configuration")
except Exception as e:
logger.error(f"Failed to load gateway configuration: {e}")
return False
# Extract Cognito configuration
cognito_config = gateway_config['cognito']
region = gateway_config['region']
user_pool_id = cognito_config['user_pool_id']
# Construct OpenID Connect discovery URL for Cognito
# This URL provides OAuth2 endpoints and configuration
discovery_url = f"https://cognito-idp.{region}.amazonaws.com/{user_pool_id}/.well-known/openid-configuration"
logger.info(f"Using Cognito discovery URL: {discovery_url}")
logger.info(f"Client ID: {cognito_config['client_id']}")
logger.info(f"Scope: {cognito_config['scope']}")
# Create bedrock-agentcore-control client for managing credential providers
try:
client = boto3.client('bedrock-agentcore-control', region_name=region)
logger.info(f"✅ Created AgentCore control client for region {region}")
except Exception as e:
logger.error(f"Failed to create AgentCore control client: {e}")
return False
# Create new credential provider configuration
# https://docs.aws.amazon.com/bedrock-agentcore-control/latest/APIReference/API_CustomOauth2ProviderConfigInput.html
# https://docs.aws.amazon.com/bedrock-agentcore-control/latest/APIReference/API_Oauth2Discovery.html
oauth2_config = {
'customOauth2ProviderConfig': {
'clientId': cognito_config['client_id'],
'clientSecret': cognito_config['client_secret'],
'oauthDiscovery': {
'discoveryUrl': discovery_url
}
}
}
try:
logger.info(f"Creating OAuth2 credential provider '{provider_name}'...")
# API Reference: https://docs.aws.amazon.com/bedrock-agentcore-control/latest/APIReference/API_CreateOauth2CredentialProvider.html
response = client.create_oauth2_credential_provider(
name=provider_name,
credentialProviderVendor='CustomOauth2',
oauth2ProviderConfigInput=oauth2_config
)
logger.info("✅ Successfully created OAuth2 credential provider!")
logger.info(f" ARN: {response['credentialProviderArn']}")
logger.info(f" Name: {provider_name}")
logger.info(f" Scope to use: ['{cognito_config['scope']}']")
logger.info(" Auth Flow: M2M (Machine-to-Machine)")
return True
except ClientError as e:
logger.error(f"Failed to create credential provider: {e}")
logger.error("Common issues:")
logger.error("- Check AWS permissions for bedrock-agentcore-control")
logger.error("- Verify Cognito client credentials are valid")
logger.error("- Ensure discovery URL is accessible")
return False
作成した credential provider (PROVIDER_NAME) は次のように @requires_access_token で指定し、メソッド実行時に認証プロセスをキックすることが出来ます。
async def get_access_token(self) -> str:
@requires_access_token(
provider_name=PROVIDER_NAME,
scopes=[self.cognito_config['scope']],
auth_flow="M2M", # Machine-to-Machine authentication
force_authentication=False # Use cached tokens when available
)
async def _get_token(*, access_token: str) -> str:
return access_token
return await _get_token()
これにより、AI エージェントが Identity を得るプロセスを簡単に実装でき、暗号化されたトークンの保管・リフレッシュといった運用を Managed でお任せすることが出来ます。
本セクションのまとめ
-
AgentCore Identity により認証認可プロセスと機密トークンの保管が安全・簡単に実装できる : AI エージェントがユーザーの代理として動く場合、ユーザーによる認証のプロセスとその結果の保管は厳密かつ安全に行われる必要があります。AgentCore Identity は、Identity を得るタイミングの制御 (
requires_access_tokenの指定)、取得プロセス、取得後のセキュアな管理を容易に実装することが出来ます
📊 : AI エージェントの動作をモニタリングする : AgentCore Observability
AgentCore Observabilityは、AI エージェントが処理を受け付けてから完了するまでの一連の過程を追跡可能にします。Observability への注目が高まる背景には、LLM 単体から回答を得る "AI アシスタント" から自律的にタスク完了まで行う "AI エージェント" へと用途が進化するにつれて End to End の処理の追跡が困難になってきていることがあります。このような問題を解決するため、Langfuse など生成 AI アプリケーションに特化した Observability ツールも登場してきています。AgentCore Observability は CloudWatch の GenAI Observability と統合されており、生成 AI に留まらないアプリケーションワークロード全体の Trace を指向しています。
体験手順
Step 1: CloudWatch の Transaction Search を有効化
AgentCore Observability を体験するには、CloudWatch の Transaction Search を ON にします (参考 : Add observability to your Amazon Bedrock AgentCore resources)。なお、この設定は Amazon Bedrock のログを記録する Model Invocation Logging と同様にリージョンごとに必要です。
Step 2: Invocation を実行
次のスクリプトでモニタリングするためのログを記録します。
cd 05_observability
uv run python test_observability.py
session id としてユーザー名 + 日時を使用し、AWS の見積りについて 3 回会話するシナリオを実行します。
Step 3: Trace を確認する
実行した結果を確認してみましょう。AgentCore の画面から遷移できます。
遷移後の画面では、直近のタイムスパン (1 時間など) でどの程度のセッション / Trace が記録されたのか、またエラーやスロットリングの全体状況が確認できます。
Session のタブで、セッションごとのエラーやスロットリングが確認できます。このようにユーザー識別子があると、どのユーザーがスロットリングやエラーに直面しているかがすぐに特定できます。
エラーを引き起こした入出力は何か ? はもちろん Session の中の Trace を追跡することで探ることが出来ます。
Trace の中で時間やエラーが発生している Span があれば、その中のリクエストを詳しく知ることが出来ます。
Span は複数のイベントのまとまりなので、その中を見ることで入出力されているプロンプトなどを確認できます。システム、ユーザー、アシスタント、またツールについて細かくイベントの内容を確認できます。
ダイアグラム形式で確認することもでき、この場合 tool でサービスコードを取得して get pricing して計算しているななど確認が出来ます。
注目すべきなのは、ボタンを ON にするだけでここまで解像度高く分析ができることです。私が一生懸命ログ出力を実装したからここまで細かいわけではなく、すべて自動計装の恩恵です。裏側は OpenTelemetry なので、AWS 以外の監視ツールに送信もできますが標準で十二分な探索を行うことが出来ます。
本セクションのまとめ
- AgentCore Observability によりボタン一つで AI エージェントが応対している Session 、その中のインタラクションを示す Trace 、Trace 内の各処理フローである Span までたどることが出来、Span 内の Event から詳細な入出力やツールの呼び出しまでトラッキングが出来ます。特に session id に意味ある id を振ることで、問合せからの探索を高速に行うことが出来ます。
🧠 : 見積の内容を「記憶」する : AgentCore Memory
AgentCore Memoryは、過去のインタラクションの内容を記憶します。これまでは外付けのデータベース (Amazon DynamoDB など) が不可欠でしたが、同一セッション内の会話やセッションをまたぐユーザーの嗜好や傾向といった長期記憶も AgentCore の中で保持できるようになりました。
短期記憶、長期記憶の具体例として以下 2 つの実装を紹介します。
- 見積比較 : 過去のインタラクションの見積りを記憶しておき、見積り結果の比較を行う
- アーキテクチャ提案 : Web サイトや Web アプリケーションなど、ユーザーの関心あるシステムを覚えておいて提案する
体験手順
全体のフローは次の通りになります。まず、ユーザーは AgentWithMemory にいくつか見積り (estimate tool を使用) を依頼します。その後、それを比較してくれるように頼みます (compare tool を使用)。この時、蓄積しておいた event の情報 (= 短期記憶) を使用します。短期記憶は所定のプロンプトで集約され長期記憶 (preference) に保存されるので、最後に最適なアーキテクチャの提案 (propose tool を使用) を依頼します。
ディレクトリの構成は次の通りです。
06_memory/
├── README.md # This documentation
├── test_memory.py # Main implementation and test suite
└── .gitignore # Git ignore patterns
Step 1: AgentCore Memory を作成し実行
cd 06_memory
uv run python test_memory.py
Memory を作り直したい場合は --force オプションを使用します。ただ、Memory の作成なかなか時間がかかるのでなるべく使いまわすことを推奨です。正常に動作すれば、compare を行っている際は記憶している見積りがきちんと比較されていること、
2025-07-29 05:25:52,712 - __main__ - INFO - ✅ Comparison completed for 2 estimates
Here's a comprehensive comparison of your two AWS architecture estimates:
## Cost Comparison Overview
| Architecture | Monthly Cost | Cost Ratio |
|-------------|-------------|------------|
| **Small Blog** (EC2 t3.micro + RDS) | **$26.71** | Baseline |
| **Medium Web App** (Load balanced t3.small + RDS) | **$78.98** | **2.96× more expensive** |
propose の際はメモリをロードして嗜好を理解したうえで提案していることがわかります。
💡 Getting personalized recommendation...
I'd be happy to propose the best architecture for your specific needs! However, to provide you with the most suitable and cost-effective recommendation, I need to understand your requirements better.
Tool #4: propose
2025-07-29 05:29:25,924 - __main__ - INFO - 💡 Generating architecture proposal based on user history...
2025-07-29 05:29:26,272 - bedrock_agentcore.memory.client - INFO - Retrieved 3 memories from namespace: /preferences/user123
2025-07-29 05:29:26,272 - __main__ - INFO - 🔍 Generating proposal with requirements: Based on user's previous estimates for small blog and medium web app architectures, need to understand specific preferences for optimal architecture recommendation
実装の解説
AgentCore の Memory は strategies を指定しなければ短期記憶専用、指定する場合どんな戦略で保管するのか (userPreferenceMemoryStrategy)、どのような単位で分けるのか (namespaces) を指定できます。Strategy はデフォルトでいくつか用意されていますが、自分で作成することも可能です (Getting started with AgentCore Memory)。
class AgentWithMemory:
def __init__(self, actor_id: str, region: str = "us-west-2", force_recreate: bool = False):
# Initialize AgentCore Memory with user preference strategy
self.memory = self.memory_client.create_memory_and_wait(
name="cost_estimator_memory",
strategies=[{
"userPreferenceMemoryStrategy": {
"name": "UserPreferenceExtractor",
"description": "Extracts user preferences for AWS architecture decisions",
"namespaces": [f"/preferences/{self.actor_id}"]
}
}],
event_expiry_days=7,
)
短期記憶は次のように create_event を使い行います。messages 見て頂いてわかる通り、通常の会話をロールをつけて記録するイメージになります。
@tool
def estimate(self, architecture_description: str) -> str:
# Generate cost estimate
result = cost_estimator.estimate_costs(architecture_description)
# Store interaction in memory for future comparison
self.memory_client.create_event(
memory_id=self.memory_id,
actor_id=self.actor_id,
session_id=self.session_id,
messages=[
(architecture_description, "USER"),
(result, "ASSISTANT")
]
)
return result
引き出す時は list_events を使います。
events = self.memory_client.list_events(
memory_id=self.memory_id,
actor_id=self.actor_id,
session_id=self.session_id,
max_results=max_results
)
短期から長期にどのタイミングで移行するかは明確な記述を見つけられなかったですが、we wait for 1 minute to allow the long-term memory strategies to process and extract meaningful information before retrieving it. との記述から一定時間経過後にイベントが増えていれば要約が行われと推定されます (Getting started with AgentCore Memory)。
長期記憶を引き出すには retrieve_memories を使用します。
memories = self.memory_client.retrieve_memories(
memory_id=self.memory_id,
namespace=f"/preferences/{self.actor_id}",
query=f"User preferences and decision patterns for: {requirements}",
top_k=3
)
contents = [memory.get('content', {}).get('text', '') for memory in memories]
引き出した記憶をプロンプトに埋め込むことで、応答をカスタマイズできます。
本セクションのまとめ
- AgentCore Memory を使用することで、外部データベースに依存せずユーザーとのインタラクションを記録できます。短期記憶はメッセージの配列をイベントとして登録する形で、長期記憶は新規イベント発生時に指定もしくはカスタムのプロンプトで自動登録されます。これらを活用することで、よりパーソナライズした体験やより長い処理を任せることが出来るようになります。
結論と次のステップ
本記事は、AI エージェントの実装では技術的負債の蓄積が避けがたい課題を挙げ、その「予防」策として Amazon Bedrock AgentCore の一連の機能をまるっと紹介しました。 AWS ユーザーの方もそうでない方も、"AI エージェントの実装に集中する" ために試して頂ければ幸いです！