View a markdown version of this page

在网关上使用 MCP 会 AgentCore 话 - Amazon Bedrock 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 服务器目标的会话在网关会话超时之前过期，则网关会透明地使用目标重新初始化并更新存储的目标会话 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)