本記事は
AWSアワード受賞者祭り
8日目の記事です。
✨🏆
7日目
▶▶ 本記事 ▶▶
9日目
🏆✨
はじめに
AWS Summit New York City 2025で、Bedrock AgentCoreというサービスが発表されました。
Amazon Bedrock AgentCoreは、AIエージェントをセキュアにデプロイ・運用するための統合サービス群です。AgentCoreを用いることで、これまでAIエージェントを構築する際の課題であった、インフラ構築、セキュリティ実装、運用管理の複雑さを解決することができます。今回は、AgentCoreの中でもRuntime, Built-in Tools, Gatewaysの3つのコンポーネントに焦点をあてて、実際に実装してみました。
1. AgentCore Runtime
AgentCore Runtimeは、AIエージェントとツールをデプロイ・スケールするための安全なサーバーレス実行環境です。
主な特徴
フレームワーク・モデル非依存：Strands Agents、LangChain、LangGraph、CrewAI等、任意のフレームワークに対応しており、Amazon Bedrock内外のモデルが利用可能
簡単デプロイ：Python SDKの
@app.entrypointデコレータでローカル関数をHTTPサービス化でき、最小限のコード変更でプロトタイプから本番環境へ移行可能統合機能：Memory、Gateway、Observability、Toolsと統一SDK経由で連携できる
ローカルで実行してみる
公式ドキュメントに記載されているstarter toolkitを使ったやり方で実装してみます。
まずは必要なモジュールをインストールします。
uv add bedrock-agentcore bedrock-agentcore-starter-toolkit
今回はエージェントフレームワークとしてStrands Agentsを使っていきます。 初期状態では、特にプロンプトの設定は行っていません。
from bedrock_agentcore.runtime import BedrockAgentCoreApp from strands import Agent agent = Agent() app = BedrockAgentCoreApp() @app.entrypoint def invoke(payload): """Process user input and return a response""" user_message = payload.get("prompt", "Hello") response = agent(user_message) return str(response) if __name__ == "__main__": app.run()
エージェントが定義できたら、下のコマンドで設定をしていきます。
uv run agentcore configure --entrypoint myagent.py -er <YOUR_IAM_ROLE_ARN>
コマンド実行後、Dockerfileとbedrock_agentcore.yamlという設定ファイルが作成されます。
続いて、以下のコマンドでエージェントを起動します。
-lオプションを付けることでコンテナをローカルで起動してテストすることができます。
uv run agentcore launch -l
localhost:8080で立ち上がりました。
> uv run agentcore launch -l Launching Bedrock AgentCore (local mode)... Launching Bedrock AgentCore agent 'myagent' locally Docker image built: bedrock_agentcore-myagent:latest ✓ Docker image built: bedrock_agentcore-myagent:latest ✓ Ready to run locally Starting server at http://localhost:8080
ではこのエージェントに対してリクエストを送ってみます。
uv run agentcore invoke -l '{"prompt": "こんにちは"}'
> uv run agentcore invoke -l '{"prompt": "こんにちは"}'
Payload:
{
"prompt": "こんにちは"
}
Invoking BedrockAgentCore agent 'myagent' locally
Found credentials in shared credentials file: ~/.aws/credentials
Getting workload access token...
Successfully retrieved workload access token
Session ID: 959e31fb-05d6-4027-9469-d979583575f6
Response:
{
"response": "こんにちは！お元気ですか？何かお手伝いできることがあれば、お気軽にお声かけください。
"
}
レスポンスが返ってくることが確認できました。
AWSへのデプロイ
ではローカル上で構築したこのエージェントをAWS環境にデプロイしてみます。
といっても先程のコマンドの-lオプションを外すだけでとても簡単にデプロイできます。
uv run agentcore launch
launchコマンドを実行すると以下のようなことを聞かれます。 今回はデフォルトのまま起動していきます。ECRも勝手に作ってくれるのはありがたいです。
🏗️ ECR Repository Press Enter to auto-create ECR repository, or provide ECR Repository URI to use existing ECR Repository URI (or skip to auto-create): ✓ Will auto-create ECR repository 🔍 Detected dependency file: requirements.txt Press Enter to use this file, or type a different path (use Tab for autocomplete): Path or Press Enter to use detected dependency file: ✓ Using detected file: requirements.txt 🔐 Authorization Configuration By default, Bedrock AgentCore uses IAM authorization. Configure OAuth authorizer instead? (yes/no) [no]: ✓ Using default IAM authorization Configuring BedrockAgentCore agent: myagent
デプロイが完了すると、コンソール上からもRuntimeができているのが確認できます。
同じように実行してみます
uv run agentcore invoke '{"prompt": "Hello"}'
Hello! How are you doing today? Is there anything I can help you with?
実行ログ
> uv run agentcore invoke '{"prompt": "Hello"}'
Payload:
{
"prompt": "Hello"
}
Invoking BedrockAgentCore agent 'myagent' via cloud endpoint
Found credentials in shared credentials file: ~/.aws/credentials
Session ID: 959e31fb-05d6-4027-9469-d979583575f6
Response:
{
"ResponseMetadata": {
"RequestId": "4dc9f97b-1917-4df6-b548-1b435919b3cc",
"HTTPStatusCode": 200,
"HTTPHeaders": {
"date": "Sun, 20 Jul 2025 01:08:44 GMT",
"content-type": "application/json",
"transfer-encoding": "chunked",
"connection": "keep-alive",
"x-amzn-requestid": "4dc9f97b-1917-4df6-b548-1b435919b3cc",
"baggage":
"Self=1-687c4192-10efad390223e82b01b972bc,session.id=959e31fb-05d6-4027-9469-d979583575f6",
"x-amzn-bedrock-agentcore-runtime-session-id":
"959e31fb-05d6-4027-9469-d979583575f6",
"x-amzn-trace-id":
"Root=1-687c4192-32a722622391d2d93e1a75a7;Self=1-687c4192-10efad390223e82b01b972bc"
},
"RetryAttempts": 0
},
"runtimeSessionId": "959e31fb-05d6-4027-9469-d979583575f6",
"traceId":
"Root=1-687c4192-32a722622391d2d93e1a75a7;Self=1-687c4192-10efad390223e82b01b972bc",
"baggage":
"Self=1-687c4192-10efad390223e82b01b972bc,session.id=959e31fb-05d6-4027-9469-d979583575f6",
"contentType": "application/json",
"statusCode": 200,
"response": [
"b'\"Hello! How are you doing today? Is there anything I can help you with?\\\\n\"'"
]
}
Lambda関数を作るかのように、とても簡単にデプロイ、起動まで実行できました。 従来のLambda + API GatewayやECSを使ったデプロイよりも圧倒的に楽ですね。
2. AgentCore Built-in Tools
AgentCore Toolsは、AIエージェントが複雑なタスクを安全かつ効率的に実行するための機能群です。 現在は以下の2つのToolが組み込みのものとして提供されています。今回はCode Interpreterを使用してみます。
- Amazon Bedrock AgentCore Code Interpreter
- Amazon Bedrock AgentCore Browser Tool
主な特徴
安全なコード実行：分離されたサンドボックス環境で内部データソースにアクセスしながらセキュアに実行可能
AWS完全管理ソリューション：Strands Agents、LangGraph、CrewAI等のフレームワークとシームレス統合可能
高度な設定サポート：大容量ファイル対応（入出力）、インターネットアクセスに対応
多言語対応：JavaScript、TypeScript、Python等の事前構築ランタイムモードに対応
ツール単体で実行してみる
まずはこのCode Interpreterを単体で実行してみます。
コード
import json from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter # Configure and Start the code interpreter session code_client = CodeInterpreter("us-east-1") code_client.start() # Execute the hello world code response = code_client.invoke( "executeCode", {"language": "python", "code": 'print("Hello World!!!")'} ) # Process and print the response for event in response["stream"]: print(json.dumps(event["result"], indent=2)) # Clean up and stop the code interpreter sandbox session code_client.stop()
実行結果
> uv run python test/run_code.py
{
"content": [
{
"type": "text",
"text": "Hello World!!!"
}
],
"structuredContent": {
"stdout": "Hello World!!!",
"stderr": "",
"exitCode": 0,
"executionTime": 0.034195899963378906
},
"isError": false
}
PythonコードでHello, World!!!と出力させて、その結果を表示することができました。
ライブラリの確認
インストール済みのライブラリを確認してみました。 Numpy, Pandas, Matplotlib, BeautifulSoupなど有名どころのライブラリは一通りインストールされているようです。
コードと実行結果
import json from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter code_client = CodeInterpreter("us-east-1") code_client.start() library_check_code = """ import importlib.metadata as metadata import sys print("Python version:", sys.version) print("Platform:", sys.platform) print() print("Installed packages:") distributions = metadata.distributions() packages = [(dist.metadata['Name'], dist.version) for dist in distributions] packages.sort(key=lambda x: x[0].lower()) for name, version in packages: print(f"{name}: {version}") print(f"\\nTotal packages: {len(packages)}") """ # Execute the library check code response = code_client.invoke( "executeCode", {"language": "python", "code": library_check_code} ) # Process and print the response for event in response["stream"]: print(json.dumps(event["result"], indent=2)) # Clean up and stop the code interpreter sandbox session code_client.stop()
{
"content": [
{
"type": "text",
"text": "Python version: 3.10.16 (main, Jul 10 2025, 21:12:52) [GCC 11.3.1 20221121 (Red Hat 11.3.1-4)]\nPlatform: linux\n\nInstalled packages:\n\nabsl-py: 2.1.0\naiohappyeyeballs: 2.3.5\naiohttp: 3.11.11\naiosignal: 1.3.1\naiosignal: 1.2.0\nalabaster: 0.7.10\nannotated-types: 0.7.0\nannotated-types: 0.7.0\nanyio: 4.9.0\nanyio: 4.9.0\nasgiref: 3.8.1\nastor: 0.8.1\nasttokens: 2.4.1\nasttokens: 2.4.1\nasync-timeout: 4.0.3\nattrs: 23.2.0\nattrs: 24.2.0\nbabel: 2.17.0\nbackcall: 0.1.0\nbackoff: 2.2.1\nbcrypt: 4.1.3\nbeautifulsoup4: 4.12.3\nblinker: 1.8.2\nblis: 0.7.11\nbokeh: 2.4.3\nbranca: 0.7.2\nBrazilPython-pytest: 3.x\nBrazilPython-Pytest-Guard: 2.x\nBrazilPythonTestSupport: 3.0\ncachetools: 5.3.3\ncatalogue: 2.0.10\ncertifi: 2024.6.2\ncertifi: 2024.7.4\ncffi: 1.16.0\nchardet: 5.2.0\ncharset-normalizer: 3.3.2\ncharset-normalizer: 3.4.1\nclick: 8.1.7\nclick: 8.1.7\nclick-plugins: 1.1.1\ncloudpathlib: 0.16.0\ncloudpickle: 3.0.0\ncolorama: 0.4.6\ncomm: 0.2.2\nconfection: 0.1.5\ncontourpy: 1.2.1\ncoverage: 7.2.7\ncryptography: 40.0.2\ncssselect2: 0.7.0\ncycler: 0.12.1\ncymem: 2.0.8\ndebugpy: 1.8.1\ndecorator: 4.4.2\ndecorator: 4.4.0\ndistro: 1.9.0\nDjango: 5.0.6\ndnspython: 2.6.1\ndocutils: 0.18.1\ndocx2txt: 0.8\nduckdb: 0.8.1\nemail_validator: 2.1.1\nentrypoints: 0.4\net-xmlfile: 1.1.0\nexceptiongroup: 1.3.0\nexecnet: 2.1.1\nexecuting: 2.0.1\nexecuting: 0.8.2\nFaker: 19.13.0\nfastapi: 0.111.0\nfastapi: 0.115.12\nffmpeg-python: 0.2.0\nffmpy: 0.3.2\nfilelock: 3.14.0\nFlask: 3.0.3\nFlask-Cors: 4.0.1\nFlask-Login: 0.6.3\nfonttools: 4.53.0\nfpdf: 1.7.2\nfrozenlist: 1.4.1\nfrozenlist: 1.4.0\nfsspec: 2024.5.0\nfuture: 1.0.0\ngenesis-1p-tools-rpm-bundle: 1.0.0\ngreenlet: 3.0.3\ngTTS: 2.5.1\ngunicorn: 22.0.0\nh11: 0.14.0\nh11: 0.16.0\nh2: 4.1.0\nhpack: 4.0.0\nhttpcore: 1.0.7\nhttptools: 0.6.1\nhttpx: 0.28.1\nhttpx-sse: 0.4.0\nHypercorn: 0.17.3\nhyperframe: 6.0.1\nidna: 3.7\nidna: 3.10\nimageio: 2.34.1\nimageio-ffmpeg: 0.5.1\nimagesize: 1.4.1\nimgkit: 1.2.3\nimportlib-metadata: 7.0.1\nimportlib_metadata: 7.1.0\nimportlib_resources: 6.4.0\niniconfig: 0.0.0\nipykernel: 6.29.5\nipython: 8.25.0\nipython: 8.12.0\nitsdangerous: 2.2.0\nitsdangerous: 2.2.0\njedi: 0.19.1\njedi: 0.19.1\nJinja2: 3.1.4\nJinja2: 3.1.5\njoblib: 1.4.2\njupyter_client: 8.6.0\njupyter_core: 5.7.1\nkiwisolver: 1.4.5\nlangcodes: 3.4.0\nlanguage_data: 1.2.0\nlazy_loader: 0.4\nloguru: 0.7.2\nlxml: 5.2.2\nmarisa-trie: 1.1.1\nmarkdown-it-py: 3.0.0\nmarkdown2: 2.4.13\nmarkdownify: 0.12.1\nMarkupSafe: 2.1.5\nMarkupSafe: 2.1.5\nmatplotlib: 3.9.0\nmatplotlib-inline: 0.1.7\nmatplotlib-inline: 0.1.3\nmatplotlib-venn: 0.11.10\nmcp: 1.9.2\nmdurl: 0.1.2\nmizani: 0.9.3\nmoviepy: 1.0.3\nmpmath: 1.3.0\nmultidict: 6.0.5\nmultidict: 6.0.4\nmultipart: 1.2.1\nmurmurhash: 1.0.10\nmutagen: 1.47.0\nnest-asyncio: 1.5.6\nnetworkx: 3.3\nnltk: 3.8.1\nnumexpr: 2.10.0\nnumpy: 1.26.4\nopenai: 1.33.0\nopencv-python: 4.10.0.82\nopenpyxl: 3.1.3\norjson: 3.10.3\npackaging: 24.0\npackaging: 24.2\npandas: 1.5.3\nparso: 0.8.4\nparso: 0.8.3\npatsy: 0.5.6\npdf2image: 1.17.0\npdfkit: 1.0.0\npdfminer.six: 20231228\npdfplumber: 0.11.0\npdfrw: 0.4\npexpect: 4.9.0\npexpect: 4.7.0\npickleshare: 0.7.5\npillow: 10.3.0\npip: 25.1.1\npip: 23.0.1\nplatformdirs: 4.2.2\nplatformdirs: 4.2.0\nplotly: 5.22.0\npluggy: 0.13.0\npreshed: 3.0.9\npriority: 2.0.0\nproglog: 0.1.10\nprompt_toolkit: 3.0.45\nprompt_toolkit: 3.0.51\npropcache: 0.2.1\npsutil: 5.9.5\npsycopg2-binary: 2.9.9\nptyprocess: 0.7.0\nptyprocess: 0.5.1\npure-eval: 0.2.2\npure-eval: 0.2.1\npy: 1.11.0\npycparser: 2.22\npydantic: 2.10.5\npydantic: 2.10.5\npydantic-settings: 2.7.1\npydantic_core: 2.27.2\npydantic_core: 2.27.2\npydub: 0.25.1\nPygments: 2.18.0\nPygments: 2.16.1\npymongo: 4.7.2\nPyNaCl: 1.5.0\npyOpenSSL: 23.2.0\npyparsing: 3.1.2\nPyPDF2: 3.0.1\npypdfium2: 4.30.0\npypng: 0.20220715.0\npyshp: 2.3.1\npytest: 6.2.5\npytest-asyncio: 0.20.3\npytest-cov: 4.0.0\npytest-html: 1.17.0\npytest-metadata: 1.7.0\npytest-xdist: 2.5.0\nPython-amazon-doc-utils: 1.0\npython-dateutil: 2.9.0.post0\npython-dateutil: 2.9.0\npython-docx: 1.1.2\npython-dotenv: 1.0.1\npython-dotenv: 0.21.1\npython-multipart: 0.0.9\npytz: 2024.1\npytz: 2023.3.post1\nPyWavelets: 1.6.0\npyxlsb: 1.0.10\nPyYAML: 6.0.1\npyzbar: 0.1.9\npyzmq: 25.1.2\npyzmq: 25.1.1\nqrcode: 7.4.2\nrarfile: 4.2\nredis: 5.0.4\nregex: 2024.5.15\nreportlab: 4.2.0\nrequests: 2.32.4\nretrying: 1.3.4\nrich: 13.7.1\nscikit-image: 0.23.2\nscikit-learn: 1.5.0\nscipy: 1.13.1\nsetuptools: 65.5.0\nsetuptools: 57.4.0\nshapely: 2.0.4\nshellingham: 1.5.4\nsix: 1.16.0\nsix: 1.16.0\nsmart-open: 6.4.0\nsniffio: 1.3.1\nsniffio: 1.3.0\nsnowballstemmer: 2.2.0\nsoupsieve: 2.5\nspacy: 3.7.4\nspacy-legacy: 3.0.12\nspacy-loggers: 1.0.5\nSphinx: 7.1.2\nsphinxcontrib-applehelp: 1.0.1\nsphinxcontrib-devhelp: 1.0.1\nsphinxcontrib-htmlhelp: 2.0.0\nsphinxcontrib-jsmath: 1.0.1\nsphinxcontrib-qthelp: 1.0.2\nsphinxcontrib-serializinghtml: 1.1.5\nSQLAlchemy: 1.4.52\nsqlparse: 0.5.0\nsrsly: 2.4.8\nsse-starlette: 2.3.4\nstack-data: 0.6.3\nstack-data: 0.1.4\nstarlette: 0.47.1\nstarlette: 0.46.2\nstatsmodels: 0.14.2\nsvglib: 1.5.1\nsvgwrite: 1.4.3\nsympy: 1.12.1\ntenacity: 8.3.0\ntextblob: 0.18.0.post0\nthinc: 8.2.3\nthreadpoolctl: 3.5.0\ntifffile: 2024.5.22\ntinycss2: 1.3.0\ntoml: 0.10.2\ntomli: 2.1.0\ntorch: 2.3.0\ntorchaudio: 2.3.0\ntorchvision: 0.18.0\ntornado: 6.4\ntornado: 6.4.2\ntqdm: 4.66.4\ntraitlets: 5.14.3\ntraitlets: 5.14.3\ntyper: 0.9.4\ntyping_extensions: 4.12.1\ntyping_extensions: 4.13.2\ntzdata: 2024.1\nujson: 5.10.0\nurllib3: 1.26.19\nuvicorn: 0.30.1\nuvicorn: 0.34.0\nuvloop: 0.19.0\nWand: 0.6.13\nwasabi: 1.1.3\nwatchfiles: 0.22.0\nwcwidth: 0.2.13\nwcwidth: 0.2.13\nweasel: 0.3.4\nwebencodings: 0.5.1\nWerkzeug: 3.0.3\nwsproto: 1.2.0\nxgboost: 2.0.3\nxlrd: 2.0.1\nXlsxWriter: 3.2.0\nxml-python: 0.4.3\nxyzservices: 2024.4.0\nyarl: 1.9.4\nyarl: 1.18.3\nzipp: 3.21.0\n\nTotal packages: 311"
AgentCore Runtimeから使ってみる
では先程のAgentCore RuntimeとCode Interpreterを組み合わせて、コード実行機能付きエージェントを作成してみます。
コード
import asyncio from bedrock_agentcore.runtime import BedrockAgentCoreApp from bedrock_agentcore.tools.code_interpreter_client import code_session from strands import Agent, tool # システムプロンプトの定義 SYSTEM_PROMPT = """コードを実行して回答を検証してください。 利用可能なツール： - execute_python: Pythonコードを実行""" # コードインタープリターツールの定義 @tool def execute_python(code: str, description: str = "") -> str: """Execute Python code in the sandbox.""" if description: code = f"# {description}\n{code}" print(f"\n実行コード: {code}") # コードインタープリターセッション内でコードを実行 try: with code_session("us-east-1") as code_client: response = code_client.invoke( "executeCode", {"code": code, "language": "python", "clearContext": False}, ) for event in response["stream"]: result = event["result"] # エラーチェック if result.get("isError", False): error_content = result.get("content", []) if error_content: return ( f"エラー: {error_content[0].get('text', 'Unknown error')}" ) # 正常な実行結果を返す content = result.get("content", []) if content and content[0].get("type") == "text": return content[0].get("text", "実行完了（出力なし）") return "実行完了" except Exception as e: return f"実行エラー: {str(e)}" # ツール付きエージェントの設定 agent = Agent(tools=[execute_python], system_prompt=SYSTEM_PROMPT) # BedrockAgentCoreアプリケーションの設定 app = BedrockAgentCoreApp() @app.entrypoint def invoke(payload): """Process user input and return a response with code execution capability""" user_message = payload.get("prompt", "Hello") async def run_agent(): response_text = "" async for event in agent.stream_async(user_message): if "data" in event: chunk = event["data"] response_text += chunk return response_text # asyncio.run で非同期処理を実行 try: response = asyncio.run(run_agent()) return response except Exception as e: print(f"エージェント実行エラー: {str(e)}") return f"処理中にエラーが発生しました: {str(e)}" if __name__ == "__main__": app.run()
出力結果
こちらも、AgentCoreからコードを実行して、その結果を得ることができました。
uv run agentcore invoke -l '{"prompt": "これらの値の平均値をコードを実行して求めて 1231, 3423, 2342, 7732, 2342"}'
Response:
{
"response": "これらの値の平均値をPythonコードで計算します。
# 値のリスト
values = [1231, 3423, 2342, 7732, 2342]
# 平均値を計算
average = sum(values) / len(values)
print(f"値: {values}")
print(f"合計: {sum(values)}")
print(f"個数: {len(values)}")
print(f"平均値: {average}")
実行結果：
値: [1231, 3423, 2342, 7732, 2342]
合計: 17070
個数: 5
平均値: 3414.0
**答え: 3414.0**
これらの値の平均値は **3414** です。
"
}
3. AgentCore Gateway
最後はAgentCore Gatewayです。
AgentCore Gatewayは、開発者が既存リソースをMCPのようなAIエージェント対応ツールに変換し、スケーラブルに接続するためのサービスです。
主な特徴
簡単ツール変換: API、Lambda関数を数行のコードでModel Context Protocol（MCP）対応ツールに変換
1クリック統合：Salesforce、Slack、Jira、Asana、Zendesk等の人気ツールとの即座連携
統一アクセス：複数ツールソースを単一の安全なエンドポイントに統合
一括認証：インバウンド認証（エージェント身元確認）とアウトバウンド認証（ツール接続）を一元管理
LambdaからGatewayを作成する
Gatewayは、Lambda関数、既存のAPI、Integrationが選択できます。Integrationでは、様々なSaasとすぐに連携することができるようです。
今回はLambda関数からGatewayを作成してみたいと思います。
Lambda関数の作成
まずはもととなるLambda関数を作成していきます。 今回は擬似的に天気を取得できる関数を作成します。
コード
import json def get_weather(location): return f"The weather in {location} is Sunny!" def lambda_handler(event, context): print(event) toolName = context.client_context.custom['bedrockAgentCoreToolName'] delimiter = "___" if delimiter in toolName: toolName = toolName[toolName.index(delimiter) + len(delimiter):] print(toolName) if toolName == 'get_weather': return {'statusCode': 200, 'body': get_weather(event['location'])}
作成したLambdaのARNを後で使うので控えておきます。
Gatewayの作成
コンソール上から作成していきます。
Inbound Auth configurationは、「Quick create configurations with Cognito - recommended」を選択します。
Permissionは「Use an IAM service role」を選択し、適切な権限を持ったIAMロールを設定します。
最後にTargetの設定です。ここでGatewayとToolの紐づけを行っていきます。
nameとdescriptionは任意で設定して、今回はTarget Typeを「Lambda ARN」,Lambda ARNを先程作成したLambda関数のARNにします。
最後にスキーマの設定です。S3のファイルを参照する方法とインラインで書く方法がありますが、今回はインラインでそのまま書いていきます。今回は以下のようにスキーマを設定しました。
{ "description": "tool to get weather information for a specified location", "inputSchema": { "properties": { "location": { "type": "string", "description": "The location to get weather information for" } }, "required": [ "location" ], "type": "object" }, "name": "get_weather" }
Strands Agentsから使ってみる
では作成したGatewayをローカルのStrands Agentsから使ってみます。GatewayはMCP互換の仕様で作成されるので、そのままMCPのToolと同じように使用できます。ただgatewayの作成時に設定したように、Cognitoを使ったBearerトークンによる認証をアクセス時に行う必要があります。
Cognitoのコンソールページを確認すると、新しいユーザープールが作成されていることが確認できます
後ほどgatewayアクセス時に使用するため、このユーザープールのアプリケーションクライアントから、クライアントID, クライアントシークレット、カスタムスコープ、Discovery URLを控えておきます。
コード
Strands Agentsを用いた実装例です
import base64 import os import requests from mcp.client.streamable_http import streamablehttp_client from strands import Agent from strands.tools.mcp import MCPClient def get_access_token(): # 環境変数から値を取得 client_id = os.environ.get("CLIENT_ID") client_secret = os.environ.get("CLIENT_SECRET") discovery_url = os.environ.get("DISCOVERY_URL") custom_scope = os.environ.get("CUSTOM_SCOPE") if not all([client_id, client_secret, discovery_url, custom_scope]): raise ValueError( "必要な環境変数が設定されていません: CLIENT_ID, CLIENT_SECRET, DISCOVERY_URL, CUSTOM_SCOPE" ) # Basic認証用のヘッダーを作成（base64エンコード） credentials = f"{client_id}:{client_secret}" auth_header = base64.b64encode(credentials.encode()).decode() # ディスカバリーURLからトークンエンドポイントを取得 discovery_response = requests.get(discovery_url) discovery_response.raise_for_status() discovery_data = discovery_response.json() token_endpoint = discovery_data["token_endpoint"] # アクセストークンを取得 token_headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {auth_header}", } token_data = {"grant_type": "client_credentials", "scope": custom_scope} token_response = requests.post( token_endpoint, headers=token_headers, data=token_data ) token_response.raise_for_status() token_json = token_response.json() access_token = token_json["access_token"] return access_token access_token = get_access_token() streamable_http_mcp_client = MCPClient( lambda: streamablehttp_client( AGENTCORE_GATEWAY_ENDPOINT, headers={"Authorization": f"Bearer {access_token}"}, ) ) with streamable_http_mcp_client: tools = streamable_http_mcp_client.list_tools_sync() agent = Agent(tools=tools) agent("東京の天気を教えて")
環境変数には、以下の値をセットしておきましょう。
export CLIENT_ID=XXXX export CLIENT_SECRET=XXXX export DISCOVERY_URL=XXXX export CUSTOM_SCOPE=XXXX
今回は検証のためそのままコンソールから環境変数を設定していますが、セキュリティ面を考慮するとdotenvで.envファイルから値を読み込むようにしたり、Secret Mangerでこれらの値を管理することをお勧めします。
処理の流れとしては、まず環境変数から先程控えたCognitoの各種認証情報を取得し、それらをもとにアクセストークンを取得後、そのアクセストークンをもとにGatewayにBearer認証でアクセスするという流れになります。
実行結果
> uv run python agent.py 東京の天気情報を取得いたします。 Tool #1: get-weather-target___get_weather 東京の天気は**晴れ**です！☀️ 良いお天気ですね。外出には最適な天候のようです。
Gatewayで定義したget_weatherツールを使用して回答が生成されていることが確認できました。
おわりに
今回は、AgentCoreの中でも、Runtime, Tools, Gatewayを実際に使ってみました。まだ軽くしか触ってないですが、AgentやMCPの開発、デプロイがとても簡単にでき、すごく可能性を感じるサービスだと思いました。特にAgentCore RuntimeとStrands Agentsを組み合わせることで、最小限のコードとコマンドで、すぐフルマネージドのエージェント実行環境ができるのはとても便利だと感じました。まだMemoryやIdentityなど触れてない機能がいっぱいあるので、これから色々と試してみたいと思います。
