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

# 使用以下命令开始双向流式传输 WebSocket
<a name="runtime-get-started-websocket"></a>

Amazon Bedrock AgentCore Runtime 允许您部署支持 WebSocket 流式传输的代理，实现实时双向通信。本指南引导您使用 WebSocket创建、测试和部署您的第一个双向流媒体代理。

在本部分中，您将学习：
+  AgentCore Runtime 如何支持 WebSocket 连接
+ 如何创建具有双向流媒体功能的代理应用程序
+ 如何在本地测试您的代理
+ 如何将代理部署到 Amazon 
+ 如何调用已部署的代理
+ 如何使用带有 WebSocket 连接的会话

有关该 WebSocket 协议的更多信息，请参阅 [ WebSocket RFC 645 ](https://tools.ietf.org/html/rfc6455) 5。

**Topics**
+ [AgentCore Runtime 如何支持 WebSocket 连接](#websocket-support-overview)
+ [WebSocket 与 AgentCore 运行时一起使用](#using-websocket-with-runtime)
+ [会话管理](#websocket-session-management)
+ [可观测性](#websocket-observability)
+ [自定义标头](#websocket-custom-headers)
+ [附录](#websocket-appendix)

## AgentCore Runtime 如何支持 WebSocket 连接
<a name="websocket-support-overview"></a>

AgentCore Runtime WebSocket 支持在客户端和代理之间实现持久的双向流媒体连接。 AgentCore Runtime 期望容器`8080`在`/ws`路径的端口上实现 WebSocket 端点，这符合标准 WebSocket 服务器惯例。

AgentCore Runtime 的 WebSocket 支持提供了与之相同的无服务器、会话隔离、身份和可观察性功能。`InvokeAgentRuntime`此外，它使用SigV4或OAuth 2.0身份验证通过 WebSocket 连接实现低延迟、实时的双向消息传输，使其成为实时对话语音代理等应用程序的理想之选。

### 支持的 WebSocket 库
<a name="websocket-supported-libraries"></a>

 WebSockets 在 AgentCore Runtime 上使用的双向流式传输支持使用任何 WebSocket 语言库的应用程序。唯一的要求是客户端通过 WebSocket 协议连接连接到服务端点：

```
wss://bedrock-agentcore.<region>.amazonaws.com/runtimes/<agentRuntimeArn>/ws
```

使用一种支持的身份验证方法（SigV4 标头、SigV4 预签名 URL 或 OAuth 2.0），并且代理应用程序按照 HTTP 协议合同中的规定实现 WebSocket 服务合同。[HTTP 协议合约](runtime-http-protocol-contract.md)

这种灵活性允许您在不同的编程语言和框架中使用您的首选 WebSocket 实现，从而确保与现有代码库和开发工作流程的兼容性。

## WebSocket 与 AgentCore 运行时一起使用
<a name="using-websocket-with-runtime"></a>

在本入门教程中，您将使用 b ** edrock-agentcore Python SDK ** 和 CLI 进行部署，创建、测试和部署支持双向流式传输的代理应用程序。** AgentCore **

**Topics**
+ [先决条件](#websocket-prerequisites)
+ [第 1 步：设置项目并安装依赖项](#setup-websocket-project)
+ [第 2 步：创建您的双向流媒体代理](#create-websocket-agent)
+ [第 3 步：在本地测试您的双向流媒体代理](#step-2-test-websocket-locally)
+ [第 4 步：将双向流媒体代理部署到 AgentCore Runtime](#step-3-deploy-websocket-to-aws)
+ [第 5 步：调用已部署的双向流媒体代理](#step-4-invoke-deployed-websocket)

### 先决条件
<a name="websocket-prerequisites"></a>

在开始之前，请确保：
+  ** Amazon 配置**了凭据的账户。要配置您的 Amazon 证书，请参阅 C Amazon LI 中的[配置和凭据文件设置。](https://docs.amazonaws.cn/cli/latest/userguide/cli-configure-files.html)
+  **已安装 Python 3.10\+ **
+  **Node.js 已安装 20 多个 **
+  ** Amazon 权限**：要使用 AgentCore CLI 创建和部署代理，必须具有相应的权限。有关更多信息，请参阅[使用 AgentCore CLI ](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/runtime-permissions.html#runtime-permissions-cli)。

### 第 1 步：设置项目并安装依赖项
<a name="setup-websocket-project"></a>

创建项目文件夹并安装所需的软件包：

```
mkdir agentcore-runtime-quickstart-websocket
cd agentcore-runtime-quickstart-websocket
python3 -m venv .venv
source .venv/bin/activate
```

将 pip 升级到最新版本：

```
pip install --upgrade pip
```

安装以下必需的软件包：
+  **bedrock-agentcore **-用于构建 AI 代理的亚马逊 Bedrock AgentCore SDK，包括 python 库依赖项 `websockets`

```
pip install bedrock-agentcore
```

### 第 2 步：创建您的双向流媒体代理
<a name="create-websocket-agent"></a>

为您的双向流媒体代理代码创建一个名`websocket_echo_agent.py`为的源文件。添加以下代码：

#### 双向流媒体代理的实现 (websocket\_echo\_agent.py)
<a name="websocket-agent-code"></a>

```
from bedrock_agentcore import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

@app.websocket
async def websocket_handler(websocket, context):
    """Simple echo WebSocket handler."""
    await websocket.accept()

    try:
        data = await websocket.receive_json()
        # Echo back
        await websocket.send_json({"echo": data})
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await websocket.close()

if __name__ == "__main__":
    app.run(log_level="info")
```

创建`requirements.txt`并添加以下内容：

```
bedrock-agentcore
```

包含 python `websockets` 库依赖关系

#### 理解代码
<a name="understanding-websocket-code"></a>
+  **BedrockAgentCoreApp**: 创建代理应用程序，用于扩展 Starlette 以进行 AI 代理部署，提供 WebSocket 支持、HTTP 路由、中间件和异常处理功能
+  **WebSocket 装饰器**：`@app.websocket`装饰器自动处理端口 8080 上`/ws`路径上的连接
+  **Echo Logic**：使用发送回接收到的数据 `{"echo": data}` 
+  **错误处理**：使用 try/except /finally 结构来确保正确的错误记录和正常关闭连接。

### 第 3 步：在本地测试您的双向流媒体代理
<a name="step-2-test-websocket-locally"></a>

#### 启动您的双向流媒体代理
<a name="start-websocket-server"></a>

打开终端窗口，使用以下命令启动双向流媒体代理：

```
python websocket_echo_agent.py
```

您应该看到表明服务器正在端口 8080 上运行的输出。

#### 测试 WebSocket 连接
<a name="test-websocket-connection"></a>

创建一个名为的本地 WebSocket 客户端`websocket_agent_client.py`：

##### 本地 WebSocket 客户端 (websocket\_agent\_client.py)
<a name="websocket-test-client"></a>

```
import asyncio
import websockets
import json

async def local_websocket():
    uri = "ws://localhost:8080/ws"

    try:
        async with websockets.connect(uri) as websocket:
            # Send a message
            await websocket.send(json.dumps({"inputText": "Hello WebSocket!"}))

            # Receive the echo response
            response = await websocket.recv()
            print(f"Received: {response}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(local_websocket())
```

打开另一个终端窗口并运行客户端，在本地测试您的双向流媒体代理：

```
python websocket_agent_client.py
```

 **成功：**您应该看到类似的回复`Received: {"echo":{"inputText":"Hello WebSocket!"}}`。在运行代理的终端窗口中，输入`Ctrl+C`以停止代理。

### 第 4 步：将双向流媒体代理部署到 AgentCore Runtime
<a name="step-3-deploy-websocket-to-aws"></a>

#### 安装部署工具
<a name="install-deployment-tools"></a>

安装 C AgentCore LI：

```
npm install -g @aws/agentcore
```

验证安装：

```
agentcore --version
```

有关可用的命令和选项，请参阅 [ AgentCore CLI 参考](agentcore-cli-reference.md)。

#### 创建项目并部署到 Amazon
<a name="configure-and-deploy"></a>

为您的双向流媒体代理创建一个新项目：

```
cd ..
agentcore create --project-name WebSocketProject --no-agent
cd WebSocketProject
agentcore add agent \
  --name WebSocketAgent \
  --type byo \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --protocol HTTP \
  --code-location ../agentcore-runtime-quickstart-websocket \
  --entrypoint websocket_echo_agent.py
```

部署您的代理：

```
agentcore deploy
```

该 AgentCore 项目引用了现有的`agentcore-runtime-quickstart-websocket`源目录。

部署后，您将收到一个代理运行时 ARN，如下所示：

```
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/websocket_echo_agent-xyz123
```

保存此 ARN，因为您需要它来调用已部署的代理。

### 第 5 步：调用已部署的双向流媒体代理
<a name="step-4-invoke-deployed-websocket"></a>

#### 设置环境变量
<a name="setup-environment-variables"></a>

设置所需的环境变量：

1. 导出您的代理 ARN：

   ```
   export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/websocket_echo_agent-xyz123"
   ```

1. 如果使用 OAuth，请导出您的持有者令牌：

   ```
   export BEARER_TOKEN="your_oauth_token_here"
   ```

#### 身份验证方法
<a name="websocket-authentication-methods"></a>

`InvokeAgentRuntimeWithWebSocketStream`API 操作建立了支持客户端和代理之间双向流式传输的 WebSocket 连接。您可以使用以下方法对 WebSocket 连接进行身份验证：
+  ** Amazon 签名版本 4 标头**：使用您的 Amazon 证书对 WebSocket 握手请求标头进行签名
+  ** Amazon 签名版本 4 Pre-signed 网址**：使用 SigV4 签名作为查询参数创建预签名 WebSocket URL
+  **OAuth 持有者令牌**：在授权标头中传递 OAuth 令牌以进行外部身份提供商集成

**提示**  
确保你有`bedrock-agentcore:InvokeAgentRuntimeWithWebSocketStream`权限。

#### 使用 SigV4 签名标头进行连接
<a name="websocket-sigv4-headers"></a>

以下示例显示如何使用 SigV4 签名标头建立 WebSocket 连接并与代理运行时通信：

##### WebSocket 带有 SigV4 标头的客户端 (websocket\_agent\_client\_sigv4\_headers.py)
<a name="websocket-sigv4-example"></a>

```
from bedrock_agentcore.runtime import AgentCoreRuntimeClient
import websockets
import asyncio
import json
import os

async def main():
    # Get runtime ARN from environment variable
    runtime_arn = os.getenv('AGENT_ARN')
    if not runtime_arn:
        raise ValueError("AGENT_ARN environment variable is required")

    # Initialize client
    client = AgentCoreRuntimeClient(region="us-west-2")

    # Generate WebSocket connection with authentication
    ws_url, headers = client.generate_ws_connection(
        runtime_arn=runtime_arn
    )

    try:
        async with websockets.connect(ws_url, additional_headers=headers) as ws:
            # Send message
            await ws.send(json.dumps({"inputText": "Hello!"}))

            # Receive response
            response = await ws.recv()
            print(f"Received: {response}")
    except websockets.exceptions.InvalidStatus as e:
        print(f"WebSocket handshake failed with status code: {e.response.status_code}")
        print(f"Response headers: {e.response.headers}")
        print(f"Response body: {e.response.body.decode()}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())
```

运行客户端来测试您部署的代理：

```
python websocket_agent_client_sigv4_headers.py
```

 **成功：**你应该看到如下回复：

```
Received: {"echo":{"inputText":"Hello!"}}
```

#### 使用预签名 URL 进行连接（通过查询参数进行 SigV4）
<a name="websocket-sigv4-presigned"></a>

以下示例说明如何使用 SigV4 查询参数创建 WebSocket URL 并建立连接：

##### WebSocket 带有预签名 URL 的客户端 (websocket\_agent\_client\_sigv4\_presigned\_url.py)
<a name="websocket-presigned-example"></a>

```
from bedrock_agentcore.runtime import AgentCoreRuntimeClient
import websockets
import asyncio
import json
import os

async def main():
    runtime_arn = os.getenv('AGENT_ARN')
    if not runtime_arn:
        raise ValueError("AGENT_ARN environment variable is required")

    client = AgentCoreRuntimeClient(region="us-west-2")

    # Generate WebSocket pre-signed URL (with SigV4 via query parameters)
    # wss://...amazonaws.com/runtimes/.../ws?X-Amz-Algorithm=AWS4-HMAC-SHA256
    #   &X-Amz-Credential=...&X-Amz-Date=...&X-Amz-Expires=300
    #   &X-Amz-SignedHeaders=...&X-Amz-Signature=...
    sigv4_url = client.generate_presigned_url(
        runtime_arn=runtime_arn,
        expires=300  # 5 minutes
    )

    try:
        async with websockets.connect(sigv4_url) as ws:
            await ws.send(json.dumps({"inputText": "Hello!"}))
            response = await ws.recv()
            print(f"Received: {response}")
    except websockets.exceptions.InvalidStatus as e:
        print(f"WebSocket handshake failed with status code: {e.response.status_code}")
        print(f"Response headers: {e.response.headers}")
        print(f"Response body: {e.response.body.decode()}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())
```

运行客户端来测试您部署的代理：

```
python websocket_agent_client_sigv4_query_parameters.py
```

 **成功：**你应该看到如下回复：

```
Received: {"echo":{"inputText":"Hello!"}}
```

#### 使用 OAuth 连接
<a name="websocket-oauth"></a>

AgentCore 运行时支持对连接进行 OAuth 持有者令牌身份验证。 WebSocket 要使用 OAuth 身份验证，您需要按照 “使用入站身份验证和出站身份验证和出站身份验证[进行身份验证和授权” 的 [ JWT 入站授权和 OAuth 出站访问示例](runtime-oauth.md#oauth-sample-overview)部分中所述，使用 JWT 授权配置代理运行时。](runtime-oauth.md)

完成 OAuth 设置并按照[步骤 4：使用不记名令牌调用 OAuth 指南](runtime-oauth.md#oauth-invoke-agent)中的代理获得持有者令牌后，即可使用该令牌建立连接。 WebSocket 

##### 带有 OAuth 的 Python 客户端
<a name="websocket-oauth-python"></a>

以下示例显示如何使用 OAuth 与 Python 建立 WebSocket 连接：

##### WebSocket 使用 OAuth 身份验证的客户端 (websocket\_agent\_client\_oauth.py)
<a name="websocket-oauth-example"></a>

```
from bedrock_agentcore.runtime import AgentCoreRuntimeClient
import websockets
import asyncio
import json
import os

async def main():
    # Get runtime ARN from environment variable
    runtime_arn = os.getenv('AGENT_ARN')
    if not runtime_arn:
        raise ValueError("AGENT_ARN environment variable is required")

    # Get OAuth bearer token from environment variable
    bearer_token = os.getenv('BEARER_TOKEN')
    if not bearer_token:
        raise ValueError("BEARER_TOKEN environment variable required for OAuth")

    # Initialize client
    client = AgentCoreRuntimeClient(region="us-west-2")

    # Generate WebSocket connection with OAuth
    ws_url, headers = client.generate_ws_connection_oauth(
        runtime_arn=runtime_arn,
        bearer_token=bearer_token
    )

    try:
        async with websockets.connect(ws_url, additional_headers=headers) as ws:
            # Send message
            await ws.send(json.dumps({"inputText": "Hello!"}))

            # Receive response
            response = await ws.recv()
            print(f"Received: {response}")
    except websockets.exceptions.InvalidStatus as e:
        print(f"WebSocket handshake failed with status code: {e.response.status_code}")
        print(f"Response headers: {e.response.headers}")
        print(f"Response body: {e.response.body.decode()}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())
```

运行客户端来测试您部署的代理：

```
python websocket_agent_client_oauth.py
```

 **成功：**你应该看到如下回复：

```
Received: {"echo":{"inputText":"Hello!"}}
```

##### 带有 OA JavaScript uth 的浏览器客户端
<a name="websocket-oauth-browser"></a>

浏览器的本机 WebSocket API 不提供在握手期间设置自定义标头的方法。为了支持浏览器的 OAuth 身份验证， AgentCore Runtime 在握手期间接受`Sec-WebSocket-Protocol`标头中嵌入的持有者令牌。 WebSocket 

该令牌必须采用 base64url 编码并以前缀为前缀`base64UrlBearerAuthorization.`，然后是哨兵子协议。`base64UrlBearerAuthorization`

以下示例显示如何 JavaScript 使用 OAuth 与浏览器建立 WebSocket 连接：

##### 带有 OAuth 的浏览器 WebSocket 客户端 (index.html)
<a name="websocket-oauth-browser-example"></a>

```
<!DOCTYPE html>
<html>
<body>
    <button onclick="connect()">Connect</button>
    <div id="output"></div>

    <script>
        function connect() {
            const bearerToken = "your_oauth_token_here";
            const runtimeArn = "arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/agent-xyz123";

            // Base64url encode token
            const base64url = btoa(bearerToken)
                .replace(/\+/g, '-')
                .replace(/\//g, '_')
                .replace(/=/g, '');

            const ws = new WebSocket(
                `wss://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${runtimeArn}/ws`,
                [`base64UrlBearerAuthorization.${base64url}`, "base64UrlBearerAuthorization"]
            );

            ws.onopen = () => ws.send(JSON.stringify({ inputText: "Hello!" }));
            ws.onmessage = (e) => document.getElementById("output").innerText = e.data;
        }
    </script>
</body>
</html>
```

**注意**  
此身份验证方法适用于无法设置自定义标头的基于浏览器的客户端。对于非浏览器客户端（Python、 Node.js 服务器等），使用带有 OAuth 的 P [ ython 客户端中显示的 OAuth 标头身份验证。](#websocket-oauth-python)

**注意**  
尚不支持除`base64UrlBearerAuthorization`之外的子协议。

**重要**  
这是一个参考示例。不建议在生产代码中对令牌进行硬编码。

## 会话管理
<a name="websocket-session-management"></a>

在 WebSocket 连接上提供 `session_id` (`X-Amzn-Bedrock-AgentCore-Runtime-Session-Id`)（作为 URL 查询参数或请求标头）会将连接路由到隔离的运行时会话。代理可以访问存储在该会话中的对话上下文，通过引用先前的交互来实现对话的连续性。不同的会话 ID 访问不同的隔离上下文，从而确保用户或对话之间的完全隔离。

有关包括跟踪、清理和错误处理在内的全面会话生命周期管理，请参阅为代理[使用隔离会话](runtime-sessions.md)。

### 使用带有 WebSocket 连接的会话
<a name="websocket-with-sessions"></a>

要使用带有 WebSocket 连接的会话，请为每个用户或对话生成一个唯一的会话 ID，并在建立连接时将其传递：

**Example**  

1. 

   ```
   from bedrock_agentcore.runtime import AgentCoreRuntimeClient
   import websockets
   import asyncio
   import json
   import os
   
   async def websocket_with_session():
       client = AgentCoreRuntimeClient(region="us-west-2")
       session_id = "user-123-conversation-456"
       runtime_arn = os.getenv('AGENT_ARN')
   
       ws_url, headers = client.generate_ws_connection(
           runtime_arn=runtime_arn,
           session_id=session_id
       )
   
       try:
           async with websockets.connect(ws_url, additional_headers=headers) as ws:
               await ws.send(json.dumps({"inputText": "Hello!"}))
               response = await ws.recv()
               print(f"Response: {response}")
       except websockets.exceptions.InvalidStatus as e:
           print(f"WebSocket handshake failed with status code: {e.response.status_code}")
           print(f"Response headers: {e.response.headers}")
           print(f"Response body: {e.response.body.decode()}")
       except Exception as e:
           print(f"Connection failed: {e}")
   
   asyncio.run(websocket_with_session())
   ```

1. 

   ```
   from bedrock_agentcore.runtime import AgentCoreRuntimeClient
   import websockets
   import asyncio
   import json
   import os
   
   async def websocket_with_session():
       client = AgentCoreRuntimeClient(region="us-west-2")
       session_id = "user-123-conversation-456"
       runtime_arn = os.getenv('AGENT_ARN')
   
       presigned_url = client.generate_presigned_url(
           runtime_arn=runtime_arn,
           session_id=session_id,
           expires=300
       )
   
       try:
           async with websockets.connect(presigned_url) as ws:
               await ws.send(json.dumps({"inputText": "Hello!"}))
               response = await ws.recv()
               print(f"Response: {response}")
       except websockets.exceptions.InvalidStatus as e:
           print(f"WebSocket handshake failed with status code: {e.response.status_code}")
           print(f"Response headers: {e.response.headers}")
           print(f"Response body: {e.response.body.decode()}")
       except Exception as e:
           print(f"Connection failed: {e}")
   
   asyncio.run(websocket_with_session())
   ```

1. 

   ```
   from bedrock_agentcore.runtime import AgentCoreRuntimeClient
   import websockets
   import asyncio
   import json
   import os
   
   async def websocket_with_session():
       client = AgentCoreRuntimeClient(region="us-west-2")
       session_id = "user-123-conversation-456"
       runtime_arn = os.getenv('AGENT_ARN')
       bearer_token = os.getenv('BEARER_TOKEN')
   
       ws_url, headers = client.generate_ws_connection_oauth(
           runtime_arn=runtime_arn,
           session_id=session_id,
           bearer_token=bearer_token
       )
   
       try:
           async with websockets.connect(ws_url, additional_headers=headers) as ws:
               await ws.send(json.dumps({"inputText": "Hello!"}))
               response = await ws.recv()
               print(f"Response: {response}")
       except websockets.exceptions.InvalidStatus as e:
           print(f"WebSocket handshake failed with status code: {e.response.status_code}")
           print(f"Response headers: {e.response.headers}")
           print(f"Response body: {e.response.body.decode()}")
       except Exception as e:
           print(f"Connection failed: {e}")
   
   asyncio.run(websocket_with_session())
   ```

**提示**  
为获得最佳结果，请为会话 ID 使用 UUID 或其他唯一标识符，以避免不同用户或对话之间发生冲突。

通过对相关 WebSocket 连接使用相同的会话 ID，您可以确保在同一个对话中保持上下文，从而使您的代理能够在先前的交互基础上提供连贯的响应。

### 带有 WebSocket 连接的会话生命周期
<a name="websocket-session-lifecycle"></a>

对于 WebSocket 连接，每次客户端和代理之间有消息活动时，会话的空闲超时都会重置。这包括任何 WebSocket 消息交换，例如从客户端发送数据、接收代理到客户端的响应或 WebSocket ping/pong 框架。这意味着，只要消息继续流动，活跃的 WebSocket 对话就会使会话保持活动状态，从而防止在持续互动期间过早终止会话。

有关配置生命周期设置的更多信息，请参阅[配置 Amazon Bedrock AgentCore 生命周期设置](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/runtime-lifecycle-settings.html)。要通过代理运行状况更直接地控制会话生命周期，请参阅[运行时会话生命周期管理](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/runtime-long-run.html#runtime-long-run-session-lifecycle)。

### 停止运行时会话
<a name="websocket-stop-session"></a>

要在可配置会话`IdleRuntimeSessionTimeout`（默认为 15 分钟）之前停止正在运行的会话，请参阅[停止正在运行的会话](runtime-stop-session.md)。

## 可观测性
<a name="websocket-observability"></a>

 [亚马逊基岩可 AgentCore 观测性](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/observability.html)可帮助您跟踪、调试和监控您在亚马逊基岩 AgentCore 运行时托管的代理。首先，按照启用 Amazon Bedrock AgentCore 运行时可观察性[中的说明](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/observability-configure.html#observability-configure-builtin)启用 CloudWatch 交易搜索。要观察您的代理，请参阅[查看您的亚马逊 Bedrock AgentCore 代理的可观测性数据。](https://docs.amazonaws.cn/bedrock-agentcore/latest/devguide/observability-view.html)

对于 WebSocket 连接，跟踪代表完整的连接会话，而不是单个消息交换。

## 自定义标头
<a name="websocket-custom-headers"></a>

自定义标头允许您在初始 WebSocket 连接时将来自应用程序的上下文信息直接传递到代理代码。有关自定义标头支持、配置和限制的完整信息，请参阅[将自定义标头传递给 Amazon Bedrock AgentCore Runtime ](runtime-header-allowlist.md)。

此外，前缀为的标头`X-Amzn-Bedrock-AgentCore-Runtime-Custom-`可以作为 WebSocket 连接中的 URL 查询参数传递。

例如，您可以在 WebSocket URL 中将自定义标头作为查询参数传递：

```
wss://bedrock-agentcore.<region>.amazonaws.com/runtimes/<agentRuntimeArn>/ws?X-Amzn-Bedrock-AgentCore-Runtime-Custom-TestHeader=query-param-test-value
```

代理应用程序容器将接收这些作为标头：

```
"headers": {
    "x-amzn-bedrock-agentcore-runtime-custom-testheader": "query-param-test-value"
  }
```

## 附录
<a name="websocket-appendix"></a>

**Topics**
+ [安全注意事项](#security-considerations)
+ [问题排查](#websocket-troubleshooting)
+ [WebSocket 与其他协议的对比](#websocket-vs-other-protocols)
+ [其他入门示例](#websocket-additional-examples)

### 安全注意事项
<a name="security-considerations"></a>

**提示**  
有关所有运行时安全建议的综合视图，请参阅 AgentCore 运行时[安全最佳实践](runtime-security-best-practices.md)。

身份验证  
所有 WebSocket 连接都需要通过 SigV4 或 OAuth 2.0 进行正确的 Amazon 身份验证

会话隔离  
每个会话都在带有专用资源的隔离执行环境中运行

传输安全  
所有连接均使用 HTTPS 上的 WSS（WebSocket 安全）进行加密通信

访问控制  
IAM 策略控制 WebSocket 连接权限和对特定代理的访问权限

### 问题排查
<a name="websocket-troubleshooting"></a>

#### 常见 WebSocket-specific 问题
<a name="common-websocket-issues"></a>

以下是您可能会遇到的常见问题：

连接失败  
验证您的代理应用程序是否在处理连接请求 `/ws` 

身份验证方法不匹配  
确保您的客户端使用与代理配置相同的身份验证方法（OAuth 或 SigV4）

由于超出限制，连接已关闭  
如果超出限制，例如消息帧速率或消息帧大小限制，连接将自动关闭。有关完整的限额信息，请参阅亚马逊 Bedrock 的[配额 AgentCore ](bedrock-agentcore-limits.md) 

已超过消息帧大小  
配置消息帧分段或实现分块以保持在 32KB 帧大小限制以下。在发送之前，将大型消息拆分成较小的块

运行状况检查失败  
确保您的代理容器按照 [ HTTP 协议合同中的规定实现`/ping`终端节点](runtime-http-protocol-contract.md)。此端点可验证您的代理是否处于运行状态并已准备好处理请求，从而实现服务监控和自动恢复

#### 错误处理
<a name="websocket-error-handling"></a>

WebSocket 错误分两个阶段出现，具体取决于它们发生的时间。

 **建立连接（ WebSocket 升级之前）** 

打开连接是一个标准的 HTTP 请求。HTTP 状态码反映异常，`x-amzn-ErrorType`响应标头带有异常名称。该服务在建立 WebSocket 连接之前可能会返回以下任何错误。


| HTTP 错误代码 | 运行时异常 (`x-amzn-ErrorType`) | 说明 | 
| --- | --- | --- | 
| 400 | ValidationException | 请求数据或参数无效 | 
| 401 | UnauthorizedException | 需要进行身份验证或凭据无效（OAuth-configured 代理） | 
| 402 | ServiceQuotaExceededException | 请求将超过服务配额 | 
| 403 | AccessDeniedException | 请求的操作权限不足 | 
| 404 | ResourceNotFoundException | 请求的资源不存在 | 
| 409 | ConflictException | 资源冲突-资源已经存在 | 
| 409 | RetryableConflictException | 会话操作正在进行中，请重试 | 
| 424 | RuntimeClientError | 您的代理的容器返回了 4xx 或 5xx 错误-请检查您的日志 CloudWatch  | 
| 429 | ThrottlingException | 请求过多-已超过请求速率限制 | 
| 500 | InternalServerException | 处理请求时出现意外错误 | 

**注意**  
当您打开与该服务正在配置或断开的会话的 WebSocket 连接时，该服务将返回 `RetryableConflictException` (HTTP 409`Session operation in progress, please retry`)。这种情况是暂时的，可以重试。使用短暂的指数退避重试。这适用于针对同一会话的并发调用。 Already-running 会话不受影响。

 **活动连接（ WebSocket 升级后）** 

一旦确定，将 WebSocket 使用标准 WebSocket 关闭代码而不是 HTTP 状态码来传达错误。常见的关闭代码包括：
+  `1000`-正常关闭
+  `1001`-走开
+  `1008`-违反政策（超过限制）
+  `1009`-消息太大（超过消息帧大小限制）
+  `1011`-服务器错误

### WebSocket 与其他协议的对比
<a name="websocket-vs-other-protocols"></a>

 **何时使用 WebSocket**：
+ Real-time 使用即时音频流进行语音对话，实现自然对话流程
+ 双向 audio/text /二进制数据流（将数据块从客户端流式传输到代理，反之亦然）
+ 中断处理（用户可以在对话中打断代理）

 **何时使用 HTTP**：
+ 无需双向流式传输的请求响应模式的 HTTP

### 其他入门示例
<a name="websocket-additional-examples"></a>

有关在 AgentCore Runtime 中使用 WebSocket 双向流式传输的其他示例，请参阅[WebSocket 双向流式传输 GitHub 示例：](https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-features/02-host-your-agent/01-runtime/03-advanced/03-bidirectional-streaming)
+  **声波实现 (Python)**：原生 Amazon Nova Sonic WebSocket 实现，支持实时音频对话、语音选择和中断
+  **Strands 实现 (Python)**：使用 Strands Framework-based BidiAgent 实现，通过自动会话管理和工具集成来简化实时音频对话
+  **Echo 实现 (Python)**：用于测试 WebSocket 连接和身份验证的简单 echo 服务器