はじめに
祝・Bedrock AgentCore東京リージョンへ登場！
その影に隠れてひっそりと、AgentCore Gatewayのターゲットタイプに MCP server が追加されました！
同じタイミングで追加されたものじゃなかったらすみません)
ドキュメントも更新されていました！
ターゲットタイプ: MCP serverとは
Gateway のターゲットとしてMCPサーバーを直接登録できるようになりました。
ここで言うMCPサーバーとは、プロバイダーなどが公開しているリモートMCPサーバーを含んでいます。
これまで、外部のMCPサーバーに対してAgentCore Runtimeからアクセスしたい場合、コードベースでAIエージェントに1つ1つ組み込んであげる必要がありました。
なので使用するMCPサーバーに変更があった場合は毎回コードを修正する必要がありました。
これからは、そのMCPサーバーの設定をGateway側でしておくことで、Runtimeへデプロイするコードの中ではGatewayへの接続さえ記述しておけば良くなります。
これによってMCPサーバーの追加/削除はGatewayのターゲットを増やす/減らすだけで良くなるため、開発プロセスを効率化することができるようになりました。
差別化されない重労働を無くすという観点で、とてもAgentCoreらしいアップデートな感じですね！
注意点
接続先はURLで到達できるMCPサーバーです。ドキュメントでは「外部 MCP サーバー（external）」に接続すると明記されています。
ローカル(localhost)はそのままでは不可で、公開 URL を用意するか、AWS内に配置してGateway から到達できるエンドポイントにする必要があります。
現状だとリモートMCPサーバーを設定するのが手っ取り早くて楽そうですね〜
認証
Outbound 認証に OAuth2(2-legged) か NoAuth を選択できます。
NoAuthは基本的に非推奨のようですが、そもそもNoAuthで使えるMCPサーバー(AWS Knowledge MCPサーバーなど)も用意されているため、それに対応しているのだと思います。
(OAuth2はAgentCore Identityに設定した OAuth プロバイダを使います。)
AWS Knowledge MCPサーバーをGatewayに登録してみる
まずは以下の感じでGatewayを設定します。
その後、IdentityでCredential Providerを設定します。
以下ブログを参考にしてみてください。
次にコードを書きます。特に個別のMCPサーバーを設定していないのがわかると思います。
knowledge_mcp_agent.py
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from typing import Dict, Any
from strands import Agent
from strands.tools.mcp import MCPClient
import logging
from boto3.session import Session
import os
import json
# MCPクライアント用のインポート
from mcp.client.streamable_http import streamablehttp_client
# AgentCore Identityからアクセストークンを取得する
from bedrock_agentcore.identity.auth import requires_access_token
app = BedrockAgentCoreApp()
logger = logging.getLogger("knowledge_mcp_agent")
logger.setLevel(logging.INFO)
logging.basicConfig(
format="%(levelname)s | %(name)s | %(message)s",
handlers=[logging.StreamHandler()]
)
_boto_session = Session()
region = _boto_session.region_name
AGENT_SYSTEM_PROMPT = """
あなたはAWSマスターです。AWS Knowledge MCPを使って最新情報を検索し、ユーザーからの質問に回答してください。
MCPで検索して得た情報のみを用いて回答してください。検索結果になかった場合はその旨を回答してください。
"""
def parse_prompt_from_payload(payload: Dict[str, Any]) -> str:
"""AgentCore Runtime互換のペイロードからプロンプトを抽出する。"""
if not payload:
return ""
# 入れ子構造（input フィールド）に対応
if "input" in payload:
input_data = payload["input"]
if isinstance(input_data, dict):
return input_data.get("prompt", "")
if isinstance(input_data, str):
try:
return json.loads(input_data).get("prompt", "")
except Exception:
return input_data
# 直接 prompt があるケース
if "prompt" in payload:
return str(payload["prompt"]) # 念のため文字列化
return ""
def _get_tool_name(tool: Any) -> str:
"""ツール名を抽出する。"""
return getattr(tool, "tool_name", getattr(tool, "name", str(tool)))
class GatewayMcpConfig:
"""
AgentCore Gatewayに接続するための設定。AgentCore Identityを使用。s
必要な環境変数：
- GATEWAY_URL: Gatewayのエンドポイント
- COGNITO_SCOPE: Cognito OAuth2のスコープ
- WORKLOAD_NAME: （オプション）workload名、デフォルトは"GatewayMcp"
- USER_ID: (オプション)user-idを設定する、デフォルトは"GatewayMcp"
"""
def __init__(self):
# 環境変数を取得して検証
gateway_url = os.environ.get("GATEWAY_URL")
if not gateway_url:
raise ValueError("GATEWAY_URL環境変数が必要です")
self.gateway_url: str = gateway_url
provider_name = os.environ.get("PROVIDER_NAME")
if not provider_name:
raise ValueError("PROVIDER_NAME環境変数が必要です")
self.provider_name: str = provider_name
cognito_scope = os.environ.get("COGNITO_SCOPE")
if not cognito_scope:
raise ValueError("COGNITO_SCOPE環境変数が必要です")
self.cognito_scope: str = cognito_scope
self.workload_name = os.environ.get("WORKLOAD_NAME", "GatewayMcp")
self.user_id = os.environ.get("USER_ID", "GatewayMcp")
self.region = region
logger.info(f"Gateway URL: {self.gateway_url}")
logger.info(f"Cognito scope: {self.cognito_scope}")
logger.info(f"Workload name: {self.workload_name}")
logger.info(f"User ID: {self.user_id}")
logger.info(f"AWS Region: {self.region}")
async def get_access_token(self) -> str:
"""AgentCore Identityを使用してアクセストークンを取得する。
Runtime環境では、runtimeUserIdはInvokeAgentRuntime API呼び出し時に
システム側が設定し、Runtimeがエージェントに渡します。
Returns:
str: 認証されたAPIコール用のアクセストークン
"""
# @requires_access_tokenデコレータ付きのラッパー関数を作成
# Runtime環境では、デコレータが内部で_get_workload_access_tokenを呼び出し、
# workload access tokenを自動的に取得する
@requires_access_token(
provider_name=self.provider_name,
scopes=[self.cognito_scope],
auth_flow="M2M",
force_authentication=False,
)
async def _get_token(*, access_token: str) -> str:
"""
AgentCore Identityからアクセストークンを受け取る内部関数。
デコレータが内部で以下を処理：
1. _get_workload_access_tokenを呼び出してworkload access tokenを取得
- workload_name: Runtime環境から取得
- user_id: InvokeAgentRuntimeのruntimeUserIdヘッダーから取得
2. workload access tokenを使用してOAuth tokenを取得
3. access_tokenパラメータとして注入
Args:
access_token: OAuthアクセストークン（デコレータによって注入）
Returns:
str: APIコールで使用するアクセストークン
"""
logger.info("✅ AgentCore Identity経由でアクセストークンの取得に成功")
logger.info(f" Workload name: {self.workload_name}")
logger.info(f" トークンプレフィックス: {access_token[:20]}...")
logger.info(f" トークン長: {len(access_token)} 文字")
return access_token
# デコレータ付き関数を呼び出してトークンを取得
return await _get_token(access_token="")
async def create_mcp_client_and_tools(self) -> MCPClient:
"""
トークン取得 → MCPクライアントを返す。
MCPクライアントはwithコンテキスト内で使用する必要があるため、
認証済みのクライアントインスタンスを返します。
Returns:
MCPClient: 認証済みMCPクライアントインスタンス
"""
# AgentCore Identityを使用してアクセストークンを取得
logger.info("ステップ1: AgentCore Identity経由でアクセストークンを取得中...")
logger.info(f"Runtimeが自動的にruntimeUserIdを渡します")
access_token = await self.get_access_token()
# 認証されたMCPクライアントを作成
logger.info("ステップ2: 認証されたMCPクライアントを作成中...")
def create_streamable_http_transport():
"""
Bearerトークン認証を使用したストリーミング可能なHTTPトランスポートを作成。
このトランスポートは、MCPクライアントがGatewayへの認証された
リクエストを行うために使用されます。
"""
logger.info(f"🔗 MCP transport作成中: {self.gateway_url}")
logger.info(f"🔑 トークンプレフィックス: {access_token[:20]}...")
transport = streamablehttp_client(
self.gateway_url,
headers={"Authorization": f"Bearer {access_token}"}
)
logger.info("✅ MCP transport作成完了")
return transport
# 認証されたトランスポートでMCPクライアントを作成
mcp_client = MCPClient(create_streamable_http_transport)
return mcp_client
def get_full_tools_list(self, client: MCPClient) -> list:
"""
ページネーションをサポートしてすべての利用可能なツールをリスト。
Gatewayはページネーションされたレスポンスでツールを返す可能性があるため、
完全なリストを取得するためにページネーションを処理する必要があります。
Args:
client: MCPクライアントインスタンス
Returns:
list: 利用可能なツールの完全なリスト
"""
tools_list = client.list_tools_sync()
return list(tools_list)
# Strandsエージェントの初期化
class AgentFactory(GatewayMcpConfig):
"""
Knowledge向けAgentのビルダー。
- MCPセッションの 'with mcp_client:' は呼び出し側で保持する（重要）
- build(...) は *必ず with の中* で呼ぶこと（ツール列挙もその場のセッションで実施）
"""
def __init__(self, model_id: str | None = None, system_prompt: str | None = None):
super().__init__()
self.model_id = model_id
self.system_prompt = system_prompt
def build(self, mcp_client: MCPClient) -> Agent:
"""
with mcp_client: の内側で呼び出すこと。
MCPツールを列挙し、Knowledge系のみを選り分けて Agent を生成して返す。
"""
tools = self.get_full_tools_list(mcp_client)
# エージェントを生成
agent = Agent(
name="KnowledgeMcpAgent",
tools=tools,
model=self.model_id,
system_prompt=self.system_prompt,
)
# ツール情報のログ出力（オプション）
try:
tool_names = [_get_tool_name(t) for t in tools]
except Exception:
tool_names = [str(t) for t in tools]
logger.info(f"KnowledgeAgent 構築: ツール数={len(tools)} -> {tool_names}")
return agent
@app.entrypoint
async def invoke_agent(payload: Dict[str, Any]):
try:
gateway = GatewayMcpConfig()
except Exception as e:
logger.error(f"初期化エラー: {e}")
yield {"error": str(e)}
user_message = parse_prompt_from_payload(payload)
if not user_message:
logger.error(f"無効なペイロード構造: {payload}")
yield {"error": "無効なペイロード: 'prompt'フィールドが必要です"}
return
try:
mcp_config = await gateway.create_mcp_client_and_tools()
with mcp_config:
logger.info("✅ MCPコンテキストに入りました - セッションアクティブ")
aws_agent = AgentFactory(
model_id=os.environ.get("MODEL_ID", "jp.anthropic.claude-sonnet-4-5-20250929-v1:0"),
system_prompt=AGENT_SYSTEM_PROMPT,
).build(mcp_config)
stream = aws_agent.stream_async(user_message)
async for event in stream:
print(event)
yield (event)
except Exception as e:
logger.error(f"エージェント処理エラー: {str(e)}")
yield {"error": f"エージェント処理が失敗しました: {str(e)}"}
if __name__ == "__main__":
app.run()
これでAgentCore Runtimeへデプロイします。
$ agentcore configure --entrypoint knowledge_mcp_agent.py
$ agentcore launch \
--env COGNITO_SCOPE=<put-your-value> \
--env GATEWAY_URL=<put-your-value> \
--env PROVIDER_NAME=<put-your-value>
$ agentcore invoke '{
"input": { "prompt": "AgentCore Gatewayのターゲットタイプって何があるか全て教えて" }
}' --user-id "user01"
# 解答例
{'result': AgentResult(stop_reason='end_turn', message={'role': 'assistant', 'content': [{'text': '完璧です！AgentCore
Gatewayのターゲットタイプについて、すべての情報が揃いました。\n\n---\n\n**Amazon Bedrock AgentCore
Gatewayのターゲットタイプ**は以下の**4種類**があります：\n\n## 1. **Lambda functions（Lambda関数）**\n- AWS Lambda関数をツールとして接続できます\n-
カスタムビジネスロジックを好きなプログラミング言語で実装可能\n- Gatewayが自動的にLambda関数を呼び出し、レスポンスをMCP形式に変換します\n\n## 2. **OpenAPI
specifications（OpenAPI仕様）**\n- 既存のREST APIをMCP対応ツールに変換できます\n-
OpenAPI仕様を提供するだけで、GatewayがMCPとREST形式間の変換を自動的に処理します\n- このターゲットタイプには、AgentCore Credential Providerの設定が必要です（API
KeyまたはOAuth認証情報を保存）\n\n## 3. **Smithy models（Smithyモデル）**\n- SmithyモデルでAPIインターフェースを定義し、MCP対応ツールを生成できます\n-
Smithyは、AWSサービスや任意のAPIとやり取りするツールの生成に使用できます\n- GatewayはSmithyモデルを使用してツールを生成します\n\n## 4. **MCP
servers（MCPサーバー）**\n- リモートMCPサーバーをエージェントランタイムに接続できます\n- MCPツール機能のみがサポートされています\n-
コントロールプレーンとデータプレーン両方の操作で、ツールが利用できない場合は操作が失敗します\n\n---\n\n各ターゲットタイプには独自のCredential
Providerを設定でき、ターゲットへのアクセスを安全に制御できます。これらすべてのターゲットを組み合わせることで、Gateway は単一のMCP
URLとして、エージェントに必要なすべてのツールへのアクセスを提供します。\n\n※ Amazon Bedrock
AgentCoreはプレビューリリース版のため、変更される可能性があります。'}]}, metrics=EventLoopMetrics(cycle_count=4,
tool_metrics={'aws-knowledge-mcp___aws___search_documentation': ToolMetrics(tool={'toolUseId': 'tooluse_7vEycPj0Q2OyNSVxgYjjDg', 'name':
'aws-knowledge-mcp___aws___search_documentation', 'input': {'search_phrase': 'AgentCore Gateway target type', 'limit': 10}}, call_count=1, success_count=1,
error_count=0, total_time=3.5158333778381348), 'aws-knowledge-mcp___aws___read_documentation': ToolMetrics(tool={'toolUseId': 'tooluse_GECrQtZdTjatrPDrMTn3Yw',
'name': 'aws-knowledge-mcp___aws___read_documentation', 'input': {'url':
'https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-building-adding-targets.html'}}, call_count=2, success_count=2, error_count=0,
total_time=4.918605089187622)}, cycle_durations=[38.161696910858154], traces=[<strands.telemetry.metrics.Trace object at 0xffff69e58f50>,
<strands.telemetry.metrics.Trace object at 0xffff6ae86a30>, <strands.telemetry.metrics.Trace object at 0xffff7c784c30>, <strands.telemetry.metrics.Trace object
at 0xffff69e58cd0>], accumulated_usage={'inputTokens': 24960, 'outputTokens': 1013, 'totalTokens': 25973}, accumulated_metrics={'latencyMs': 19763}), state={})}
ということで、きちんとAWS Knowledge MCPサーバーを設定・使用できていることがわかりますね！
まとめ
リモートMCPサーバーにGateway経由で接続できるようになりました！
なお、CognitoのM2M認証にはアプリケーションクライアントごとに最低6ドル/月の課金が発生するため、どこまで実用的かは不明です👼
参考: