View a markdown version of this page

搭配 AgentCore 閘道使用 MCP 工作階段 - Amazon Bedrock AgentCore

搭配 AgentCore 閘道使用 MCP 工作階段

MCP 工作階段可啟用用戶端與 AgentCore 閘道之間的有狀態互動。啟用工作階段時,閘道會在初始化期間產生唯一的工作階段識別符,並維護多個請求的狀態,進而啟用引出和取樣等進階 MCP 功能。

使用工作階段的優點

狀態 MCP 伺服器目標互動

閘道會存放 MCP 伺服器目標的工作階段 ID,並在後續的工具呼叫中重複使用。這可避免每次請求重新初始化,並讓目標維持呼叫的內容。

使用 AgentCore 執行期目標加快回應速度

重複使用目標的工作階段時,AgentCore 執行期不需要在每個請求上冷啟動新的 MCP 伺服器連線,進而縮短回應時間。

啟用進階 MCP 功能

工作階段是引出取樣的先決條件,需要跨多個請求追蹤狀態。

使用者範圍安全性 (已驗證的閘道)

對於具有傳入身分驗證的閘道,工作階段會繫結至已驗證的使用者身分,以防止工作階段劫持。

在閘道上啟用工作階段

若要啟用工作階段,請在建立或更新閘道時,sessionConfigurationprotocolConfiguration.mcp 欄位中指定 。

{ "protocolConfiguration": { "mcp": { "sessionConfiguration": { "sessionTimeoutInSeconds": 3600 } } } }

sessionTimeoutInSeconds 為選用參數。如果省略,預設逾時為 3600 秒 (1 小時)。有效範圍為 900 (15 分鐘) 到 28800 (8 小時)。逾時是絕對的,從第一個initialize請求開始計算。

若要同時啟用相依於工作階段的功能,例如引出和取樣,您必須另外啟用回應串流:

{ "protocolConfiguration": { "mcp": { "sessionConfiguration": { "sessionTimeoutInSeconds": 3600 }, "streamingConfiguration": { "enableResponseStreaming": true } } } }
注意

在閘道上啟用工作階段時,您無法在metadataConfiguration閘道目標標頭傳播設定的 Mcp-Session-Id中包含 。閘道會在內部管理工作階段 IDs。嘗試這麼做會傳回 HTTP 400 錯誤請求錯誤。

工作階段生命週期

工作階段生命週期遵循 MCP 通訊協定的初始化流程:

  1. 用戶端會將 initialize請求傳送至閘道。

  2. 閘道會建立工作階段、儲存工作階段中繼資料,並在回應標頭Mcp-Session-Id中傳回唯一的 。

  3. 用戶端會在所有後續請求中包含 Mcp-Session-Id標頭。

  4. 閘道會驗證每個請求的工作階段存在、到期和使用者身分 (適用於已驗證的閘道)。

  5. 當工作階段逾時或用戶端中斷連線時,工作階段會過期。

在工作階段中對 MCP 伺服器目標進行第一次工具呼叫時,閘道會初始化與目標的連線,並存放目標的工作階段 ID。後續的工具會呼叫相同的目標,重複使用此預存工作階段 ID,避免重複初始化。

使用者身分和工作階段範圍

工作階段的範圍是已驗證的使用者身分,以防止工作階段遭到劫持。閘道會根據閘道上設定的傳入身分驗證方法,以不同的方式衍生使用者身分:

身分驗證方法 使用者識別符 Behavior (行為)

OAuth/OIDC

sub 來自 JWT 權杖的宣告

完整範圍。只有建立工作階段的使用者才能使用它。OIDC 規格需要此sub宣告,在發行者內部為本機唯一、區分大小寫且從未重新指派。

AWS IAM (SigV4)

主體 ARN

完整範圍。只有建立工作階段的 IAM 主體才能使用它。主體 ARN 在 IAM 實體的生命週期中是全域唯一的 AWS。範例:arn:aws:iam::123456789012:user/john-doe

無身分驗證

沒有使用者範圍。工作階段可用,但不會繫結至任何身分。具有工作階段 ID 的任何人都可以與工作階段互動。

重要

對於沒有傳入身分驗證的閘道,工作階段具有工作階段劫持風險,如 MCP 規格安全考量中所述。如果工作階段 ID 遭到洩漏或猜測,另一方可以繼續工作階段。僅將未經驗證的工作階段用於開發和測試,不適用於處理敏感資料的生產工作負載。

對於已驗證的閘道,如果不同的使用者嘗試使用現有的工作階段 ID,閘道會傳回 HTTP 404 找不到 — 其他使用者看不到工作階段。

工作階段逾時和過期

工作階段逾時是根據第一個initialize請求計算。在逾時期間之後,工作階段會過期且無法使用。

  • 預設逾時:3600 秒 (1 小時)

  • 可設定的範圍:900 秒 (15 分鐘) 到 28800 秒 (8 小時)

如果 MCP 伺服器目標的工作階段在閘道工作階段逾時之前過期,閘道會以透明的方式使用目標重新初始化,並更新儲存的目標工作階段 ID。閘道工作階段會保持作用中狀態。

錯誤處理

案例 HTTP 狀態 說明

啟用工作階段的閘道上缺少Mcp-Session-Id標頭

400 錯誤的請求

之後的所有請求initialize必須包含工作階段標頭。

無效或過期的工作階段 ID

404 找不到

工作階段不存在或已逾時。

不同的使用者嘗試使用另一個使用者的工作階段 (已驗證的閘道)

404 找不到

其他使用者看不到工作階段。

Mcp-Session-Id 啟用工作階段metadataConfiguration時在目標 中

400 錯誤的請求

在建立或更新目標時於控制平面傳回。

程式碼範例

範例
curl
  1. 傳送啟動工作階段的initialize請求:

    curl -X POST \ https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json" \ -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \ -d '{ "jsonrpc": "2.0", "id": "init-request", "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "my-agent", "version": "1.0.0" } } }'

    回應包含 Mcp-Session-Id標頭:

    HTTP/1.1 200 OK Mcp-Session-Id: session-abc123def456 Content-Type: application/json { "jsonrpc": "2.0", "id": "init-request", "result": { "protocolVersion": "2025-06-18", "capabilities": { "tools": { "listChanged": true } }, "serverInfo": { "name": "agentcore-gateway", "version": "1.0.0" } } }
  2. 在後續請求中包含工作階段 ID:

    curl -X POST \ https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json" \ -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \ -H "Mcp-Session-Id: session-abc123def456" \ -d '{ "jsonrpc": "2.0", "id": "call-tool-request", "method": "tools/call", "params": { "name": "searchProducts", "arguments": { "query": "wireless headphones" } } }'
Python requests package
  1. import requests import json gateway_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp" headers = { "Content-Type": "application/json", "Accept": "application/json", "Authorization": "Bearer YOUR_ACCESS_TOKEN" } # Step 1: Initialize and get session ID init_response = requests.post(gateway_url, headers=headers, json={ "jsonrpc": "2.0", "id": "init-request", "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "my-agent", "version": "1.0.0"} } }) session_id = init_response.headers["Mcp-Session-Id"] print(f"Session ID: {session_id}") # Step 2: Use session ID in subsequent requests headers["Mcp-Session-Id"] = session_id tool_response = requests.post(gateway_url, headers=headers, json={ "jsonrpc": "2.0", "id": "call-tool-request", "method": "tools/call", "params": { "name": "searchProducts", "arguments": {"query": "wireless headphones"} } }) print(json.dumps(tool_response.json(), indent=2))
MCP Client
  1. from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client import asyncio async def use_session(url, token): headers = {"Authorization": f"Bearer {token}"} async with streamablehttp_client(url=url, headers=headers) as ( read_stream, write_stream, _ ): async with ClientSession(read_stream, write_stream) as session: # Initialize - session ID is managed automatically by the MCP client init_response = await session.initialize() print(f"Initialized: {init_response}") # Subsequent calls reuse the session automatically tool_response = await session.call_tool( name="searchProducts", arguments={"query": "wireless headphones"} ) print(f"Tool response: {tool_response}") return tool_response asyncio.run(use_session( url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp", token="YOUR_ACCESS_TOKEN" ))
Strands MCP Client
  1. from mcp.client.streamable_http import streamablehttp_client from strands import Agent from strands.tools.mcp import MCPClient mcp_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp" access_token = "YOUR_ACCESS_TOKEN" mcp_client = MCPClient( lambda: streamablehttp_client( mcp_url, headers={"Authorization": f"Bearer {access_token}"} ) ) # Strands MCP client handles session management automatically with mcp_client: agent = Agent(tools=mcp_client.list_tools_sync()) response = agent("Search for wireless headphones") print(response)