View a markdown version of this page

在网关上使用 MCP 会 AgentCore 话 - 亚马逊基岩 AgentCore

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

在网关上使用 MCP 会 AgentCore 话

MCP 会话支持客户端和网关之间的状态交互。 AgentCore 启用会话后,网关会在初始化期间生成一个唯一的会话标识符,并保持多个请求的状态,从而启用高级的 MCP 功能,例如激发和采样。

使用会话的好处

有状态的 MCP 服务器目标交互

网关存储 MCP 服务器目标的会话 ID,并在后续的工具调用中重复使用它。这样可以避免对每个请求进行重新初始化,并使目标能够在调用之间维护上下文。

使用 AgentCore 运行时目标实现更快的响应

当目标的会话被重复使用时, AgentCore Runtime 不需要对每个请求冷启动新的 MCP 服务器连接,从而缩短了响应时间。

启用高级 MCP 功能

会话是启动和采样的先决条件,这需要跟踪多个请求的状态。

User-scoped 安全(经过身份验证的网关)

对于具有入站身份验证的网关,会话绑定到经过验证的用户身份,从而防止会话劫持。

在网关上启用会话

要启用会话,请在创建或更新网关时sessionConfiguration在protocolConfiguration.mcp字段中指定。

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

sessionTimeoutInSeconds 参数是可选的。如果省略,则默认超时为 3600 秒(1 小时)。有效范围为 900(15 分钟)到 28800(8 小时)。超时是绝对的,根据第一个initialize请求计算得出。

要同时启用依赖会话的功能,例如激发和采样,还必须启用响应流:

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

在网关上启用会话后,您不能将会话包含Mcp-Session-Id在metadataConfiguration网关目标的标头传播设置中。网关在内部管理会话 ID。尝试这样做会返回 HTTP 400 错误请求错误。

会话生命周期

会话生命周期遵循 MCP 协议的初始化流程:

  1. 客户端向网关发送initialize请求。

  2. 网关创建会话,存储会话元数据,并在响应标头Mcp-Session-Id中返回唯一值。

  3. 客户端在所有后续请求中都包含Mcp-Session-Id标头。

  4. 网关会验证每个请求的会话存在、到期时间和用户身份(适用于经过身份验证的网关)。

  5. 当会话超时或客户端断开连接时,会话将过期。

在会话中首次调用 MCP 服务器目标的工具时,网关会初始化与目标的连接并存储目标的会话 ID。对同一目标的后续工具调用会重复使用该存储的会话 ID,从而避免了重复初始化。

用户身份和会话范围界定

会话的范围限定为经过身份验证的用户身份,以防止会话劫持。根据网关上配置的入站身份验证方法,网关派生用户身份的方式有所不同:

身份验证方法 用户标识符 行为

OAuth/OIDC

sub从 JWT 代币中申领

范围齐全。只有创建会话的用户才能使用它。该sub索赔是OIDC规范所要求的,在发行人中是本地唯一的,区分大小写,并且从不重新分配。

AWS IAM (SigV4)

主体 ARN

范围齐全。只有创建会话的 IAM 委托人才能使用它。主要 ARN 在全球范围内是独一无二的 AWS,在 IAM 实体的生命周期内是不可变的。示例:arn:aws:iam::123456789012:user/john-doe

不使用身份验证

无

没有用户范围限制。会话可用,但不绑定到任何身份。拥有会话 ID 的任何人都可以与会话进行交互。

重要

对于没有入站身份验证的网关,会话存在会话劫持风险,如 MCP 规范安全注意事项中所述。如果会话 ID 被泄露或被猜到,另一方可以恢复会话。仅将未经身份验证的会话用于开发和测试,不用于处理敏感数据的生产工作负载。

对于经过身份验证的网关,如果其他用户尝试使用现有会话 ID,则网关返回 HTTP 404 Not Found,其他用户看不到该会话。

会话超时和到期

会话超时是根据第一个initialize请求计算得出的。超时时间过后,会话将过期且无法使用。

  • 默认超时:3600 秒(1 小时)

  • 可配置范围:900 秒(15 分钟)至 28800 秒(8 小时)

如果 MCP 服务器目标的会话在网关会话超时之前过期或丢失(例如,如果目标重启),则对该目标的后续工具调用将返回客户端错误 (4xx),例如。session not found要恢复,请发送新initialize请求以启动新的网关会话,重新初始化与网关的 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)