どうもこんにちは。
今回は、Strands AgentsからAmazon Bedrock AgentCore Runtime上にデプロイしたMCPサーバを呼び出す方法をまとめました。
この記事で学べること
- ローカル環境で構築したStrandsAgentsからAmazon Bedrock AgentCore Runtime上にデプロイしたMCPサーバへ接続する方法
- Runtime上にデプロイしたStrandsAgentsからAmazon Bedrock AgentCore Runtime上にデプロイしたMCPサーバへ接続する方法
関連記事
以下の記事でStrandsAgentsとMCPサーバを同一Runtime上へデプロイする方法を記載しています。
Amazon Bedrock AgentCore Runtime上にデプロイする方法は以下の記事で記載しています。
今回はこちらの記事の方が内容は近いかなと思います。
前提条件
- 何かしらのMCPサーバをAmazon Bedrock AgentCore Runtime上へデプロイする（IAM認証を使用する想定）
デプロイ時に使用したスクリプトです。（starter-tool-kitを使用していないので。）
- あらかじめECRにDockerイメージをデプロイしておく必要があります。こちらの記事を参考にデプロイしてみてください
import boto3
account_id = 'xxxxxxxxxxxxxxxx'
region_name = '<リージョン>'
runtime_name = 'sample_mcp_1'
agent_runtime_id = 'sample_mcp_1-xxxxxxxxxx'
client = boto3.client('bedrock-agentcore-control', region_name=region_name)
try:
# エージェントランタイム取得
response = client.get_agent_runtime(
agentRuntimeId=agent_runtime_id
)
update_response = client.update_agent_runtime(
agentRuntimeId=agent_runtime_id,
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f'{account_id}.dkr.ecr.{region_name}.amazonaws.com/{runtime_name}:latest'
}
},
protocolConfiguration={
'serverProtocol': 'MCP' # プロトコルを`MCP`に設定
},
networkConfiguration={"networkMode": "PUBLIC"},
roleArn=f'arn:aws:iam::{account_id}:role/SampleRole'
)
print(f"Agent Runtime updated successfully!")
print(f"Agent Runtime ARN: {update_response['agentRuntimeArn']}")
print(f"Status: {update_response['status']}")
except Exception as e:
print('エージェントランタイムの取得または更新に失敗しました。新規作成を試みます。')
create_response = client.create_agent_runtime(
agentRuntimeName=runtime_name,
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f'{account_id}.dkr.ecr.{region_name}.amazonaws.com/{runtime_name}:latest'
}
},
protocolConfiguration={
'serverProtocol': 'MCP' # プロトコルを`MCP`に設定
},
networkConfiguration={"networkMode": "PUBLIC"},
roleArn=f'arn:aws:iam::{account_id}:role/SampleRole'
)
print(f"Agent Runtime created successfully!")
print(f"Agent Runtime ARN: {create_response['agentRuntimeArn']}")
print(f"Status: {create_response['status']}")
AWSマネジメントコンソールへデプロイした時に、以下のようになっていればOKです。
ローカルでの構築手順
1. エージェントの実行環境を用意する
こちらの記事とほぼ同じですが、インストールするライブラリが変わります。
$ uv init --python 3.13
$ uv add 'uvicorn[standard]' fastapi httpx pydantic bedrock-agentcore strands-agents mcp-proxy-for-aws
treeコマンドを実行したときに、以下のようになっていればOKです。
.
├── agent.py # main.pyになっている場合は変更しておく
├── pyproject.toml
├── README.md
└── uv.lock
2. エージェントのコードを記述
aws_iam_streamablehttp_clientを使用して、IAM認証をしてMCPサーバと接続するように記述します。
from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
from bedrock_agentcore import BedrockAgentCoreApp
from strands.tools.mcp import MCPClient
from strands import Agent
app = BedrockAgentCoreApp()
agent = Agent()
agent_arn = "<MCPサーバがデプロイされているRuntimeのARN>"
encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
region = "<リージョン>"
mcp_url = f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations"
mcp_client_factory = lambda: aws_iam_streamablehttp_client(
endpoint=mcp_url,
aws_region=region,
aws_service="bedrock-agentcore",
terminate_on_close=False
)
mcp_client = MCPClient(mcp_client_factory)
agent = Agent(tools=[mcp_client])
print(f"Available tools: {agent.tool_names}", flush=True)
@app.entrypoint
def invoke(payload):
user_message = payload.get("prompt", "Hello!")
result = agent(user_message)
return {"result": result.message}
if __name__ == "__main__":
app.run()
3. エージェント起動
ターミナルで以下を実行すると、エージェントサーバが起動します。
$ uv run uvicorn agent:app --host 0.0.0.0 --port 8080
正常に起動されていれば、Available tools: ....のように使用可能なツールがログ出力されます。
4. curlコマンドでMCPに接続できるか確認
以下のようにリクエストを送信します。
$ curl -X POST http://localhost:8080/invocations \
-H "Content-Type: application/json" \
-d '{"prompt": "<〇〇ツールを使用して〇〇してください。>"}'
正常に接続がされていれば、ちゃんとツールを使った結果が返ってくるはずです。
Runtime上へデプロイしてから呼び出し
1. Dockerfile準備
Bedrock AgentCoreでは、ECRにDockerイメージをデプロイする必要があるので、Dockerfileを用意します。まず、以下のコマンドを実行します。
$ touch Dockerfile
Dockerfileが作成されるので、以下のコードをペーストします。
# Use uv's ARM64 Python base image
FROM --platform=linux/arm64 ghcr.io/astral-sh/uv:python3.13-bookworm-slim
WORKDIR /app
# Copy uv files
COPY pyproject.toml uv.lock ./
# Install dependencies (including strands-agents)
RUN uv sync --frozen --no-cache
# Copy agent file
COPY agent.py ./
# Expose port
EXPOSE 8080
# Run application
CMD ["uv", "run", "uvicorn", "agent:app", "--host", "0.0.0.0", "--port", "8080"]
Dockerfileでポート番号を8081とかにしてデプロイするとちゃんと動かないので、ポート番号は8080で固定しておいてください！
MCPサーバの時は8000です！
2. Dockerイメージをビルド
ターミナルで以下のコマンドを実行します。
$ docker buildx create --use
$ docker buildx build --platform linux/arm64 -t sample_agent_1:arm64 --load .
ここは確認なので飛ばしてOK
docker runコマンドでイメージを起動します。
$ docker run --platform linux/arm64 -p 8080:8080 \
-e AWS_ACCESS_KEY_ID="$AWS_ACCESS_KEY_ID" \
-e AWS_SECRET_ACCESS_KEY="$AWS_SECRET_ACCESS_KEY" \
-e AWS_SESSION_TOKEN="$AWS_SESSION_TOKEN" \
-e AWS_REGION="$AWS_REGION" \
sample_agent_1:arm64
以下のように、ログが出てくればOKです。
INFO: Started server process [10]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
3. ECRにログインする
以下のコマンドを実行します。
$ export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
$ export AWS_REGION=ap-north-east-1
$ aws ecr get-login-password --region ${AWS_REGION} | docker login --username AWS --password-stdin ${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com
4. DockerイメージをビルドしてECRにデプロイ
以下のコマンドを実行します。
# 初回デプロイの時だけ実行
$ aws ecr create-repository \
--repository-name sample_agent_1 \
--region us-east-1
# デプロイの度に実行
$ docker buildx build --platform linux/arm64 -t ${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com/sample_agent_1:latest --push .
ここは確認なので飛ばしてOK
以下のコマンドを実行します。
$ aws ecr describe-images --repository-name sample_agent_1 --region ${AWS_REGION}
これで、ECR上のDockerイメージがデプロイされたことが確認できました。
5. デプロイスクリプト実行
以下のコマンドでスクリプトファイルを作成したら、
$ touch deploy_agent.py
import boto3
account_id = 'xxxxxxxxxxxxxxxx'
region_name = '<リージョン>'
runtime_name = 'sample_agent_1'
agent_runtime_id = 'sample_agent_1-xxxxxxxxxx' # ここは初回デプロイの時は不明なはずなので、適当で良いです。
client = boto3.client('bedrock-agentcore-control', region_name=region_name)
try:
# エージェントランタイム取得
response = client.get_agent_runtime(
agentRuntimeId=agent_runtime_id
)
update_response = client.update_agent_runtime(
agentRuntimeId=agent_runtime_id,
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f'{account_id}.dkr.ecr.{region_name}.amazonaws.com/{runtime_name}:latest'
}
},
networkConfiguration={"networkMode": "PUBLIC"},
roleArn=f'arn:aws:iam::{account_id}:role/AgentRuntimeRoleTest'
)
print(f"Agent Runtime updated successfully!")
print(f"Agent Runtime ARN: {update_response['agentRuntimeArn']}")
print(f"Status: {update_response['status']}")
except Exception as e:
print('エージェントランタイムの取得または更新に失敗しました。新規作成を試みます。')
create_response = client.create_agent_runtime(
agentRuntimeName=runtime_name,
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f'{account_id}.dkr.ecr.{region_name}.amazonaws.com/{runtime_name}:latest'
}
},
networkConfiguration={"networkMode": "PUBLIC"},
roleArn=f'arn:aws:iam::{account_id}:role/AgentRuntimeRoleTest'
)
print(f"Agent Runtime created successfully!")
print(f"Agent Runtime ARN: {create_response['agentRuntimeArn']}")
print(f"Status: {create_response['status']}")
作成できたら以下を実行します。
$ uv run invoke_agent.py
6. 動作テスト
touch invoke_agent.py
import boto3
import json
client = boto3.client('bedrock-agentcore', region_name='ap-northeast-1')
payload = json.dumps({"prompt": "<〇〇ツールを使用して〇〇してください。>"})
response = client.invoke_agent_runtime(
agentRuntimeArn='<デプロイしたBedrock AgentCore RuntimeのARN>',
# Must be 33+ char. Every new SessionId will create a new MicroVM
runtimeSessionId='<任意の33文字以上の文字列>',
payload=payload,
qualifier="DEFAULT"
)
response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)
$ uv run invoke_agent.py
これでMCPサーバのツールを使用した結果が返ってきていればOK
まとめ
MCPサーバをAWS環境内から使用するなら、IAM認証にしておいた方が構築が楽です。
別のインフラサービスから使用する場合はIAM認証を使えないので、JWT認証を使用する必要があります。
細かい制御をしたくなったらAgentCore Policyも必要になりそうですね。
以上