Comience con la transmisión bidireccional con WebSocket

Amazon Bedrock AgentCore Runtime le permite implementar agentes que admiten la WebSocket transmisión para una comunicación bidireccional en tiempo real. Esta guía explica cómo crear, probar e implementar su primer agente de streaming bidireccional utilizando. WebSocket

En esta sección, aprenderá lo siguiente:

Para obtener más información sobre el WebSocket protocolo, consulte el WebSocket RFC 6455.

Cómo admite AgentCore Runtime las conexiones WebSocket

AgentCore El WebSocket soporte de Runtime permite conexiones de transmisión bidireccionales y persistentes entre clientes y agentes. AgentCore Runtime espera que los contenedores implementen WebSocket puntos finales en el puerto de la /ws ruta, lo que se ajusta 8080 a las prácticas estándar WebSocket de los servidores.

AgentCore El WebSocket soporte de Runtime proporciona las mismas capacidades sin servidor, aislamiento de sesiones, identidad y observabilidad que. InvokeAgentRuntime Además, permite la transmisión bidireccional de mensajes en tiempo real y de baja latencia a través de WebSocket conexiones mediante la autenticación SiGv4 u OAuth 2.0, lo que lo hace ideal para aplicaciones como los agentes de voz conversacionales en tiempo real.

Bibliotecas compatibles WebSocket

La transmisión bidireccional que se utiliza WebSockets en AgentCore Runtime admite aplicaciones que utilizan cualquier biblioteca de WebSocket idiomas. Los únicos requisitos son que los clientes se conecten al punto final del servicio mediante una conexión de WebSocket protocolo:

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

utilizando uno de los métodos de autenticación compatibles (encabezados de SigV4, URL prefirmada de SigV4 u OAuth 2.0) y que la aplicación agente implemente el contrato de WebSocket servicio tal como se especifica en el contrato de protocolo HTTP.

Esta flexibilidad le permite utilizar la WebSocket implementación que prefiera en diferentes lenguajes y marcos de programación, lo que garantiza la compatibilidad con las bases de código y los flujos de trabajo de desarrollo existentes.

Se utiliza WebSocket con Runtime AgentCore

En este tutorial de introducción, creará, probará e implementará una aplicación de agente que admita la transmisión bidireccional mediante el SDK de Python fundamental y la CLI para la implementación. AgentCore

Requisitos previos

Antes de empezar, asegúrese de que dispone de lo siguiente:

Paso 1: Configurar el proyecto e instalar las dependencias

Cree una carpeta de proyecto e instale los paquetes necesarios:

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

Actualice pip a la última versión:

pip install --upgrade pip

Instale los siguientes paquetes necesarios:

pip install bedrock-agentcore

Paso 2: Crea tu agente de streaming bidireccional

Cree un archivo fuente para su agente de streaming bidireccional con el nombre en clave. websocket_echo_agent.py Añada el código siguiente:

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")

Cree requirements.txt y añada lo siguiente:

bedrock-agentcore

Se incluye la dependencia websockets de la biblioteca de Python

Entender el código

Paso 3: Pruebe su agente de transmisión bidireccional localmente

Inicie su agente de streaming bidireccional

Abre una ventana de terminal e inicia tu agente de streaming bidireccional con el siguiente comando:

python websocket_echo_agent.py

Debería ver un resultado que indica que el servidor se está ejecutando en el puerto 8080.

Pruebe la conexión WebSocket

Cree un WebSocket cliente local llamadowebsocket_agent_client.py:

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())

Pruebe su agente de streaming bidireccional de forma local abriendo otra ventana de terminal y ejecutando el cliente:

python websocket_agent_client.py

Éxito: deberías ver una respuesta comoReceived: {"echo":{"inputText":"Hello WebSocket!"}}. En la ventana de la terminal en la que se ejecuta el agente, ingresa Ctrl+C para detenerlo.

Paso 4: Implemente su agente de streaming bidireccional en Runtime AgentCore

Instale las herramientas de despliegue

Instale la AgentCore CLI:

npm install -g @aws/agentcore

Verifique la instalación:

agentcore --help

Cree un proyecto e impleméntelo en AWS

Cree un nuevo proyecto para su agente de streaming bidireccional:

agentcore create

Despliega a tu agente:

agentcore deploy

Tras la implementación, recibirá un ARN de tiempo de ejecución del agente con el siguiente aspecto:

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

Guarde este ARN, ya que lo necesitará para invocar el agente desplegado.

Paso 5: invoque el agente de streaming bidireccional desplegado

Configure las variables de entorno

Configure las variables de entorno necesarias:

Métodos de autenticación

La acción InvokeAgentRuntimeWithWebSocketStream de la API establece una WebSocket conexión que admite la transmisión bidireccional entre el cliente y el agente. Puede autenticar WebSocket las conexiones mediante los siguientes métodos:

Conéctese mediante encabezados firmados SiGv4

El siguiente ejemplo muestra cómo establecer una WebSocket conexión y comunicarse con un agente en tiempo de ejecución mediante encabezados firmados por SigV4:

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())

Ejecute el cliente para probar el agente desplegado:

python websocket_agent_client_sigv4_headers.py

Éxito: deberías ver una respuesta como la siguiente:

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

Connect mediante una URL prefirmada (SiGv4 mediante parámetros de consulta)

El siguiente ejemplo muestra cómo crear una WebSocket URL con parámetros de consulta SigV4 y establecer una conexión:

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())

Ejecute el cliente para probar el agente desplegado:

python websocket_agent_client_sigv4_query_parameters.py

Éxito: deberías ver una respuesta como la siguiente:

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

Conéctate mediante OAuth

AgentCore Runtime admite la autenticación con el token OAuth Bearer para las conexiones. WebSocket Para usar la autenticación OAuth, debes configurar el tiempo de ejecución del agente con la autorización JWT, tal y como se describe en la sección de ejemplos de autorización entrante de JWT y acceso saliente de OAuth de Autenticar y autorizar con autenticación entrante y autenticación saliente.

Una vez que hayas completado la configuración de OAuth y hayas obtenido un token de portador siguiendo el paso 4: Usa el token de portador para invocar a tu agente en la guía de OAuth, puedes usar ese token para establecer WebSocket conexiones.

Cliente Python con OAuth

El siguiente ejemplo muestra cómo establecer una WebSocket conexión desde Python mediante OAuth:

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())

Ejecute el cliente para probar el agente desplegado:

python websocket_agent_client_oauth.py

Éxito: deberías ver una respuesta como la siguiente:

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

JavaScript Cliente de navegador con OAuth

La WebSocket API nativa del navegador no proporciona un método para configurar encabezados personalizados durante el apretón de manos. Para admitir la autenticación OAuth desde los navegadores, AgentCore Runtime acepta el token portador incrustado en el encabezado durante el Sec-WebSocket-Protocol apretón de manos. WebSocket

El token debe estar codificado en base64url y tener el prefijo del subprotocolo centinela, seguido del subprotocolo centinela. base64UrlBearerAuthorization. base64UrlBearerAuthorization

El siguiente ejemplo muestra cómo establecer una conexión desde el navegador mediante OAuth: WebSocket JavaScript

<!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>

Administración de sesiones

Si se proporciona un session_id (X-Amzn-Bedrock-AgentCore-Runtime-Session-Id) en la WebSocket conexión (como parámetro de consulta de URL o encabezado de solicitud), se enruta la conexión a una sesión de tiempo de ejecución aislada. El agente puede acceder al contexto de la conversación almacenado en esa sesión para implementar la continuidad de una conversación haciendo referencia a las interacciones anteriores. Los distintos identificadores de sesión acceden a distintos contextos aislados, lo que garantiza un aislamiento total entre los usuarios o las conversaciones.

Para obtener información sobre una gestión integral del ciclo de vida de las sesiones, que incluye el seguimiento, la limpieza y la gestión de errores, consulte Utilizar sesiones aisladas para los agentes.

Uso de sesiones con conexiones WebSocket

Para usar sesiones con WebSocket conexiones, genere un identificador de sesión único para cada usuario o conversación y páselo al establecer la conexión:

Al utilizar el mismo ID de sesión para WebSocket las conexiones relacionadas, se asegura de que se mantenga el contexto en la misma conversación, lo que permite al agente ofrecer respuestas coherentes basadas en interacciones anteriores.

Ciclo de vida de la sesión con WebSocket conexiones

En el WebSocket caso de las conexiones, el tiempo de espera de inactividad de la sesión se restablece cada vez que hay actividad de mensajes entre el cliente y el agente. Esto incluye cualquier intercambio de WebSocket mensajes, como el envío de datos del cliente al agente, la recepción de respuestas del agente al cliente o los WebSocket ping/pong marcos. Esto significa que WebSocket las conversaciones activas mantendrán activa la sesión mientras los mensajes sigan fluyendo, lo que evitará que la sesión se cierre prematuramente durante las interacciones en curso.

Para obtener más información sobre cómo configurar los ajustes del ciclo de vida, consulte Configurar los ajustes AgentCore del ciclo de vida de Amazon Bedrock. Para obtener un control más directo del ciclo de vida de la sesión mediante el estado del agente, consulte Gestión del ciclo de vida de las sesiones en tiempo de ejecución.

Detenga la sesión en ejecución

Para detener una sesión en ejecución antes de la configuración IdleRuntimeSessionTimeout (el valor predeterminado es de 15 minutos), consulte Detener una sesión en ejecución.

Observabilidad

Amazon Bedrock AgentCore Observability le ayuda a rastrear, depurar y supervisar los agentes que aloja en Amazon Bedrock Runtime. AgentCore En primer lugar, active la búsqueda de CloudWatch transacciones siguiendo las instrucciones de Habilitar la observabilidad en tiempo de AgentCore ejecución de Amazon Bedrock. Para observar a su agente, consulte Ver datos de observabilidad de sus agentes de Amazon Bedrock AgentCore .

En el WebSocket caso de las conexiones, un rastreo representa la sesión de conexión completa y no los intercambios de mensajes individuales.

Encabezados personalizados

Los encabezados personalizados le permiten pasar la información contextual de la aplicación directamente al código de agente en la WebSocket conexión inicial. Para obtener información completa sobre el soporte, la configuración y las limitaciones de los encabezados personalizados, consulte Transferir encabezados personalizados a Amazon Bedrock Runtime AgentCore .

Además, los encabezados con el prefijo se X-Amzn-Bedrock-AgentCore-Runtime-Custom- pueden pasar como parámetros de consulta de URL en las conexiones. WebSocket

Por ejemplo, puede pasar encabezados personalizados como parámetros de consulta en la URL: WebSocket

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

El contenedor de aplicaciones del agente los recibirá como encabezados:

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

Apéndice

Consideraciones de seguridad

Resolución de problemas

Problemas comunes WebSocket-specific

Los siguientes son problemas comunes que pueden surgir:

Gestión de errores

WebSocket las conexiones utilizan códigos de cierre estándar para la comunicación de errores. Los códigos de cierre más comunes incluyen:

WebSocket frente a otros protocolos

Cuándo usar WebSocket:

Cuándo usar HTTP:

Ejemplos adicionales de introducción

Para ver ejemplos adicionales que utilizan la transmisión WebSocket bidireccional con AgentCore Runtime, consulte los ejemplos de transmisión WebSocket GitHub bidireccional: