AI エージェント開発の技術的負債を予防する : Amazon Bedrock AgentCore をゼロからまるっと体験

Last updated at 2025-08-16Posted at 2025-07-29

技術的負債の蓄積しやすい 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 Identity
🔌 : 外部サービスを MCP に変換し接続する : AgentCore Gateway
📊 : 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/aws-samples/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/
│   ├── 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 エージェントのための認証認可 : Amazon Bedrock AgentCore Identity

AgentCore Runtime にデプロイしたエージェントへアクセスするにはデフォルトで IAM SigV4 による認証が必要です。より Web サービス利用時の認可方式として一般的な OAuth 認可を実装するには、AgentCore Identity が役立ちます。これは、MCP や AI エージェントの実行環境をよりガバナンスやコストをコントロールするために認証付きクラウド環境へ移行したい場合に有用です。

AgentCore Identity には次のような特徴があります。

AI エージェントごとの認証を workload identity で管理し、IAM Role や OAuth による認可、API Key によるアクセスが「どの Agent に」付与されているかを一貫して管理
@requires_access_token のデコレーターを付与するだけで OAuth による認可プロセスを実行し access token を取得できる
GitHub や Google、Microsoft、Slack といった多様な Credential Provider と容易に連携
AgentCore Identity のデータは Token Vaults に暗号化され格納。暗号化キーは AWS Managed、Customer managed key 双方を利用可能

体験手順

サンプルのディレクトリの構成は次のようになっています。

03_identity/
├── README.md                      # This documentation
├── setup_inbound_authorizer.py    # OAuth2 provider and secure runtime setup
└── test_identity_agent.py         # Test agent with identity authentication

今回は、認可を付与してコスト見積もりエージェントをデプロイします (Secure AgentCore Runtime)。デプロイしたエージェントにアクセスする際の認可のフローを示します。

Step 1: 作業ディレクトリへの移動

体験するための作業ディレクトリへ移動します。

cd sample-amazon-bedrock-agentcore-onboarding/03_identity

Step 2: AgentCore Identity と認可サーバー (Cognito) の作成

AgentCore Identity は認可サーバーと連携するため、Amazon Cognito を作成してから AgentCore Identity を構築します。これを実行しているのが setup_inbound_authorizer.py です。

uv run python setup_inbound_authorizer.py

正常に終了すれば、次の 3 つが構築されます。

認可を行う Cognito
認可プロセスを行う AgentCore Identity
認可プロセスを通じて得られるアクセストークンを要求する AgentCore Runtime

Step 3: 認可付き AgentCore Runtime を起動する

test_identity_agent.py を実行することで認可付き AgentCore Runtime にアクセスできます。アクセスには、設定したように Cognito を通じた認可が必要ですが、AgentCore Identity がこの面倒なプロセスを担い得られたアクセストークンをセキュアに保管してくれます。

uv run python test_identity_agent.py

認可の設定されたエンドポイントが正常に呼び出せることを確認できます。

実装の解説

Inbound の認可サーバーの構築

AgentCore Identity 自体は認証認可の機能を持っていないため、別途作成する必要があります。今回は、GatewayClient に簡単に M2M で認可を行う Amazon Cognito を作れる機能があるため (create_oauth_authorizer_with_cognito)、そちらを利用して Cognito を準備します。

from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient
・・・
gateway_client = GatewayClient(region_name=region)
cognito_result = gateway_client.create_oauth_authorizer_with_cognito("InboundAuthorizerForCostEstimatorAgent")
user_pool_id = cognito_result['client_info']['user_pool_id']
discovery_url = f"https://cognito-idp.{region}.amazonaws.com/{user_pool_id}/.well-known/openid-configuration"
cognito_config = {
    "client_id": cognito_result['client_info']['client_id'],
    "client_secret": cognito_result['client_info']['client_secret'],
    "token_endpoint": cognito_result['client_info']['token_endpoint'],
    "discovery_url": discovery_url,
    "scope": cognito_result['client_info']['scope'],
    "user_pool_id": user_pool_id,
    "region": region
}

認可サーバーを Runtime に設定する

Cognito を構築したら、AgentCore Identity に設定します。Identity は認可のプロセスと結果得られたアクセストークンを保管する機能を持っており、認証認可の処理自体は Cognito を含めた Credential Provider に委譲します。client_id、client_secret は認可プロセスを開始するために、discovery_url は認可サーバーの宛先を知るのに必要です。

oauth2_config = {
    'customOauth2ProviderConfig': {
        'clientId': cognito_config['client_id'],
        'clientSecret': cognito_config['client_secret'],
        'oauthDiscovery': {
            'discoveryUrl': cognito_config['discovery_url']
        }
    }
}

# API Reference: https://docs.aws.amazon.com/bedrock-agentcore-control/latest/APIReference/API_CreateOauth2CredentialProvider.html
response = identity_client.create_oauth2_credential_provider(
    name=provider_name,
    credentialProviderVendor='CustomOauth2',
    oauth2ProviderConfigInput=oauth2_config
)

構築した Cognito での認可なしには AgentCore Runtime を起動できないよう、authorizer_config を設定してデプロイを行います。

authorizer_config = {
    "customJWTAuthorizer": {
        "discoveryUrl" : config["cognito"]["discovery_url"],
        "allowedClients" : [config["cognito"]["client_id"]]
    }
}

deploy_client = boto3.client('bedrock-agentcore-control', region_name=region)
response = deploy_client.create_agent_runtime(
    agentRuntimeName=secure_agent_name,
    agentRuntimeArtifact={
        "containerConfiguration" : {
            "containerUri": f"{ecr_repository}:latest"
        }
    },
    networkConfiguration={"networkMode": network_mode},
    roleArn=execution_role,
    authorizerConfiguration=authorizer_config

)

AgentCore Identity で認可プロセスを行う

@requires_access_token で AgentCore Identity (provider_name) を指定することで cost_estimator_tool の引数 access_token に実際に取得したアクセストークンが引き渡されます。このトークンをヘッダーに追加した POST リクエストにより AgentCore Runtime が起動できることがわかります。

@tool(name="cost_estimator_tool", description="Estimate cost of AWS from architecture description")
@requires_access_token(
    provider_name= OAUTH_PROVIDER,
    scopes= [OAUTH_SCOPE],
    auth_flow= "M2M",
    force_authentication= False)
async def cost_estimator_tool(architecture_description, access_token: str) -> str:
    session_id = f"runtime-with-identity-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ')}"
    if access_token:
        logger.info("✅ Successfully load the access token from AgentCore Identity!")
        for element in access_token.split("."):
            logger.info(f"\t{json.loads(base64.b64decode(element).decode())}")

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": session_id,
        "X-Amzn-Trace-Id": session_id,
    }

    response = requests.post(
        RUNTIME_URL,
        headers=headers,
        data=json.dumps({"prompt": architecture_description})
    )

    response.raise_for_status()
    return response.text

本セクションのまとめ

AgentCore Identity により安全かつ簡易に認証認可処理をエンドポイントに実装できる : 認証認可を行うプロバイダと AgentCore Identity を一度連携させてしまえば、Runtime はもちろん Gateway などを含め認証認可を要求する Agent や tool に対し @requires_access_token をセットし簡易にアクセストークンの要求・取得処理を実装できます

🔌 : 外部サービスを MCP に変換し接続する : AgentCore Gateway

外部サービスが提供している API や、企業内ですでに実装済みの API がある時、これらを再利用したいのは自然な動機です。AgentCore Gateway は、OpenAPI や Smithy 、また AWS Lambda といった API の各実装様式を AI エージェントが扱える MCP の形式に変換するコネクタの役割を果たします。これにより、AI エージェントが実装済みの様々な API を簡単に利用できるようにします。

AgentCore Gateway には次の特徴があります。

企業内外で実装された Web API、Lambda 関数などを MCP 対応ツールへ変換する。特に、Salesforce、Slack、Jira、Asana、Zendesk などの代表的なサービスについては 1 click での統合を提供
AgentCore Identity と連携した Inbound 、外部ツールの Outbound 接続に対する OAuth をはじめとした認可双方をカバー。AI エージェントに対し、外部連携するための統一された玄関口として機能
セマンティック検索により、数千のツールが登録されていても少ないプロンプトでツールの呼び出しが可能

体験手順

AI エージェントから、AgentCore Identity でデプロイした認証付きのコスト見積もりエージェントと、Gateway を通じ提供されるメールを送るツールの 2 つを呼び出すエージェントを実装します。なお、エージェントの認証と Gateway の Inbound 認証とでは同じ Cognito を使用しています。

サンプルのディレクトリの構成は次のようになっています。

04_gateway/
├── README.md                      # This documentation
├── src/app.py                     # Lambda function implementation
├── template.yaml                  # SAM template for Lambda
├── deploy.sh                      # Lambda deployment script
├── setup_outbound_gateway.py      # Gateway setup with Cognito
└── test_gateway.py                # Gateway testing with Strands Agent

実行する処理の全体の流れは次の通りです。

Step 1: ディレクトリの移動

AgentCore Gateway 用のディレクトリに移動します：

cd sample-amazon-bedrock-agentcore-onboarding/04_gateway

Step 2: AWS Lambda のデプロイ

AgentCore Gateway に接続する外部サービスとして、AWS Lambda をデプロイします。この AWS Lambda (src/app.py) は、エージェントが作成した Markdown のコスト見積もりレポートを Amazon SES を使い HTML 形式のメール送信します。Markdown から HTML への変換には markdown を使用しています。

 html_content = markdown.markdown(
    markdown_text,
    extensions=['tables', 'nl2br']
)

では、デプロイを行いましょう。AWS SAM がインストールされていれば、次のコマンドを実行することでデプロイできます。引数として有効なメールアドレスを設定します。コスト見積もりのレポートを送る送信元のメールアドレスになります。実在のアドレスを使用しないと認証に時間がかかるため、ご自身のアドレスを設定することを推奨します。

./deploy your@email.address

デプロイすると、メールアドレスが有効か確認するメールが届きます。メール内のリンクをクリックし、認証を完了してください。

認証が完了すると次のようなページが遷移先で表示されます。

Step 3: AgentCore Gateway の作成

接続するサービスをデプロイできたので、setup_outbound_gateway.py を使用し AgentCore Gateway を作成します。AgentCore Gateway には 2 種類の認証認可があります。

Inbound : AgentCore Gateway にアクセスするための認証認可
Outbound : AgentCore Gateway が外部サービスを呼び出す時の認証認可

今回、Inbound には AgentCore Identity のセクションで作成した Cognito を使用します。Outbound は、AWS Lambda の場合は AgentCore Gateway の IAM Role (GATEWAY_IAM_ROLE) で認証します。外部の、例えば GitHub などのサービスであれば OAuth などを設定することになります。

では、実際 AgentCore Gateway を作ってみましょう。次のコマンドを実行します。

uv run setup_outbound_gateway.py

Step 4: AgentCore Gateway をテストする

test_gateway.py により AgentCore Gateway をテストできます。AgentCore Gateway の Inbound を通過するための認可は AgentCore Identity を通じ簡単にトークンを取得できます。取得したトークンを使い AgentCore Gateway にアクセスし、見積もり結果をメールで送ります。

uv run python test_gateway.py --architecture 会員1000人への推薦メール配信 --address your-mail-address@yourcompany.co.jp

上手く実行できると次のようなメールが届きます。

実装の解説

AgentCore Gateway の Inbound の実装

AgentCore Gateway の Inbound の認可は create_mcp_gateway に authorizer_config を渡して設定しています。これは、AgentCore Identity の時に認可を要求する AgentCore Runtime をデプロイした時と同じです。

gateway_name = "AWSCostEstimatorGateway"
authorizer_config = {
    "customJWTAuthorizer": {
        "discoveryUrl": identity_config["cognito"]["discovery_url"],
        "allowedClients": [identity_config["cognito"]["client_id"]]
    }
}
gateway = gateway_client.create_mcp_gateway(
    name=gateway_name,
    role_arn=None,
    authorizer_config=authorizer_config,
    enable_semantic_search=False
)

AgentCore Gateway の Outbound の実装

Outbound は次のように設定しています。AgentCore Gateway からアクセスする AWS Lambda の lambdaArn を設定し、credentialProviderType には IAM Role を設定しています。

target_name = gateway_name + "Target"
    
create_request = {
    "gatewayIdentifier": gateway_id,
    "name": target_name,
    "targetConfiguration": {
        "mcp": {
            "lambda": {
                "lambdaArn": config["lambda_arn"],
                "toolSchema": {
                    "inlinePayload": tool_schema
                }
            }
        }
    },
    "credentialProviderConfigurations": [{"credentialProviderType": "GATEWAY_IAM_ROLE"}]
}

target_response = control_client.create_gateway_target(**create_request)

AgentCore Gateway へのアクセス

AgentCore Gateway にアクセスするため、AgentCore Identity から Inbound の通過に必要なアクセストークンを取得しています。AgentCore Gateway へは、Streamable HTTP でアクセスします。

@requires_access_token(
    provider_name= OAUTH_PROVIDER,
    scopes= [OAUTH_SCOPE],
    auth_flow= "M2M",
    force_authentication= False)
async def get_access_token(access_token):
    """Helper function to get access token"""
    if access_token:
        logger.info("✅ Successfully loaded the access token!")
    return access_token

def estimate_and_send(architecture_description, address):
    logger.info("Testing Gateway with MCP client (Strands Agents)...")

    # Get the access token first
    access_token = asyncio.run(get_access_token())
    # Create the transport callable that returns the HTTP client directly
    def create_transport():
        return streamablehttp_client(
            GATEWAY_URL,
            headers={"Authorization": f"Bearer {access_token}"}
        )

    mcp_client = MCPClient(create_transport)

AgentCore Gateway から list_tools_sync でツールの一覧を取得しています。これにより、コスト見積もりと AgentCore Gateway に Target として登録されているメール送信 2 つのツールがエージェントに与えられます。

tools = [cost_estimator_tool]
    with mcp_client:
        more_tools = True
        pagination_token = None
        while more_tools:
            tmp_tools = mcp_client.list_tools_sync(pagination_token=pagination_token)
            tools.extend(tmp_tools)
            if tmp_tools.pagination_token is None:
                more_tools = False
            else:
                more_tools = True 
                pagination_token = tmp_tools.pagination_token

        _names = [tool.tool_name for tool in tools]

        agent = Agent(
            system_prompt=(
                "Your are a professional solution architect. Please estimate cost of AWS platform."
                "1. Please summarize customer's requirement to `architecture_description` in 10~50 words."
                "2. Pass `architecture_description` to 'cost_estimator_tool'."
                "3. Send estimation by `markdown_to_email`."
            ),
            tools=tools
        )

本セクションのまとめ

AgentCore Gateway により既存の API の MCP 化が簡単になる : AgentCore Gateway により、これまで使用、実装してきた様々な API を簡単かつサーバーレスで MCP 化することが出来ます。これにより、Agent の活動範囲をぐっと広げることができます。

📊 : 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 の中で保持できるようになりました。

AgentCore Memoryの特徴として、短期記憶と長期記憶、2 つの Memory 機能を提供します。

短期記憶は、エージェントとユーザーの対話をイベントとして保存し、セッション (対話) 内でのコンテキストを維持します。
長期記憶は、会話の要約、確認された事実や知識、ユーザーの好み、といった会話から推定される (ユーザーとの) 次回のセッションに引き継ぐべき重要な洞察を保存します。

Memory Strategy は、AI エージェントが短期記憶を長期記憶へ移すための戦略を定義します。Memory Strategy により短期記憶の内容が長期記憶に保存されます。

Memory Strategy は、4 つの種類があります。

SemanticMemoryStrategy: 会話から得られた事実や知識に注目
SummaryMemoryStrategy: 要点や決定事項に注目
UserPreferenceMemoryStrategy: ユーザーの好みや選択パターンに注目
CustomMemoryStrategy: カスタムプロンプトで特定の情報に注目

より具体的なイメージを得るために実装してみましょう。短期記憶、長期記憶の具体例として以下 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

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 を使用することで、外部データベースに依存せずユーザーとのインタラクションを記録できます。短期記憶はメッセージの配列をイベントとして登録する形で、長期記憶は新規イベント発生時に指定もしくはカスタムのプロンプトで自動登録されます。これらを活用することで、よりパーソナライズした体験やより長い処理を任せることが出来るようになります

構築したリソースのクリーンアップ

本ハンズオンで構築したリソースは、下記手順でクリーンアップできます。

# 1. Clean up Memory resources first
cd 06_memory
uv run python clean_resources.py

# 2. Clean up Gateway resources (uses SAM CLI)
cd 04_gateway
sam delete  # Deletes Lambda function and associated resources
uv run python clean_resources.py  # Additional cleanup if needed

# 3. Clean up Identity resources
cd 03_identity
uv run python clean_resources.py

# 4. Clean up Runtime resources
cd 02_runtime
uv run python clean_resources.py

結論と次のステップ

本記事は、AI エージェントの実装では技術的負債の蓄積が避けがたい課題を挙げ、その「予防」策として Amazon Bedrock AgentCore の一連の機能をまるっと紹介しました。 AWS ユーザーの方もそうでない方も、"AI エージェントの実装に集中する" ために試して頂ければ幸いです！

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

AI エージェント開発の技術的負債を予防する : Amazon Bedrock AgentCore をゼロからまるっと体験

技術的負債の蓄積しやすい AI エージェントの開発

実装に依存しない「技術的負債を予防する」ランタイム環境の重要性

Amazon Bedrock AgentCore の全機能を体験する

🧮 AWS の見積りを「計算」するエージェントを作成する : AgentCore Code Interpreter

体験手順

Step 1: AWS コスト見積りエージェントの実行

Step 2: Streaming で試す

実装の解説

🚀 : クラウド上に AI エージェントをデプロイする : AgentCore Runtime

体験手順

Step 1: AI エージェントの準備を行う

Step 2: 生成された agentcore configure のコマンドを実行

実装の解説

🛡️ : AI エージェントのための認証認可 : Amazon Bedrock AgentCore Identity

体験手順

Step 1: 作業ディレクトリへの移動

Step 2: AgentCore Identity と認可サーバー (Cognito) の作成

Step 3: 認可付き AgentCore Runtime を起動する

実装の解説

Inbound の認可サーバーの構築

認可サーバーを Runtime に設定する

AgentCore Identity で認可プロセスを行う

🔌 : 外部サービスを MCP に変換し接続する : AgentCore Gateway

体験手順

Step 1: ディレクトリの移動

Step 2: AWS Lambda のデプロイ

Step 3: AgentCore Gateway の作成

Step 4: AgentCore Gateway をテストする

実装の解説

AgentCore Gateway の Inbound の実装

AgentCore Gateway の Outbound の実装

AgentCore Gateway へのアクセス

📊 : AI エージェントの動作をモニタリングする : AgentCore Observability

体験手順

Step 1: CloudWatch の Transaction Search を有効化

Step 2: Invocation を実行

Step 3: Trace を確認する

🧠 : 見積の内容を「記憶」する : AgentCore Memory

体験手順

Step 1: AgentCore Memory を作成し実行

実装の解説

構築したリソースのクリーンアップ

結論と次のステップ

Step 2: 生成された `agentcore configure` のコマンドを実行