吾輩は人間である
前書き
2025年07月17日、New Yorkで開催されたAWS Summitにおいて、Amazon Bedrock AgentCoreというサービスが発表されました。
このサービスは名前にAmazon Bedrockと付いていますが、実際には別のサービスです。
マネージメントコンソールのカテゴリも、従来のAmazon Bedrockとは独立して配置されています。
では、従来のAmazon Bedrockエージェントと何が違うのでしょうか。
個人的には、SaaSとPaaSの違いに相当するのではないかと考えています。
従来のBedrockエージェントは、Action Groupで多少のコーディングが必要とはいえ、基本的にはSaaSとして提供されていました。
一方、Amazon Bedrock AgentCoreは、Amazonがホスティング環境を提供し、その他の開発や技術選択についてはビルダーに委ねられています。つまり、PaaSのような位置づけになっていると言えるでしょう。
生成AIのトレンドが頻繁に変わる中、Amazon Bedrock AgentCoreのような、開発者にプラットフォームだけを提供するサービスは、個人的にずっと欲しいと思っていました
ハンズオン
Line Bot用のAIエージェントを作って、Amazon Bedrock AgentCoreにデプロイしてみましょう。
TypeScriptのランタイムがまだ出ていないので、Strands Agentsを使って進めていきます。
Strands AgentsはAWSが出しているOSSのフレームワークで、こちらも同じNew York Summitで正式リリースされました。
前提条件
- AWSアカウント
- AWS SAM CLI
- uv
- Docker Desktop
特にAWS SAM CLIとuvはできれば最新版にしてください。
1. 環境のセットアップ
好きなディレクトリで下記のコマンドを実行してください。
% uv venv
% source .venv/bin/activate
% uv init
% uv add strands-agents bedrock-agentcore bedrock-agentcore-starter-toolkit
同じディレクトリにmain.pyを作って、以下の内容を書いてください。
main.py
from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
bedrock_model = BedrockModel(
region_name="us-east-1",
model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
)
agent = Agent(
model=bedrock_model,
system_prompt="You are a helpful AI assistant"
)
@app.entrypoint
def invoke(payload):
"""Process user input and return a response"""
user_message = payload.get("prompt","Hello")
result = agent(user_message)
return {"result": result.message}
if __name__ == "__main__":
app.run()
AWSアカウントの設定は済んでいる前提で、ローカルでテストしてみます。
% uv run main.py
INFO: Started server process [31687]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8080 (Press CTRL+C to quit)
% curl -X POST http://localhost:8080/invocations -H "Content-Type: application/json" -d '{"prompt": "こんにちは!"}'
{"result":{"role":"assistant","content":[{"text":"こんにちは！元気ですか？何かお手伝いできることがあれば、お気軽にお声かけください。"}]}}
2. AWSリソース作成
Amazon Bedrock AgentCoreのCDKがまだ出ていないので、少し手作業になりますが、まず必要なリソースを準備します。
同じディレクトリで下記の内容を実行していきます。
ECR リポジトリの作成
YOU_REPOSITORY_NAMEは実際作成するよっていの名前に置き換えてください。
% aws ecr create-repository \
--repository-name YOU_REPOSITORY_NAME \
--region us-east-1
リポジトリ作られたら、コマンドラインからログインもしてください。
AWS IAM ロールの設定
信頼ポリシーファイルの作成。
個人的には公式より少し緩めの権限設定にしていますが、公式ドキュメントはこちらなので、必要に応じて適宜修正してください。
cat > trust-policy.json << EOF
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "bedrock-agentcore.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
実行権限ポリシーファイルの作成、
ACCOUNT_ID、YOU_REPOSITORY_NAMEは実際のものに置き換えてください。
cat > execution-policy.json << EOF
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "ECRImageAccess",
"Effect": "Allow",
"Action": [
"ecr:BatchGetImage",
"ecr:GetDownloadUrlForLayer"
],
"Resource": [
"arn:aws:ecr:us-east-1:YOUR_ACCOUNT_ID:repository/YOU_REPOSITORY_NAME"
]
},
{
"Effect": "Allow",
"Action": [
"logs:*"
],
"Resource": [
"arn:aws:logs:us-east-1:YOUR_ACCOUNT_ID:*"
]
},
{
"Sid": "ECRTokenAccess",
"Effect": "Allow",
"Action": [
"ecr:GetAuthorizationToken"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"xray:*"
],
"Resource": [ "*" ]
},
{
"Effect": "Allow",
"Resource": "*",
"Action": "cloudwatch:*"
},
{
"Sid": "GetAgentAccessToken",
"Effect": "Allow",
"Action": [
"bedrock-agentcore:*"
],
"Resource": "*"
},
{
"Sid": "BedrockModelInvocation",
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": [
"arn:aws:bedrock:*::foundation-model/*",
"arn:aws:bedrock:us-east-1:YOUR_ACCOUNT_ID:*"
]
}
]
}
IAMロールの作成
aws iam create-role \
--role-name BedrockAgentCoreExecutionRole \
--assume-role-policy-document file://trust-policy.json \
--region us-east-1
インラインポリシーのアタッチ
aws iam put-role-policy \
--role-name BedrockAgentCoreExecutionRole \
--policy-name BedrockAgentCoreExecutionPolicy \
--policy-document file://execution-policy.json \
--region us-east-1
完了後、ロールARNの取得して、後で使用する
aws iam get-role \
--role-name BedrockAgentCoreExecutionRole \
--query 'Role.Arn' \
--output text
3. AWSにデプロイ
同じディレクターに下記のファイルを作ってください。
requirements.txt
strands-agents
bedrock-agentcore
IAM_ROLE_ARNの環境変数を設定します。
% export IAM_ROLE_ARN=<先ほど取得したARN>
下記のコマンドを実行すると、いくつかの項目について質問されます。
% agentcore configure --entrypoint main.py -er $IAM_ROLE_ARN
ECRのURIはAWSのマネコンから確認できます、
dependency fileはrequirements.txtを指定します、
OAuthについては、今回使用しないため「no」で問題ありません。
完了すると、同じディレクトリに.bedrock_agentcore.yamlが生成されます。
基本的にはそのままの設定で大丈夫ですので、下記のコマンドを実行してデプロイしましょう。
% agentcore launch
もしECRの公開リポジトリからイメージpullできない関連のエラーが出たら、下記のコマンドも試してみてください。
% aws ecr-public get-login-password --region us-east-1 | docker login --username AWS --password-stdin public.ecr.aws
デプロイ完了したら、下記の内容を確認できます。
Amazon Bedrock AgentCoreのAgent RuntimeでデプロイされたAgentも確認できます。
接続テストもしてみましょう！
デプロイ成功していれば、responseからAgentの出力を確認できるはずです。
% agentcore invoke '{"prompt": "hello world!"}'
Payload:
{
"prompt": "hello world!"
}
Invoking BedrockAgentCore agent 'main' via cloud endpoint
Found credentials in environment variables.
Session ID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
Response:
{
"ResponseMetadata": {
"RequestId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
"HTTPStatusCode": 200,
"HTTPHeaders": {
...
},
"RetryAttempts": 0
},
"runtimeSessionId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
"traceId": "Root=x-xxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx;Self=x-xxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx",
"baggage": "Self=x-xxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx,session.id=xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
"contentType": "application/json",
"statusCode": 200,
"response": [
"b'{\"result\":{\"role\":\"assistant\",\"content\":[{\"text\":\"Hello! Nice to meet you! How are you doing today? Is there anything I can help you
with?\"}]}}'"
]
}
一度でもリクエストを送信していれば、トレースデータも取得できているはずです。
もし下記のようにトレースデータが取得できない場合は、X-RayのTransaction Searchが有効化されていないことが原因だと考えられます。
CloudWatchのGenAI Observabilityの通知メッセージから、またはCloudWatchの設定にあるApplication Signalsから有効化することができます。
X-RayのTransaction Searchを有効化します。
再度ローカルからチャットしてみます。
% agentcore invoke '{"prompt": "Tell me what is langchan?"}'
その後ロググループからaws/spansが作られたことが確認できます。
トレース取れるようになりました。
実際のチャットトレースはSessionsから該当Sessionの利用モデルのEventsから確認できます。
Line Bot作ってみよう
LINE Botとは、ユーザーのチャットメッセージに対して自動的に返答するLINE用ロボット
これまでの構成を利用して、簡単なAPI GatewayとLambdaを追加するだけです。
LINEからでなくても動作するため、通常のAPIとして呼び出しても動く構成になっています。
LINE Developersコンソールから新規チャンネルを作成し、Messaging APIを選択します。
チャネルシークレットとチャネルアクセストークンを取得してメモしておきます。
現在、Messaging APIチャネルの作成にはLINE公式アカウントの作成が必要になっているため、単体のAPI Gatewayでも十分動作しますので、無理に作成する必要はありません。
SAMでAPIをデプロイします
先ほど作成したStrandsAgentディレクトリに戻り、任意の名前のフォルダを作成します。
その中に、lambda_handler.py、requirements.txt、template.yamlを作成していきます。
ディレクトリ構成は下記の通りです。
your_project_directory/
├── {任意の名前のフォルダ}
├──├── lambda_handler.py
├──├── requirements.txt
├──├── template.yaml
├── main.py
├── requirements.txt
└── ...その他
ローカルでもテストする場合、boto3も入れましょう。
requirements.txt
boto3>=1.34.0
line-bot-sdk==3.14.2
YOUR_ACCOUNT_IDとYOUR_RUNTIME_IDはご自身のものに置き換えてくださいLINEでテストしたい場合LineChannelAccessToken,LineChannelSecretも置き換えましょう。
template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
SAM Template for Bedrock AgentCore Lambda API
Parameters:
BedrockAgentRuntimeArn:
Type: String
Default: "arn:aws:bedrock-agentcore:us-east-1:YOUR_ACCOUNT_ID:runtime/YOUR_RUNTIME_ID"
Description: ARN of the Bedrock Agent Runtime
LineChannelAccessToken:
Type: String
Default: "YOUR_LINE_CHANNEL_ACCESS_TOKEN"
Description: LINE Channel Access Token
LineChannelSecret:
Type: String
Default: "YOUR_LINE_CHANNEL_SECRET"
Description: LINE Channel Secret
Globals:
Function:
Timeout: 30
MemorySize: 128
Runtime: python3.13
Tracing: PassThrough
Environment:
Variables:
BEDROCK_AGENT_RUNTIME_ARN: !Ref BedrockAgentRuntimeArn
LINE_CHANNEL_ACCESS_TOKEN: !Ref LineChannelAccessToken
LINE_CHANNEL_SECRET: !Ref LineChannelSecret
Resources:
BedrockAgentCoreFunction:
Type: AWS::Serverless::Function
Properties:
FunctionName: bedrock-agent-core
CodeUri: ./
Handler: lambda_handler.lambda_handler
Runtime: python3.13
Tracing: PassThrough
Events:
BedrockAgentCoreApi:
Type: Api
Properties:
Path: /invoke
Method: post
RestApiId: !Ref BedrockAgentCoreApi
Policies:
- BedrockAgentCoreFullAccess
- Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- logs:CreateLogGroup
- logs:CreateLogStream
- logs:PutLogEvents
Resource: '*'
BedrockAgentCoreApi:
Type: AWS::Serverless::Api
Properties:
StageName: prod
Cors:
AllowMethods: "'GET,POST,OPTIONS'"
AllowHeaders: "'Content-Type,X-Amz-Date,Authorization,X-Api-Key,X-Amz-Security-Token'"
AllowOrigin: "'*'"
Outputs:
BedrockAgentCoreApi:
Description: "API Gateway endpoint URL for Prod stage"
Value: !Sub "https://${BedrockAgentCoreApi}.execute-api.${AWS::Region}.amazonaws.com/prod/invoke"
BedrockAgentCoreFunction:
Description: "Bedrock Agent Core Lambda Function ARN"
Value: !GetAtt BedrockAgentCoreFunction.Arn
BedrockAgentCoreFunctionIamRole:
Description: "Implicit IAM Role created for Bedrock Agent Core function"
Value: !GetAtt BedrockAgentCoreFunctionRole.Arn
lambda_handler.py
import boto3
import json
import os
import uuid
from typing import Dict, Any
from linebot.v3.messaging import (
Configuration,
ApiClient,
MessagingApi,
ReplyMessageRequest,
TextMessage
)
# Initialize LINE Bot API configuration globally
LINE_CONFIGURATION = None
if 'LINE_CHANNEL_ACCESS_TOKEN' in os.environ:
LINE_CONFIGURATION = Configuration(access_token=os.environ['LINE_CHANNEL_ACCESS_TOKEN'])
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
"""
Lambda handler for Bedrock AgentCore with LINE Message API support
"""
try:
# Parse request body
if 'body' not in event:
return {
'statusCode': 400,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({'error': 'Missing request body'})
}
# Parse JSON body
body = json.loads(event['body']) if isinstance(event['body'], str) else event['body']
# Check if this is a LINE webhook event
if 'events' in body and len(body['events']) > 0:
return handle_line_webhook(body)
# Original API Gateway handling
prompt = body.get('prompt', 'Hello')
# Call bedrock-agentcore
result = call_bedrock_agentcore(prompt)
return {
'statusCode': 200,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({
'result': result
})
}
except json.JSONDecodeError:
return {
'statusCode': 400,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({'error': 'Invalid JSON in request body'})
}
except Exception as e:
return {
'statusCode': 500,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*'
},
'body': json.dumps({'error': str(e)})
}
def handle_line_webhook(body: Dict[str, Any]) -> Dict[str, Any]:
"""
Handle LINE webhook events
"""
try:
if body['events'][0]['type'] == 'message':
if body['events'][0]['message']['type'] == 'text':
message = body['events'][0]['message']['text']
# Call bedrock-agentcore
result = call_bedrock_agentcore(message)
# Send reply via LINE
if LINE_CONFIGURATION:
try:
with ApiClient(LINE_CONFIGURATION) as api_client:
line_bot_api = MessagingApi(api_client)
line_bot_api.reply_message_with_http_info(
ReplyMessageRequest(
reply_token=body['events'][0]['replyToken'],
messages=[TextMessage(text=str(result))]
)
)
except Exception as line_error:
print(f"ERROR: Failed to send LINE message: {str(line_error)}")
except Exception as e:
print(f"ERROR: LINE webhook processing failed: {str(e)}")
return {
'statusCode': 200,
'body': json.dumps('Success!')
}
def call_bedrock_agentcore(prompt: str) -> str:
"""
Call Bedrock AgentCore and return the result
"""
# Initialize the Bedrock AgentCore client
agent_core_client = boto3.client('bedrock-agentcore')
# Get agent runtime ARN from environment
agent_arn = os.environ.get('BEDROCK_AGENT_RUNTIME_ARN')
if not agent_arn:
return 'Agent runtime ARN not configured'
# Prepare the payload
payload = json.dumps({"prompt": prompt}).encode()
# Invoke the agent
response = agent_core_client.invoke_agent_runtime(
agentRuntimeArn=agent_arn,
contentType="application/json",
payload=payload,
traceId=str(uuid.uuid4()).replace('-', ''),
)
# Process the response
result = process_response(response)
# Extract text content from the result
if isinstance(result, dict) and 'result' in result:
inner_result = result['result']
if isinstance(inner_result, dict) and 'content' in inner_result:
content = inner_result['content']
if isinstance(content, list) and len(content) > 0:
first_item = content[0]
if isinstance(first_item, dict) and 'text' in first_item:
return first_item['text']
# Fallback to string conversion
return str(result)
def process_response(response: Dict[str, Any]) -> Any:
"""
Process and return the response from Bedrock AgentCore
"""
try:
# Handle standard JSON response
if response.get("contentType") == "application/json":
content = []
response_data = response.get("response", [])
for chunk in response_data:
decoded_chunk = chunk.decode('utf-8')
content.append(decoded_chunk)
full_content = ''.join(content)
return json.loads(full_content)
# Handle other content types
else:
return response
except Exception as e:
return f"Error processing response: {str(e)}"
次はSAM CLIでデプロイしましょう!
% sam build
% sam deploy --guided
Setting default arguments for 'sam deploy'
=========================================
Stack Name [sam-app]: bedrock-core-api
AWS Region [us-east-1]:
Parameter BedrockAgentRuntimeArn [YOU_AGENT_ARN]:
#Shows you resources changes to be deployed and require a 'Y' to initiate deploy
Confirm changes before deploy [y/N]: y
#SAM needs permission to be able to create roles to connect to the resources in your template
Allow SAM CLI IAM role creation [Y/n]: Y
#Preserves the state of previously provisioned resources when an operation fails
Disable rollback [y/N]: N
BedrockAgentCoreFunction has no authentication. Is this okay? [y/N]: y
Save arguments to configuration file [Y/n]: Y
SAM configuration file [samconfig.toml]:
SAM configuration environment [default]:
デプロイが完了したら、API GatewayのAPIが確認できます。
% curl -X POST https://xxxxxxxxxx.execute-api.us-east-1.amazonaws.com/prod/invoke -H "Content-Type: application/json" -d '{"prompt": "Hello, how are you?"}'
{"result": {"result": {"role": "assistant", "content": [{"text": "Hello! I'm doing well, thank you for asking. I'm here and ready to help with whatever you'd like to discuss or work on. How are you doing today?"}]}}}%
LINEの公式アカウントでテストします。
Webhook URLをAPI GatewayのURLに変更します。
QRコードで友達追加して、メッセージ送ってみましょう!
返信が確認でき、トレースも取得できました。完璧です。
実際に使用したリポジトリはこちらになります。よろしければ参考にしてください。
参考資料