mcpField
B
bedrock-agentcore

bedrock-agentcore

github.com/adaptationioクラウド・DevOps47 回インストール
リポジトリ

bedrock-agentcore について

Amazon Bedrock AgentCore は、AI エージェントの構築、デプロイ、運用に使うマネージドプラットフォームで、AWS 上で本番エージェントを扱うチーム向けです。ランタイム、ツールゲートウェイ、ブラウザーやコード実行、ID連携、可観測性を組み合わせて利用できます。

主な機能

  1. 01マネージドエージェントランタイム
  2. 02APIやLambda関数をツール化
  3. 03ブラウザーワークフローを安全に実行
  4. 04コード実行とデータ分析
  5. 05IDとアクセストークンの連携
  6. 06CloudWatchとOpenTelemetryによる可観測性

活用シーン

  1. 01AWSで本番AIエージェントをデプロイ
  2. 02既存APIやLambda関数をエージェントに公開
  3. 03ウェブ自動化タスクを実行
  4. 04分離されたコードセッションでデータ分析
  5. 05エージェントの呼び出し、遅延、エラーを監視

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/adaptationio/skrillz --skill bedrock-agentcore

インストール後、Claude Code で/bedrock-agentcoreを入力して読み込まれたか確認します。

詳細

Amazon Bedrock AgentCore

概要

Amazon Bedrock AgentCore は、インフラストラクチャを管理することなく、効果的な AI エージェントを安全に大規模に構築、デプロイ、運用するためのエージェントプラットフォームです。一般的なオープンソースのフレームワーク(Strands、LangGraph、CrewAI、Autogen)やあらゆるモデルで動作する、フレームワークに依存しないプリミティブを提供します。

目的:あらゆる AI エージェントを、エンタープライズグレードのインフラストラクチャを備えた本番環境対応アプリケーションに変換します

パターン:機能ベース(独立した 6 つのサービスモジュール)

主な原則(AWS により 2025 年 12 月に検証済み):

  1. フレームワーク非依存 - あらゆるエージェントフレームワークやモデルで動作します
  2. インフラストラクチャ不要 - 完全マネージド型で、運用の負担がありません
  3. セッション分離 - セッション間でデータを完全に分離します
  4. エンタープライズセキュリティ - VPC、PrivateLink、ID 統合に対応します
  5. 組み合わせ可能なサービス - 必要なサービスのみを使用できます
  6. 本番環境対応 - 拡張性、信頼性、安全性を考慮して構築されています

品質目標:

  • デプロイ:コードから本番環境まで 5 分未満
  • レイテンシ:低レイテンシから 8 時間の非同期ワークロードまで対応
  • オブザーバビリティ:CloudWatch と完全に統合

使用する場面

次のような場合に bedrock-agentcore を使用します。

  • AWS 上で本番環境向けの AI エージェントを構築する場合
  • エージェントのデプロイにマネージドインフラストラクチャが必要な場合
  • セッション分離とエンタープライズセキュリティが必要な場合
  • 既存のエージェントフレームワーク(LangGraph、CrewAI など)を使用する場合
  • ブラウザーの自動化やコード実行機能が必要な場合
  • 既存の ID プロバイダーと統合する場合

使用しない場面:

  • シンプルな Bedrock モデル呼び出し(bedrock-runtime を使用)
  • アクショングループを備えた標準の Bedrock Agents(bedrock-agent を使用)
  • AWS 以外へのデプロイ

前提条件

必須
  • Bedrock にアクセスできる AWS アカウント
  • AgentCore サービスの IAM 権限
  • Python 3.10+(SDK 用)
推奨
  • bedrock-agentcore-sdk-python のインストール
  • bedrock-agentcore-starter-toolkit CLI
  • 基盤モデルへのアクセスを有効化(Claude など)
インストール
# Install SDK and CLI
pip install bedrock-agentcore strands-agents bedrock-agentcore-starter-toolkit

# Verify installation
agentcore --help

コアサービス

1. AgentCore Runtime

エージェントコードを実行するための、安全でセッションが分離されたコンピューティング環境です。

Boto3 クライアント:

import boto3

# Data plane operations
client = boto3.client('bedrock-agentcore')

# Control plane operations
control = boto3.client('bedrock-agentcore-control')

エージェントランタイムの作成:

# Using starter toolkit
# agentcore configure -e main.py -n my-agent
# agentcore deploy

# Using boto3 control plane
response = control.create_agent_runtime(
    name='my-production-agent',
    description='Customer service agent',
    agentRuntimeArtifact={
        's3': {
            'uri': 's3://my-bucket/agent-package.zip'
        }
    },
    roleArn='arn:aws:iam::123456789012:role/AgentCoreExecutionRole',
    pythonRuntime='PYTHON_3_13',
    entryPoint=['main.py']
)
agent_runtime_arn = response['agentRuntimeArn']

エージェントの呼び出し:

# Invoke deployed agent
response = client.invoke_agent_runtime(
    agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/xxx',
    runtimeSessionId='session-123',
    payload={
        'prompt': 'What is my order status?',
        'context': {'user_id': 'user-456'}
    }
)

result = response['payload']
print(result)

エージェントのエントリーポイント構造:

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp(debug=True)
agent = Agent()

@app.entrypoint
def invoke(payload):
    """Main agent entry point"""
    user_message = payload.get("prompt", "Hello!")
    app.logger.info(f"Processing: {user_message}")

    result = agent(user_message)
    return {"result": result.message}

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

2. AgentCore Gateway

既存の API と Lambda 関数を、セマンティック検索による検出に対応した、エージェント互換のツールに変換します。

ゲートウェイの作成:

response = control.create_gateway(
    name='customer-service-gateway',
    description='Gateway for customer service tools',
    protocolType='REST'
)
gateway_arn = response['gatewayArn']

ゲートウェイターゲット(ツール)の追加:

# Add an existing Lambda as a tool
response = control.create_gateway_target(
    gatewayId='gateway-xxx',
    name='GetOrderStatus',
    description='Retrieves order status by order ID',
    targetConfiguration={
        'lambdaTarget': {
            'lambdaArn': 'arn:aws:lambda:us-east-1:123456789012:function:GetOrder'
        }
    },
    toolSchema={
        'name': 'get_order_status',
        'description': 'Get the current status of a customer order',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'order_id': {
                    'type': 'string',
                    'description': 'The unique order identifier'
                }
            },
            'required': ['order_id']
        }
    }
)

ツールの同期:

# Sync gateway tools for discovery
control.synchronize_gateway_targets(
    gatewayId='gateway-xxx'
)

3. Browser Runtime

複雑なウェブベースのワークフローを安全に実行します。

ブラウザーセッションの開始:

response = client.start_browser_session(
    browserId='browser-xxx',
    sessionConfiguration={
        'timeout': 300,
        'viewport': {'width': 1920, 'height': 1080}
    }
)
session_id = response['browserSessionId']

ブラウザー操作の実行:

# Navigate and interact
response = client.update_browser_stream(
    browserSessionId=session_id,
    action={
        'navigate': {'url': 'https://example.com'},
        'click': {'selector': '#submit-button'},
        'type': {'selector': '#search', 'text': 'query'}
    }
)

4. Code Interpreter

データ分析や可視化などのタスクでコードを安全に実行します。

Code Interpreter セッションの開始:

response = client.start_code_interpreter_session(
    codeInterpreterId='interpreter-xxx'
)
session_id = response['codeInterpreterSessionId']

コードの実行:

response = client.invoke_code_interpreter(
    codeInterpreterSessionId=session_id,
    code='''
import pandas as pd
import matplotlib.pyplot as plt

# Analyze data
df = pd.DataFrame({'x': [1,2,3,4,5], 'y': [2,4,6,8,10]})
plt.plot(df['x'], df['y'])
plt.savefig('output.png')
print(df.describe())
''',
    language='PYTHON'
)

output = response['output']
files = response['files']  # Generated files

5. ID 統合

認証と権限の委任のために、既存の ID プロバイダーとネイティブに統合します。

OAuth2 プロバイダーの作成:

response = control.create_oauth2_credential_provider(
    name='okta-provider',
    credentialProviderVendor='OKTA',
    oauth2ProviderConfig={
        'clientId': 'your-client-id',
        'clientSecret': 'your-client-secret',
        'authorizationServerUrl': 'https://your-domain.okta.com/oauth2/default',
        'scopes': ['openid', 'profile', 'email']
    }
)

ワークロード ID の作成:

response = control.create_workload_identity(
    name='agent-identity',
    allowedRoleArns=['arn:aws:iam::123456789012:role/AgentRole']
)

アクセストークンの取得:

# Get token for workload
response = client.get_workload_access_token(
    workloadIdentityId='identity-xxx'
)
access_token = response['accessToken']

6. オブザーバビリティ

CloudWatch と OpenTelemetry によるリアルタイムの可視性を提供します。

オブザーバビリティの有効化:

# In your agent entry point
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

# Configure tracing
provider = TracerProvider()
trace.set_tracer_provider(provider)

# Entry point with OTel
# entryPoint: ['opentelemetry-instrument', 'main.py']

CloudWatch メトリクス:

  • セッションごとのトークン使用量
  • レイテンシ(p50、p95、p99)
  • セッション時間
  • エラー率
  • ツール呼び出しの成功/失敗

Boto3 クライアントリファレンス

データプレーン(bedrock-agentcore)
メソッド目的
invoke_agent_runtimeエージェントのロジックを実行
stop_runtime_sessionアクティブなセッションを停止
list_sessionsすべてのセッションを一覧表示
start_browser_sessionブラウザーを初期化
stop_browser_sessionブラウザーセッションを終了
start_code_interpreter_sessionインタープリターを起動
invoke_code_interpreterコードを実行
batch_create_memory_recordsメモリーを作成
retrieve_memory_recordsメモリーを取得
get_workload_access_token認証トークンを取得
evaluate評価を実行
コントロールプレーン(bedrock-agentcore-control)
メソッド目的
create_agent_runtimeランタイムを作成
delete_agent_runtimeランタイムを削除
update_agent_runtimeランタイムを変更
create_gatewayゲートウェイを作成
create_gateway_targetツールを追加
create_memoryメモリストアを作成
create_policyポリシーを作成
create_evaluator評価器を作成
create_browserブラウザーを作成
create_code_interpreterインタープリターを作成

クイックスタート:Hello World エージェント

ステップ 1:エージェントファイルを作成
# main.py
from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp()
agent = Agent(model="anthropic.claude-sonnet-4-20250514-v1:0")

@app.entrypoint
def invoke(payload):
    prompt = payload.get("prompt", "Hello!")
    result = agent(prompt)
    return {"response": result.message}

if __name__ == "__main__":
    app.run()
ステップ 2:設定してデプロイ
# Configure
agentcore configure -e main.py -n hello-world-agent

# Test locally
python main.py &
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Hello!"}'

# Deploy to AWS
agentcore deploy

# Test deployed
agentcore invoke '{"prompt": "Hello from production!"}'
ステップ 3:プログラムから呼び出す
import boto3

client = boto3.client('bedrock-agentcore')

response = client.invoke_agent_runtime(
    agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/hello-world',
    runtimeSessionId='test-session-1',
    payload={'prompt': 'What can you help me with?'}
)

print(response['payload'])

エラー処理

from botocore.exceptions import ClientError

try:
    response = client.invoke_agent_runtime(
        agentRuntimeArn=agent_arn,
        runtimeSessionId='session-1',
        payload={'prompt': 'test'}
    )
except ClientError as e:
    error_code = e.response['Error']['Code']

    if error_code == 'ResourceNotFoundException':
        print("Agent runtime not found")
    elif error_code == 'ValidationException':
        print("Invalid request parameters")
    elif error_code == 'ThrottlingException':
        print("Rate limited - implement backoff")
    elif error_code == 'AccessDeniedException':
        print("Check IAM permissions")
    else:
        raise

IAM 権限

最小限の実行ロール
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "bedrock-agentcore:InvokeAgentRuntime",
        "bedrock-agentcore:StartBrowserSession",
        "bedrock-agentcore:InvokeCodeInterpreter"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream"
      ],
      "Resource": "arn:aws:bedrock:*::foundation-model/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    }
  ]
}

関連スキル

  • bedrock-agentcore-policy:Cedar ポリシーの作成と適用
  • bedrock-agentcore-evaluations:エージェントのテストと品質評価
  • bedrock-agentcore-memory:エピソード記憶と短期記憶の管理
  • bedrock-agentcore-deployment:本番環境へのデプロイパターン
  • bedrock-agentcore-multi-agent:マルチエージェントのオーケストレーション(A2A プロトコル)
  • boto3-eks:EKS でホストされるエージェント向け
  • terraform-aws:コードとしてのインフラストラクチャ

参考資料

  • references/gateway-configuration.md - ゲートウェイ設定の詳細
  • references/identity-integration.md - OAuth とワークロード ID
  • references/troubleshooting.md - よくある問題と解決方法

ソース

コメント