Comience con la transmisión bidireccional mediante WebSocket

Amazon Bedrock AgentCore Runtime le permite implementar agentes compatibles con la WebSocket transmisión para una comunicación bidireccional en tiempo real. Esta guía le explica cómo crear, probar e implementar su primer agente de streaming bidireccional con. 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 streaming bidireccionales y persistentes entre clientes y agentes. AgentCore Runtime espera que los contenedores implementen WebSocket puntos finales en el puerto 8080 de la /ws ruta, lo que se alinea con las prácticas estándar WebSocket de los servidores.

AgentCore El WebSocket soporte de Runtime proporciona las mismas capacidades sin servidor, aislamiento de sesión, 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 que utilizan 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 utilicen 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

utilizar uno de los métodos de autenticación admitidos (encabezados de SIGv4, URL prefirmada de SIGv4 u OAuth 2.0) y que la aplicación del agente implemente el contrato de WebSocket servicio tal como se especifica en el contrato de protocolo HTTP. 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.

Utilización 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 de Bedrock-agentcore y la CLI para la implementación. AgentCore

Requisitos previos

Antes de empezar, asegúrate de tener:

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

Actualiza pip a la última versión:

pip install --upgrade pip

Instala los siguientes paquetes necesarios:

pip install bedrock-agentcore

Paso 2: Cree su agente de streaming bidireccional

Crea un archivo fuente para tu agente de streaming bidireccional con el nombre en código. 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

Comprender el código

Paso 3: Pruebe su agente de streaming bidireccional de forma local

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 ejecuta 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 del terminal en la que se ejecuta el agente, escribe Ctrl+C para detener el agente.

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

Instala las herramientas de implementación

Instale la AgentCore CLI:

npm install -g @aws/agentcore

Verifique la instalación:

agentcore --version

Para ver los comandos y las opciones disponibles, consulte la referencia de la AgentCore CLI.

Cree un proyecto e impleméntelo en AWS

Cree un nuevo proyecto para su agente de streaming bidireccional:

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

Despliegue su agente:

agentcore deploy

El AgentCore proyecto hace referencia al directorio agentcore-runtime-quickstart-websocket fuente existente.

Tras la implementación, recibirá un ARN en 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 al agente desplegado.

Paso 5: Invoca tu agente de streaming bidireccional implementado

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 con SigV4

El siguiente ejemplo muestra cómo establecer una WebSocket conexión y comunicarse con el tiempo de ejecución de un agente mediante encabezados firmados de 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ía ver una respuesta como:

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

Conéctese 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 de 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ía ver una respuesta como:

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

Conéctese mediante OAuth

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

Cuando hayas completado la configuración de OAuth y 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 de 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ía ver una respuesta como:

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

JavaScript Cliente de navegador con OAuth

La WebSocket API nativa del navegador no proporciona un método para establecer encabezados personalizados durante el apretón de manos. Para admitir la autenticación OAuth desde los navegadores, AgentCore Runtime acepta el token del 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 «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

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

Para obtener información completa sobre la gestión 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 usar el mismo ID de sesión para WebSocket las conexiones relacionadas, te aseguras de que se mantenga el contexto en la misma conversación, lo que permite al agente ofrecer respuestas coherentes que se basen 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 de un cliente a otro, la recepción de respuestas de un agente a otro o la recepción de WebSocket ping/pong marcos. Esto significa que WebSocket las conversaciones activas mantendrán la sesión activa mientras los mensajes sigan fluyendo, lo que evita la finalización prematura de la sesión durante las interacciones en curso.

Para obtener más información sobre la configuración 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 a través del estado de los agentes, consulte Gestión del ciclo de vida de las sesiones en tiempo de ejecución.

Detenga la sesión de ejecución

Para detener una sesión en ejecución antes que la configurable 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, habilite CloudWatch la búsqueda de transacciones siguiendo las instrucciones que aparecen en Habilitar la observabilidad en tiempo de ejecución de Amazon Bedrock. AgentCore Para observar a su agente, consulte Ver los datos de observabilidad de sus agentes de Amazon Bedrock. AgentCore

En el WebSocket caso de las conexiones, un seguimiento representa la sesión de conexión completa en lugar de los intercambios de mensajes individuales.

Encabezados personalizados

Los encabezados personalizados 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 la compatibilidad, la configuración y las limitaciones de los encabezados personalizados, consulte Transferir los encabezados personalizados a Amazon Bedrock AgentCore Runtime.

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, puedes 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 la aplicación del agente recibirá los siguientes 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 los errores aparecen en dos fases, según el momento en que se produzcan.

Establecimiento de la conexión (antes de la WebSocket actualización)

La apertura de la conexión es una solicitud HTTP estándar. El código de estado HTTP refleja la excepción y el encabezado de la x-amzn-ErrorType respuesta incluye el nombre de la excepción. El servicio puede devolver cualquiera de los siguientes errores antes de establecer la WebSocket conexión.

Conexión activa (después de la WebSocket actualización)

Una vez establecida, los errores se comunican con códigos de WebSocket cierre estándar en lugar de con códigos de estado HTTP. WebSocket Los códigos de cierre más comunes incluyen:

WebSocket frente a otros protocolos

Cuándo usar WebSocket:

Cuándo usar HTTP:

Otros ejemplos de cómo empezar

Para ver más ejemplos sobre el uso de la transmisión WebSocket bidireccional con AgentCore Runtime, consulta los ejemplos de transmisión WebSocket GitHub bidireccional: