はじめに

昨今、AIエージェントの開発が流行っており、私の所属する会社の取引先でも、AIエージェントの開発や相談が多く寄せられているらしい👀
そのため、私も本格的にAIエージェントの開発方法を学ばなければならないと思い、AIエージェントを安心・安全に構築できるAmazon Bedrock AgentCoreについて学んだので、記録として残しておこうと思う💪

というのは実は建前で、楽しみにしていたAmazon Bedrock AgentCoreが、やっと東京リージョンに対応したので、良い機会なので学習したいと思ったというのがホントのところ🫣

学習素材は『Amazon Bedrock AgentCoreのワークショップ』を使って学習を進めていく。

サンプルコード

とりあえず、サンプルコードをクローンするところから始める。

git clone https://github.com/aws-samples/sample-amazon-bedrock-agentcore-onboarding.git

Amazon Bedrock AgentCoreとは？

AIエージェントを安心・安全に、かつ大規模にデプロイおよび運用するためのサービス。

CrewAI、LangGraph、LlamaIndex、Strands Agents などのあらゆるフレームワークに対応しており、Amazon Bedrock 内外のあらゆる基盤モデルと連携して動作でき、柔軟性のあるインフラを作ることができる。

AgentCoreの主な機能は以下の通り。

🧮 Code Interpreter : AIエージェントのための安全なサンドボックス環境を提供
🚀 Runtime : スケーラブルなデプロイと管理
🛡️ Identity: AI エージェントのための認証認可
🔌 Gateway: 外部サービスを MCP に変換し接続する
📊 Observability: End to End での動作監視
🧠 Memory: 短期・長期記憶機能による文脈認識

🧮 Code Interpreter

AIエージェントのための安全なサンドボックス環境を提供してくれる機能

AIの行動は完全に予測することできないので、突然よからぬファイルを削除してしまって「あー、プログラムが動かなくなったー」みたいなことが起きる可能性があります。
そのため、AIを動かす時は、他の環境とは隔離された環境下で動かせるように環境を用意しましょうという機能です。

いわゆる「ゼロトラスト（「何も信用しない」ことを前提としたセキュリティの考え方）」で、制限された環境（インターネットへの接続やOS領域への書き込みを禁止）でAIを動かせる機能です。

使い方としては、AIによる実行前と後に start() と stop() するだけのようです。

# start()
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.error(f"❌ Failed to setup Code Interpreter: {e}")
        return  # Handle the error instead of re-raising

# stop()
def cleanup(self) -> None:
    """Clean up resources"""
    logger.info("🧹 Cleaning up resources...")
    
    if self.code_interpreter:
        try:
            self.code_interpreter.stop() # サンドボックス環境を使用停止
            logger.info("✅ Code Interpreter session stopped")
        except Exception as e:
            logger.warning(f"⚠️ Error stopping Code Interpreter: {e}")
        finally:
            self.code_interpreter = None

AIを動かすコードは以下の通り。
@contextmanagerを書くことで、処理が成功しても失敗しても、Code Interpreterを確実に停止されるようになる様です。

なお、このコードをそのまま東京リージョンで動かそうとすると、モデルによっては東京リージョンに対応していないモデルもあるので、クロスリージョンに対応させて動かすようにしました。
（Bedrock呼び出し時に us-east-1 で実行できるように REGION を追記）

@contextmanager
def _estimation_agent(self) -> Generator[Agent, None, None]:
    """
    Context manager for cost estimation components
    
    Yields:
        Agent with all tools configured and resources properly managed
        
    Ensures:
        Proper cleanup of Code Interpreter and MCP client resources
    """        
    try:
        logger.info("🚀 Initializing AWS Cost Estimation Agent...")
        
        # Setup components in order
        self._setup_code_interpreter()
        aws_pricing_client = self._setup_aws_pricing_client()
        
        # Create agent with persistent MCP context
        with aws_pricing_client:
            pricing_tools = aws_pricing_client.list_tools_sync()
            logger.info(f"Found {len(pricing_tools)} AWS pricing tools")
            
            # Create agent with both execute_cost_calculation and MCP pricing tools
            all_tools = [self.execute_cost_calculation] + pricing_tools
            agent = Agent(
                BedrockModel(
                    boto_client_config=Config(
                        read_timeout=900,
                        connect_timeout=900,
                        retries=dict(max_attempts=3, mode="adaptive"),
                    ),
                    model_id=DEFAULT_MODEL,
                    region_name=AWS_REGION # 追記:クロスリージョンに対応
                ),
                tools=all_tools,
                system_prompt=SYSTEM_PROMPT
            )
            
            yield agent
            
    except Exception as e:
        logger.exception(f"❌ Component setup failed: {e}")
        raise
    finally:
        # Ensure cleanup happens regardless of success/failure
        self.cleanup()

驚くべきことに、このサンドボックス機能は、AWSへデプロイしていなくても使用できます。

サンプルのように、AI側でコードを書いてもらって計算・実行させるような場合には、サンドボックス環境があると安心・安全ですね。なんて素晴らしい✨

🚀 AgentCore Runtime

スケーラブルなデプロイと管理を提供してくれる機能

いよいよデプロイです。

以下コマンドで、Code Interpreterで使用したコードのコピーとIAMロールを作成します。

cd 02_runtime
uv run prepare_agent.py --source-dir ../01_code_interpreter/cost_estimator_agent

AgentCore Runtimeに必要なIAM 権限はPermissions for AgentCore Runtimeで確認できます。

bedrock-agentcore:GetWorkloadAccessToken - AgentCore で稼働する Agent に割り当てられる認証トークンの取得
ecr:BatchGetImage - Amazon Elastic Container Registry (ECR) からのイメージ取得
bedrock:InvokeModel - Bedrockモデルの実行
bedrock-agentcore:CreateCodeInterpreter - Code Interpreter の作成
logs:* - CloudWatch Logsへの書き込み

次に、エントリーポイント（AgentCore Runtime が呼び出すところ）を用意します。
@app.entrypointで指定する様です。

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()

    # Batch
    return agent.estimate_costs(user_input)


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

あとは、 agentcore コマンドを使って、デプロイとエージェントを起動させて、テストを実施します。

# エージェントランタイムを設定 (アカウント id は実行環境に依存します。 `prepare_agent.py` の実行結果を確認してください)
uv run agentcore configure --entrypoint ./deployment/invoke.py --name cost_estimator_agent --execution-role arn:aws:iam::123456789012:role/AgentCoreRole-cost_estimator_agent --requirements-file ./deployment/requirements.txt --disable-otel --region us-east-1

# エージェントを起動
uv run agentcore launch

# エージェントをテスト
uv run agentcore invoke '{"prompt": "SSH用の小さなEC2を準備したいです。コストはいくらですか？"}'

agentcore cli は、 bedrock-agentcore-starter-toolkit から使用できるとのこと。

使い方はRuntime Quickstart - Amazon Bedrock AgentCoreを読むといいかも。。。

デプロイしたAIエージェントをPythonから呼び出すには、 boto3 を使用します。

import boto3
import json

client = boto3.client('bedrock-agentcore', region_name='ap-northeast-1')
payload = json.dumps({"prompt": "Explain machine learning in simple terms"})

response = client.invoke_agent_runtime(
    agentRuntimeArn='arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/cost_estimator_agent-xxxxxxx',
    runtimeSessionId='dfmeoagmreaklgmrkleafremoigrmtesogmtrskhmtkrlshmt',  # Must be 33+ chars
    payload=payload,
    qualifier="DEFAULT" # Optional
)
response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)

IAMロールさえ予め作ってしまえば、Lambda的な感覚でデプロイできますね。
ビルドも CodeBuild とローカルで選べるので、柔軟に対応できそうです。

🛡️ AgentCore Identity

AI エージェントのための認証認可機能

AIエージェントをboto3からではなく、直接HTTPリクエストで呼び出すには、SigV4による署名が必要です。
これを Amazon Cognito による OAuth 2.0認証に変更できる機能を持ちます。

以下コマンドで、 Amazon Cognito を作成して、Identityとアクセスに認可が必要な AgentCore Runtime を構築します。

cd 03_identity
uv run python setup_inbound_authorizer.py

以下で認可サーバー（Amazon Cognito）を作成

gateway_client = GatewayClient(region_name=region)
# Use simple interface for creating OAuth authorizer with Cognito from Gateway Client
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
}
save_config({"cognito" : cognito_config})
logger.info("✅ Cognito configuration saved")

Cognitoのclient_idやclient_secret、discovery_urlを渡して、認可サーバーと連携するAgentCore Identityを作成

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
)

最後に、AgentCore Runtime を認可なしには起動できないよう構築します。
authorizer_configを設定することで必要な認可を設定できます。authorizer_configで設定するのは、認可先を表すdiscovery_urlとアプリケーションを識別するためのclient_idです。

AgentCore Runtimeの時は、agentcoreコマンドを使ってデプロイしました。
agentcoreコマンドにもちゃんと--authorizer-configオプションが用意されており、そちらを使えば同じ様に、認可なしで起動できないようにすることができます。

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

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

)

最後に、テストを行いますが、テストは「ローカル上でStrands Agent を動かし、『cost_estimator_tool』を使用して見積もりを出して欲しい」というもの。

やはりクロスリージョンの対応がされていなかったので、以下のように修正

agent = Agent(
    BedrockModel(
        boto_client_config=Config(
            read_timeout=900,
            connect_timeout=900,
            retries=dict(max_attempts=3, mode="adaptive"),
        ),
        model_id="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name="us-east-1"
    ),
    system_prompt=(
        "You are a professional solution architect. "
        "You will receive architecture descriptions or requirements from customers. "
        "Please provide estimate by using 'cost_estimator_tool'"
    ),
    tools=[cost_estimator_tool]
)

cost_estimator_tool上で、M2Mによる認証経由で直接HTTPリクエストで呼び出しています。
なお、@requires_access_tokenと書いておくことで、アクセストークンを取得できます。

# Internal function with authentication decorator
@requires_access_token(
    provider_name=OAUTH_PROVIDER,
    scopes=[OAUTH_SCOPE],
    auth_flow="M2M",
    force_authentication=False
)
async def _cost_estimator_with_auth(architecture_description: str, access_token: str = None) -> str:
    """Internal function that handles the actual API call with authentication"""
    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!")
        # Parse and log JWT token parts for debugging
        log_jwt_token_details(access_token)

    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 Gateway

外部サービスを MCP に変換し接続する機能

Lambdaや外部のMCP Server、REST API、その他の統合サービス（SlackやTeamsなど）をMCPへ変換してくれる機能です。

ここでは、AWS SAMを使って、Lambda + SESのサービスをデプロイします。
このLambda + SESをAgentCore Gatewayを使ってMCPとして動かそうというもの。

AgentCore Gatewayには、以下2種類の認証認可があります。

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

Inboundは、AgentCore Identity / AgentCore Runtime に設定した Cognito を使用します。

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
)

Outbound は、AWS Lambda の場合は AgentCore Gateway の IAM Role (GATEWAY_IAM_ROLE) で認証。GitHub などの外部のサービスであれば OAuth などを設定することになります。

つまり、今回は AWS Lambda で実行するため、AgentCore GatewayにLambdaを実行する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)

最後にテストを行い、ツールとしてMCPを使用しているかを確認しました。

やはりクロスリージョンの対応がされていなかったので、以下のように修正

agent = Agent(
    BedrockModel(
        boto_client_config=Config(
            read_timeout=900,
            connect_timeout=900,
            retries=dict(max_attempts=3, mode="adaptive"),
        ),
        model_id="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name="us-east-1"
    ),
    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 Observability

End to End での動作監視をする機能

CloudWatch の Transaction Search の有効化になっているか、以下コマンドで確認します。
ACTIVE になっていればOKです。

コマンドが認識されない方は、aws cliのアップデートを行ってください。

$ aws xray get-trace-segment-destination
{
    "Destination": "CloudWatchLogs",
    "Status": "ACTIVE"
}

Amazon Bedrock AgentCoreでは、エージェントの動作やユーザーとのやり取りの状況を詳細に監視・分析するため、「Session」「Trace」「Span」という3つの階層構造でオブザーバビリティ（可観測性）が提供されます。

詳細はこちら。

Session
- ユーザーとエージェントが会話する一連の流れ（インタラクション全体）
Trace
- セッション内で発生する個々のリクエスト―レスポンスの流れ。リクエストごとの詳細（タイムスタンプ、入力パラメータ、処理過程、リソース利用状況、エラー内容、最終レスポンスなど）が含まれまる
Span
- トレースの中の個別の作業 (API 呼び出しや内部処理など)

つまり、Session > Trace > Span という単位で監視・分析が可能。

Spanでは、1つのリクエスト―レスポンスの中で、各々の作業の処理時間がわかるようになっていますね。
これは「タイムライン」で見た表示ですが、「トラジェクトリ（ダイアグラム形式）」でも見ることができます。

これができると、どこの処理がボトルネックになっているかとかがわかるので、非常に便利ですね。

🔌 AgentCore Memory

短期・長期記憶機能による文脈認識できる機能

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

短期記憶を長期記憶へ移すのは、Memory Strategyを使用して定義します。
Memory Strategyは、以下4種類あります。

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

ワークショップの処理では、UserPreferenceMemoryStrategyを使用しています。

# Create new memory
logger.info("Creating new AgentCore Memory...")
self.memory = self.memory_client.create_memory_and_wait(
    name=memory_name,
    strategies=[{
        "userPreferenceMemoryStrategy": {
            "name": "UserPreferenceExtractor",
            "description": "Extracts user preferences for AWS architecture decisions",
            "namespaces": [f"/preferences/{self.actor_id}"]
        }
    }],
    event_expiry_days=7,  # Minimum allowed value
)
self.memory_id = self.memory.get('memoryId')
logger.info(f"✅ AgentCore Memory created successfully with ID: {self.memory_id}")

どうやら、Memory Strategyだけ定義しておけば、短期記憶から長期記憶へは自動で行われるため、我々は短期記憶だけを意識しておけばよさそうです。

短期記憶の記録は create_event を使用する。

logger.info(f"🔍 Estimating costs for: {architecture_description}")

# Use the existing cost estimator agent
cost_estimator = AWSCostEstimatorAgent(region=self.region)
result = cost_estimator.estimate_costs(architecture_description)
# Store event in memory
logger.info("Store event to short term memory")
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")
    ]
)

# The memory hook will automatically store this interaction
logger.info("✅ Cost estimation completed")
return result

短期記憶から記録を呼び出すには list_events を使用する。

if not self.memory_client or not self.memory_id:
    return "❌ Memory not available"

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
)

logger.info(f"📋 Found {len(events)} events in memory")
for i, event in enumerate(events):
    logger.info(f"Event {i+1}: {json.dumps(event, indent=2, default=str)}")

return events

長期記憶から情報を検索するには、retrieve_memoriesを使用します。

logger.info("💡 Generating architecture proposal based on user history...")

if not self.memory_client or not self.memory_id:
    return "❌ Memory not available for personalized recommendations"

# Retrieve user preferences and patterns from long-term memory
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]

# Generate proposal using Bedrock
logger.info(f"🔍 Generating proposal with requirements: {requirements}\nHistorical data: {contents}")
proposal_prompt = PROPOSAL_PROMPT_TEMPLATE.format(
    requirements=requirements,
    historical_data="\n".join(contents) if memories else "No historical data available"
)

proposal = self._generate_with_bedrock(proposal_prompt)

logger.info("✅ Architecture proposal generated")
return proposal

長期記憶から情報を検索したcontentsは、以下のような情報が取れる様です。
preferenceがいわゆる好みの情報ですね。

[
    {
        "context":"ユーザーは2つの異なるAWSアーキテクチャ（最初は単一インスタンス設定、今は負荷分散設定）をリクエストしており、これらは異なるスケール要件で動作することを示しています。",
        "preference":"トラフィックレベルと規模の異なるアプリケーションに対応",
        "categories":["クラウドコンピューティング","アプリケーション開発","スケーラビリティ"]
    },
    {
        "context":"ユーザーは、小規模なブログ設定をリクエストした後、中程度のトラフィックのウェブアプリ向けに、負荷分散されたEC2 t3.smallインスタンスとRDS MySQLのAWSコスト見積もりをリクエストしています。これは、ユーザーがよりトラフィックの多いアプリケーションを計画していることを示唆しています。",
        "preference":"中程度のトラフィックのアプリケーションには負荷分散アーキテクチャを優先",
        "categories":["クラウドアーキテクチャ","スケーラビリティ","ウェブアプリケーション"]
    },
    {
        "context":"ユーザーはt3.microインスタンスタイプを具体的にリクエストしており、小規模でコスト効率の高いコンピューティングリソースを優先していることを示しています。",
        "preference":"コスト効率の高い小規模なコンピューティングインスタンスを優先(t3.micro)",
        "categories":["テクノロジー","クラウドコンピューティング","コスト最適化"]
    }
]

さいごに

長々と書きましたが、無事にワークショップを終えることができました。
Amazon Bedrock AgentCoreを使いこなすには、それなりの学習コストが必要ですが、便利な機能が多く、しかもサーバーレス + 安心・安全な環境下でAIエージェントを動かせるのは本当に魅力を感じます。
使い方を学べば、EC2やLambda + Websocket等で構築するよりも、多くのメリットがあるので大変便利だと思いました。

私もとりあえずワークショップを通じて体験はできたので、他にも何ができるか色々と試していきたいと思います！！
👋👋