「MCP 伺服器」已新增至 AgentCore 網關目標類型！

引言

祝賀・Bedrock AgentCore在東京區域上線！

在這個重大時刻的背後，MCP伺服器已悄然被添加為AgentCore Gateway的目標類型！
如果不是同時添加的，還請見諒。

文件也已更新！

目標類型：MCP伺服器是什麼

現在可以將MCP伺服器直接註冊為Gateway的目標。
這裡所指的MCP伺服器包括供應商公開的遠程MCP伺服器等。

過去，若要從AgentCore Runtime訪問外部的MCP伺服器，需將代碼逐一嵌入AI代理中。

因此，每當所使用的MCP伺服器有變更時，就必須每次修正代碼。

未來，只需在Gateway側進行MCP伺服器的設定，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伺服器。

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
    # 直接有提示的情況
    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。

    所需的環境變數：
    - 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範圍: {self.cognito_scope}")
        logger.info(f"Workload名稱: {self.workload_name}")
        logger.info(f"用戶ID: {self.user_id}")
        logger.info(f"AWS區域: {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訪問令牌
        @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訪問令牌
                - workload_name: 來源於Runtime環境
                - user_id: 來自InvokeAgentRuntime的runtimeUserId標頭
            2. 使用workload訪問令牌獲取OAuth令牌
            3. 將access_token參數注入

            Args:
                access_token: OAuth訪問令牌（由裝飾器注入）

            Returns:
                str: 用於API調用的訪問令牌
            """
            logger.info("✅ 成功透過AgentCore Identity獲取訪問令牌")
            logger.info(f"   Workload名稱: {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("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傳輸: {self.gateway_url}")
            logger.info(f"🔑 令牌前綴: {access_token[:20]}...")
            transport = streamablehttp_client(
                self.gateway_url, 
                headers={"Authorization": f"Bearer {access_token}"}
            )
            logger.info("✅ MCP傳輸創建完成")
            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向的代理構建器。
    - 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類別的代理並返回。
        """
        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函數（Lambda functions）**\n- 可將AWS Lambda函數連接為工具\n- 可以用任意程式語言實現自訂商業邏輯\n- Gateway會自動調用Lambda函數，並將響應轉換為MCP格式\n\n## 2. **OpenAPI規範（OpenAPI specifications）**\n- 可將現有的REST API轉換為MCP兼容的工具\n- 只需提供OpenAPI規範，Gateway會自動處理MCP與REST格式之間的轉換\n- 此目標類型需要設定AgentCore Credential Provider（保存API金鑰或OAuth認證資訊）\n\n## 3. **Smithy模型（Smithy models）**\n- 可用Smithy模型定義API介面並生成MCP兼容工具\n- Smithy可用於生成與AWS服務或任何API的互動工具\n- Gateway會使用Smithy模型生成工具\n\n## 4. **MCP伺服器（MCP servers）**\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伺服器已成功設置與使用！