搭配 AgentCore 閘道使用 MCP 工作階段
MCP 工作階段可啟用用戶端與 AgentCore 閘道之間的有狀態互動。啟用工作階段時,閘道會在初始化期間產生唯一的工作階段識別符,並維護多個請求的狀態,進而啟用引出和取樣等進階 MCP 功能。
使用工作階段的優點
- 狀態 MCP 伺服器目標互動
-
閘道會存放 MCP 伺服器目標的工作階段 ID,並在後續的工具呼叫中重複使用。這可避免每次請求重新初始化,並讓目標維持呼叫的內容。
- 使用 AgentCore 執行期目標加快回應速度
-
重複使用目標的工作階段時,AgentCore 執行期不需要在每個請求上冷啟動新的 MCP 伺服器連線,進而縮短回應時間。
- 啟用進階 MCP 功能
-
工作階段是引出和取樣的先決條件,需要跨多個請求追蹤狀態。
- 使用者範圍安全性 (已驗證的閘道)
-
對於具有傳入身分驗證的閘道,工作階段會繫結至已驗證的使用者身分,以防止工作階段劫持。
在閘道上啟用工作階段
若要啟用工作階段,請在建立或更新閘道時,sessionConfiguration在 protocolConfiguration.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 通訊協定的初始化流程:
-
用戶端會將 initialize請求傳送至閘道。
-
閘道會建立工作階段、儲存工作階段中繼資料,並在回應標頭Mcp-Session-Id中傳回唯一的 。
-
用戶端會在所有後續請求中包含 Mcp-Session-Id標頭。
-
閘道會驗證每個請求的工作階段存在、到期和使用者身分 (適用於已驗證的閘道)。
-
當工作階段逾時或用戶端中斷連線時,工作階段會過期。
在工作階段中對 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請求計算。在逾時期間之後,工作階段會過期且無法使用。
如果 MCP 伺服器目標的工作階段在閘道工作階段逾時之前過期,閘道會以透明的方式使用目標重新初始化,並更新儲存的目標工作階段 ID。閘道工作階段會保持作用中狀態。
錯誤處理
| 案例 |
HTTP 狀態 |
說明 |
|
啟用工作階段的閘道上缺少Mcp-Session-Id標頭
|
400 錯誤的請求
|
之後的所有請求initialize必須包含工作階段標頭。
|
|
無效或過期的工作階段 ID
|
404 找不到
|
工作階段不存在或已逾時。
|
|
不同的使用者嘗試使用另一個使用者的工作階段 (已驗證的閘道)
|
404 找不到
|
其他使用者看不到工作階段。
|
|
Mcp-Session-Id 啟用工作階段metadataConfiguration時在目標 中
|
400 錯誤的請求
|
在建立或更新目標時於控制平面傳回。
|
程式碼範例
範例
- curl
-
-
傳送啟動工作階段的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"
}
}
}
-
在後續請求中包含工作階段 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
-
-
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
-
-
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
-
-
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)