Desplegando un agente LangGraph en AWS Bedrock AgentCore, paso a paso

Muchos agentes de LangGraph funcionan bien en una laptop, pero se detienen en la fase de puesta en producción. AWS Bedrock AgentCore aborda esta limitación mediante un alojamiento sin servidores que se escala automáticamente y no requiere un rediseño. Las secciones siguientes tratan sobre cómo adaptar un agente existente para AgentCore, generar un proyecto, ejecutarlo localmente, enviarlo a AWS y llamarlo una vez esté en funcionamiento.

Qué es AgentCore

AgentCore es un entorno de ejecución alojado por AWS donde los agentes se despliegan, escalan y ejecutan sin que nadie tenga que gestionar servidores. Frameworks como LangGraph, Strands, CrewAI y LlamaIndex funcionan perfectamente, al igual que los modelos de OpenAI, Google y Anthropic, por lo que el código del agente permanece sin cambios. La isolación de sesiones, la memoria persistente, la capacidad de observabilidad y la gestión de identidades vienen integradas de forma nativa.

Todo está controlado por la CLI de AgentCore. Se trata de un programa de Node.js que genera el esqueleto del proyecto, sirve al agente localmente y lo envía a AWS con CDK ejecutándose en segundo plano.

Requisitos previos

• Python 3.12 o posterior
• uv o pip para gestionar paquetes de Python
• Node.js 20 o posterior, ya que la CLI se distribuye a través de npm
• AWS CDK instalado globalmente con npm install -g aws-cdk
• Una cuenta de AWS con credenciales configuradas localmente mediante aws configure
• Código LangGraph existente que genere un StateGraph compilado

Paso 1: instalar la CLI

Una sola herramienta de línea de comandos controla todo: la creación del proyecto, la ejecución de la aplicación localmente y el despliegue en AWS.

npm install -g @aws/agentcore

Confirme que la instalación funcionó:

agentcore --help

Paso 2: agregar dependencias

Hay dos paquetes que deben incorporarse a las dependencias del agente. bedrock-agentcore es el SDK de Python que proporciona la clase wrapper BedrockAgentCoreApp, mientras que aws-opentelemetry-distro es la versión de AWS de la instrumentación de Python de OpenTelemetry.

# pyproject.toml
[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "aws-opentelemetry-distro==0.17.0",
 "bedrock-agentcore>=1.6.3",
    "boto3>=1.42.0",
    "langgraph>=1.1.0",
    "langchain-core>=1.2.0",
    # ... your other existing dependencies
]

Instálalo con el flujo de trabajo que sea habitual para el proyecto:

uv sync
# or: pip install -e .

Paso 3: escribir el punto de entrada en main.py

AgentCore espera un archivo de Python como punto de entrada, que contenga una instancia de BedrockAgentCoreApp y una función decorada con @app.entrypoint. La lógica de LangGraph apenas cambia; simplemente se envuelve en un pequeño punto de entrada de tiempo de ejecución.

# main.py
from langchain_core.messages import HumanMessage
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from graphs.my_agent_graph import build_my_agent  # your existing graph builder

# 1. Instantiate the AgentCore app and logger
app = BedrockAgentCoreApp()
log = app.logger

# 2. Build your graph at module load time (startup)
#    AgentCore initialises the module once, then handles concurrent invocations.
#    Any failure here will prevent a broken agent from going live.
def create_agent():
    log.info("Initialising agent...")
    graph = build_my_agent()   # returns your compiled LangGraph StateGraph
    log.info("Agent ready")
    return graph
try:
    graph = create_agent()
except Exception as e:
    log.error(f"Critical failure during agent initialisation: {e}")
    raise  # fail fast - don't let a broken agent start serving requests

# 3. Async helper that drives the LangGraph streaming loop
async def run_agent(user_input: str, session_id: str = "default-session") -> str:
    responses = []
    config = {"configurable": {"thread_id": session_id}}
    async for chunk in graph.astream(
        {"messages": [HumanMessage(content=user_input)]},
        config=config,
        stream_mode="values",
    ):
        messages = chunk.get("messages", [])
        if messages:
            last = messages[-1]
            if getattr(last, "type", None) == "ai":
                responses.append(last)
    if not responses:
        return "No response"
    content = getattr(responses[-1], "content", "")
    return content if isinstance(content, str) else str(content)

# 4. The entrypoint - this is what AgentCore calls on every invocation
@app.entrypoint
async def invoke(payload, context):
    try:
        log.info("Invoke received")
        user_input = payload.get("prompt", "")
        session_id = payload.get("session_id", "default-session")
        if not user_input.strip():
            return {"error": "Prompt cannot be empty"}
        response = await run_agent(user_input, session_id)
        return {"response": response}
    except Exception as e:
        log.error(f"Error: {e}")
        return {"error": str(e)}

# 5. Run the app (only executed when running locally using agentcore dev)
if __name__ == "__main__":
    app.run()

Lo importante en ese archivo

• BedrockAgentCoreApp() proporciona al entorno de ejecución su estructura básica: un servidor HTTP, rutas de verificación de estado y registros estructurados, todo ya configurado.
• El grafo se construye a nivel de módulo, una sola vez al importarlo y no en cada llamada. Por lo tanto, cualquier problema al iniciar se detecta de inmediato, y una aplicación defectuosa nunca recibe tráfico.
• @app.entrypoint marca el manejador para las llamadas entrantes. Se le proporciona payload, el cuerpo JSON decodificado, junto con context para la solicitud de AgentCore. Lo que devuelve debe ser un diccionario simple que se pueda serializar a JSON. Una cadena simple o una clase personalizada no sirven.
• El session_id entrante se convierte en thread_id de LangGraph, de modo que MemorySaver mantiene una memoria separada por sesión.

Paso 4: crear el proyecto

A continuación viene el diseño del proyecto. Ejecute el asistente de inicialización desde una carpeta raíz limpia y deje que cree el esqueleto:

agentcore create

El asistente solicita detalles como el nombre del proyecto, el lenguaje, la versión de Python y el archivo de punto de entrada. El punto de entrada y la versión de Python deben coincidir con la aplicación. Posteriormente, aparece un archivo agentcore.json en la carpeta agentcore, que indica a AgentCore dónde se encuentra el código y cómo ejecutarlo.

{
  "name": "MyAgentOnAgentcore",
    "runtimes": [
    {
      "name": "my-agent",
      "build": "CodeZip",
      "entrypoint": "main.py",
      "codeLocation": "app/my-agent/",
      "runtimeVersion": "PYTHON_3_12",
      "networkMode": "PUBLIC",
      "protocol": "HTTP",
	  "envVars": []
	}
  ]
}

Paso 5: coloque el código donde lo busca AgentCore

Mueva o copie el código del agente al directorio especificado por codeLocation en agentcore.json. Es fácil cometer errores, y una ubicación incorrecta hace que el despliegue sea innecesariamente complicado. Un error común es colocar el código directamente en la carpeta agentcore en sí, lo cual puede costar mucho tiempo. El layout debería verse así:

MyAgentOnAgentcore/
├──agentcore
	└── agentcore.json
├──app/
	└──my-agent/          			 ← codeLocation
		├── main.py                  ← entrypoint (matches agentcore.json)
		├── pyproject.toml           ← or requirements.txt
		├── src/
		│   └── my_agent_graph.py    ← your LangGraph graph builder
		│   └── tools.py             ← your tools, utilities, etc.
		└── .env					 ← your configuration variables

main.py debe encontrarse en el nivel más alto de codeLocation y coincidir exactamente con el punto de entrada en agentcore.json.

Variables de entorno

AgentCore puede pasar variables de entorno al contenedor de ejecución. A nivel local, un archivo .env es la opción más sencilla. En entornos de producción, los valores pueden almacenarse en agentcore.json dentro de envVars, aunque también funciona un archivo .env. Los datos confidenciales como contraseñas y claves API es mejor guardarlos en AWS Secrets Manager y cargarlos en tiempo de ejecución; en ese caso, el rol de ejecución de AgentCore necesita permisos para leerlos. Tenga en cuenta que envVars es un array de objetos JSON, cada uno con los campos name y value.

Paso 6: validar

Antes de ejecutar cualquier cosa, permita que la CLI verifique la configuración:

agentcore validate

Los errores simples, como la falta de un archivo de punto de entrada o una ruta incorrecta en la configuración, se detectan justo aquí.

Paso 7: ejecutar y probar localmente

Inicie el entorno de ejecución local con:

agentcore dev

Se inicia un servidor HTTP local que imita con precisión el entorno de producción. Cuando todo está configurado correctamente, el resultado muestra que la aplicación está lista y escuchando en un puerto local.

Convocar al servidor local

Se puede realizar la llamada a través de la CLI:

agentcore invoke "Summarise last month sales"

o mediante curl:

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Summarise last month sales", "session_id": "test-001"}'

Una buena respuesta es JSON con la respuesta del agente dentro:

{"response": "Last month, total sales were $1.2M across 3 regions..."}

Paso 8: despliegue en AWS

Cuando las pruebas locales den buenos resultados, realice el despliegue en AgentCore Runtime:

agentcore deploy

Detrás de escena, la CLI empaqueta el código, proporciona un bucket de S3 para el despliegue directo del código, crea un rol de ejecución IAM y despliega el AgentCore Runtime con CDK. El primer despliegue tarda unos minutos, mientras que los posteriores suelen ser más rápidos.

Una vez finalizado, verifique el estado del runtime. La salida muestra el ARN del runtime y la URL HTTP para invocar al agente desplegado.

agentcore status

Paso 9: invocar el agente desplegado

La CLI es la forma más sencilla de probar la versión desplegada:

agentcore invoke --prompt "Who are the top 5 customers by revenue?" --session-id "session-id-with-length-greater-than-or-equal-33"

Para realizar llamadas mediante HTTP es necesario firmar la solicitud con AWS SigV4:

curl -X POST "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn-of-MyAgentOnAgentcore-xxxx/invocations" \
  --user "$AWS_ACCESS_KEY_ID:$AWS_SECRET_ACCESS_KEY" \
  --aws-sigv4 "aws:amz:us-east-1:bedrock-agentcore" \
  -H "Content-Type: application/json" \
  -H "x-amzn-bedrock-agentcore-runtime-session-id: session-id-with-length-greater-than-or-equal-33" \
  -d '{"prompt": "Who are the top 5 customers by revenue?"}'

Utilice la URL HTTP de agentcore status.

Para llamadas desde otro código en Python, el SDK suele ser la opción más adecuada. Almacene el ARN en tiempo de ejecución en una variable de entorno, proporcione el comando y envíe nuevamente el ID de la sesión anterior en cada llamada posterior.

# invoke_agent.py
import json
import boto3
import os
from dotenv import load_dotenv

load_dotenv()

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
runtime_arn = os.getenv("AGENTCORE_RUNTIME_ARN")

def invoke(prompt: str, session_id: str) -> str:
    payload = json.dumps({"prompt": prompt, "session_id": session_id})
    response = client.invoke_agent_runtime(
        agentRuntimeArn=runtime_arn,
        payload=payload,
        contentType="application/json"
    )
    body = json.loads(response["response"].read().decode())
    return body["response"]

if __name__ == "__main__":
    session_id = ""
    print("Ready. Type 'exit' to quit.")

    while True:
        prompt = input("Your question: ").strip()
        if prompt == "exit":
            break

        if not prompt:
            continue

        params = {
            "agentRuntimeArn": runtime_arn,
            "payload": json.dumps({"prompt": prompt}),
            "contentType": "application/json"
        }
        if session_id:
            params["runtimeSessionId"] = session_id

        try:
            response = client.invoke_agent_runtime(**params)
            body = json.loads(response["response"].read().decode())
            session_id = response.get("runtimeSessionId", session_id)
            print(f"Agent: {body['response']}\n")
        except Exception as e:
            print(f"Error: {e}")

Establezca AGENTCORE_RUNTIME_ARN en el archivo .env:

AGENTCORE_RUNTIME_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/MyAgentOnAgentcore-xxxx

Mantener el estado de la sesión entre llamadas

Observe runtimeSessionId en el ejemplo de boto3. AgentCore devuelve uno con cada respuesta. Al enviarlo de vuelta en la siguiente solicitud, las consultas posteriores se enrutan al mismo contexto de sesión, y combinado con MemorySaver de LangGraph, el agente conserva toda la memoria de la conversación a través de llamadas HTTP separadas.

Problemas comunes

• Los fallos de inicio son intencionados. Cuando create_agent() lanza un error al momento de la importación, AgentCore no iniciará el entorno de ejecución. Ese comportamiento evita que un agente mal configurado, como uno sin credenciales o con una URL de base de datos incorrecta, se active silenciosamente.
• Es necesario agregar permisos. El rol de ejecución AgentCore requiere acceso explícito a cualquier servicio de AWS que utilice el agente, incluidos Secrets Manager y S3. Añada esos permisos después del primer despliegue.
• Las versiones deben coincidir. La versión de Python que se declare en pyproject.toml debe coincidir con la configuración de tiempo de ejecución en agentcore.json.
• Devuelva diccionarios simples. Si el punto de entrada devuelve una cadena o un objeto personalizado, el entorno de tiempo de ejecución lo rechazará.

Qué cambia realmente

Para pasar de un agente local de LangGraph a AgentCore solo se necesitan unos pocos cambios: instalar dos paquetes adicionales, envolver el grafo en BedrockAgentCoreApp, crear una estructura básica para el proyecto, probarlo localmente y luego desplegarlo y llamarlo. El grafo, las herramientas, las llamadas al modelo y la configuración de memoria permanecen en su mayor parte sin cambios. AgentCore se encarga del lado de la infraestructura, mientras que LangGraph sigue siendo responsable de la lógica del agente.