Lab práctico · Semana 5: IA agéntica, herramientas y Amazon Bedrock AgentCore

Agente con Strands Agents y herramientas propias

Qué vas a construir

Un agente de atención al cliente de una tienda online, escrito en Python con Strands Agents y el modelo Amazon Nova Lite en Amazon Bedrock. El agente tendrá tres herramientas propias:

• consultar_pedido: lee el estado de un pedido.
• calcular_envio: calcula un coste con validación de parámetros y errores útiles.
• cancelar_pedido: una acción con efectos que solo se ejecuta tras aprobación humana (human-in-the-loop) mediante un hook e interrupciones.

Además pondrás condiciones de parada (límite de turnos), verás cómo el agente mantiene la memoria a corto plazo entre preguntas y leerás sus métricas (tokens y uso de herramientas).

flowchart LR
    U["Tú (CloudShell)"] --> A["Agente Strands"]
    A -->|"Converse"| M["Amazon Nova Lite (perfil eu.)"]
    A --> T1["consultar_pedido"]
    A --> T2["calcular_envio"]
    A --> H{"Hook: ¿herramienta sensible?"}
    H -->|"sí"| I["Interrupción: aprobación humana"]
    I --> T3["cancelar_pedido"]

Antes de empezar

• Cuenta de AWS con permiso para bedrock:InvokeModel (un usuario o rol administrador de tu cuenta de estudio vale). Este lab no usa AgentCore, así que funciona también en Free plan.
• Región: eu-central-1 (Fráncfort). Usarás el perfil de inferencia geográfico eu.amazon.nova-lite-v1:0: el procesamiento se queda en regiones de la UE.
• Acceso al modelo: los modelos de Amazon están habilitados por defecto en Bedrock; no hay que pedir acceso.
• Coste esperado: el lab completo consume del orden de decenas de miles de tokens. A precios de Nova Lite en Fráncfort (estimación a 1/10/2026: 0,078 USD por millón de tokens de entrada y 0,312 USD por millón de salida) son menos de 0,01 USD; aunque repitas los experimentos varias veces no deberías pasar de 0,05 USD. Precios en Amazon Bedrock pricing.

Paso 1: prepara CloudShell y un entorno Python

Abre la consola en Fráncfort y lanza CloudShell (icono de terminal en la barra superior). Strands necesita Python 3.10 o superior; comprueba qué versión trae CloudShell:

python3 --version
aws configure get region || echo "$AWS_REGION"

Para no depender de la versión del sistema, usa uv (gestor de entornos de Python) e instala Python 3.12 en tu directorio personal, que es lo único que persiste entre sesiones de CloudShell:

pip3 install --user --quiet uv
export PATH="$HOME/.local/bin:$PATH"
mkdir -p ~/lab09 && cd ~/lab09
uv venv --python 3.12 .venv
source .venv/bin/activate
uv pip install "strands-agents==1.57.1"
python -c "import strands, sys; print('Strands OK con Python', sys.version.split()[0])"

Fijamos la versión 1.57.1 (la vigente al escribir este lab) para que el código se comporte igual que aquí.

Paso 2: comprueba el modelo con la AWS CLI

Antes de meter un framework, verifica que tu cuenta puede invocar el modelo con la API Converse:

aws bedrock-runtime converse \
  --region eu-central-1 \
  --model-id eu.amazon.nova-lite-v1:0 \
  --messages '[{"role":"user","content":[{"text":"Di hola en una frase."}]}]' \
  --inference-config '{"maxTokens":50,"temperature":0.2}' \
  --query 'output.message.content[0].text' --output text

Si ves un saludo, todo está listo. Si ves AccessDeniedException, revisa los permisos IAM de tu identidad.

Paso 3: escribe el agente

Crea agente.py. Léelo entero antes de ejecutarlo: cada bloque corresponde a una idea del módulo 05.

cat > agente.py <<'EOF'
import json
import os
from typing import Any

from strands import Agent, tool
from strands.agent.conversation_manager import SlidingWindowConversationManager
from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry
from strands.models import BedrockModel

# "Base de datos" de juguete: en un caso real sería DynamoDB o una API interna
PEDIDOS = {
    "P-1001": {"estado": "enviado", "peso_kg": 2.5, "destino": "Madrid"},
    "P-1002": {"estado": "preparando", "peso_kg": 12.0, "destino": "Bilbao"},
}
TARIFAS = {"estandar": {"base": 4.0, "por_kg": 0.6}, "urgente": {"base": 9.0, "por_kg": 1.1}}


@tool
def consultar_pedido(pedido_id: str) -> dict:
    """Devuelve el estado, el peso y el destino de un pedido.

    Args:
        pedido_id: Identificador del pedido con el formato P-NNNN, por ejemplo P-1001.
    """
    pedido = PEDIDOS.get(pedido_id.strip().upper())
    if pedido is None:
        # Error controlado: el modelo lo recibe como resultado de herramienta con status "error"
        return {"status": "error", "content": [{"text": f"No existe el pedido {pedido_id}"}]}
    return pedido  # Strands lo serializa a JSON como resultado correcto


@tool
def calcular_envio(peso_kg: float, modalidad: str = "estandar") -> dict:
    """Calcula el coste de envío en euros de un paquete.

    Args:
        peso_kg: Peso del paquete en kilogramos. Debe ser mayor que 0 y como máximo 30.
        modalidad: Modalidad de envío: "estandar" o "urgente".
    """
    if not 0 < peso_kg <= 30:
        return {"status": "error", "content": [{"text": "peso_kg debe estar entre 0 y 30"}]}
    if modalidad not in TARIFAS:
        return {"status": "error", "content": [{"text": "modalidad debe ser estandar o urgente"}]}
    t = TARIFAS[modalidad]
    coste = round(t["base"] + t["por_kg"] * peso_kg, 2)
    return {"coste_eur": coste, "modalidad": modalidad}


@tool
def cancelar_pedido(pedido_id: str) -> str:
    """Cancela un pedido que todavía no se ha enviado.

    Args:
        pedido_id: Identificador del pedido con el formato P-NNNN.
    """
    pedido = PEDIDOS.get(pedido_id.strip().upper())
    if pedido is None or pedido["estado"] == "enviado":
        return f"No se puede cancelar {pedido_id}"
    pedido["estado"] = "cancelado"
    return f"Pedido {pedido_id} cancelado"


class AprobacionHumana(HookProvider):
    """Human-in-the-loop: pide confirmación antes de ejecutar herramientas con efectos."""

    SENSIBLES = {"cancelar_pedido"}

    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        registry.add_callback(BeforeToolCallEvent, self.aprobar)

    def aprobar(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] not in self.SENSIBLES:
            return
        respuesta = event.interrupt("lab09-aprobacion", reason={"entrada": event.tool_use["input"]})
        if str(respuesta).lower() != "s":
            event.cancel_tool = "Un humano ha rechazado la acción"


modelo = BedrockModel(
    model_id=os.environ.get("MODEL_ID", "eu.amazon.nova-lite-v1:0"),
    region_name=os.environ.get("AWS_REGION", "eu-central-1"),
    temperature=0.2,
    max_tokens=800,
)

agente = Agent(
    model=modelo,
    system_prompt=(
        "Eres el asistente de atención al cliente de una tienda online española. "
        "Usa las herramientas para consultar datos; no inventes estados ni precios. "
        "Responde en español de España, en dos o tres frases."
    ),
    tools=[consultar_pedido, calcular_envio, cancelar_pedido],
    conversation_manager=SlidingWindowConversationManager(window_size=20),
    hooks=[AprobacionHumana()],
    callback_handler=None,
)


def preguntar(texto: str) -> None:
    resultado = agente(texto, limits={"turns": 6})
    # Bucle de interrupciones: el agente se detiene y espera la decisión humana
    while resultado.stop_reason == "interrupt":
        respuestas = []
        for interrupcion in resultado.interrupts:
            decision = input(f"¿Apruebas {interrupcion.reason}? (s/N): ")
            respuestas.append({"interruptResponse": {"interruptId": interrupcion.id, "response": decision}})
        resultado = agente(respuestas)
    print(f"\n[{resultado.stop_reason}] {resultado}")
    uso = resultado.metrics.accumulated_usage
    print(f"Tokens acumulados: entrada={uso['inputTokens']} salida={uso['outputTokens']}")


if __name__ == "__main__":
    print(json.dumps(calcular_envio.tool_spec, ensure_ascii=False, indent=2))
    preguntar("¿Dónde está el pedido P-1001?")
    preguntar("¿Cuánto costaría enviarlo por urgente?")
    preguntar("Cancela el pedido P-1002, por favor.")
EOF

Qué hace cada parte:

Paso 4: ejecuta el agente

python agente.py

Lo que debes observar:

1. Primero se imprime el esquema JSON de calcular_envio. Es exactamente lo que recibe el modelo como toolSpec: fíjate en required, en los tipos y en que la descripción sale del docstring.
2. «¿Dónde está el pedido P-1001?» → el agente llama a consultar_pedido y responde que está enviado a Madrid.
3. «¿Cuánto costaría enviarlo por urgente?» → no repites el pedido: el agente lo sabe por la memoria a corto plazo (el historial de la conversación). Tiene que encadenar razonamiento y herramienta: usa el peso de 2,5 kg y llama a calcular_envio(peso_kg=2.5, modalidad="urgente"). Resultado esperado: 11,75 €.
4. «Cancela el pedido P-1002» → el hook detiene el agente y te pregunta ¿Apruebas {'entrada': {'pedido_id': 'P-1002'}}? (s/N):. Responde s: el agente continúa, ejecuta la herramienta y confirma la cancelación.

Paso 5: experimenta con los límites

Crea un segundo script que reutiliza el agente y fuerza situaciones límite:

cat > experimentos.py <<'EOF'
import json
import agente as A

# 1) Rechazo humano: responde N cuando te pregunte
A.preguntar("Cancela el pedido P-1002.")

# 2) Error de validación: el modelo recibe el error y debe explicarlo o corregirse
A.preguntar("¿Cuánto cuesta enviar un paquete de 50 kg por estándar?")

# 3) Condición de parada: un solo turno no basta para usar una herramienta y responder
r = A.agente("¿Cuánto cuesta enviar el pedido P-1002 por urgente?", limits={"turns": 1})
print("stop_reason con 1 turno:", r.stop_reason)

# 4) Métricas de herramientas (base de la observabilidad de agentes)
resumen = A.agente.event_loop_metrics.get_summary()
print(json.dumps({k: v["execution_stats"] for k, v in resumen["tool_usage"].items()}, indent=2))
EOF
python experimentos.py

Resultados esperados:

1. Con N, el hook pone cancel_tool y el modelo recibe un resultado de error («Un humano ha rechazado la acción»): el pedido no se cancela.
2. calcular_envio devuelve el error de rango; el agente te dice que el máximo son 30 kg.
3. Con turns: 1, el agente se detiene con stop_reason igual a limit_turns: sin ese control, un agente puede seguir llamando herramientas indefinidamente.
4. Verás por herramienta el número de llamadas, éxitos, errores y tiempos (call_count, success_count, error_count, average_time…). Son las métricas que, en producción, enviarías a CloudWatch para detectar herramientas lentas o que fallan.

Paso 6 (opcional): de un agente a varios

Añade un especialista como herramienta de un supervisor (patrón agents as tools):

cat > multiagente.py <<'EOF'
from strands import Agent
import agente as A

logistica = Agent(
    name="logistica",
    description="Responde dudas sobre envíos y costes de envío usando herramientas.",
    model=A.modelo,
    system_prompt="Eres experto en logística. Usa las herramientas y responde en una frase.",
    tools=[A.consultar_pedido, A.calcular_envio],
    callback_handler=None,
)

supervisor = Agent(
    model=A.modelo,
    system_prompt="Eres el supervisor. Delega las dudas de envíos en la herramienta logistica.",
    tools=[logistica],
    callback_handler=None,
)

r = supervisor("¿Cuánto costaría enviar por urgente el pedido P-1001?", limits={"turns": 4})
print(r.stop_reason, str(r))
print("Herramientas del supervisor:", supervisor.tool_names)
EOF
python multiagente.py

Verás que el supervisor solo tiene una herramienta, logistica, que es a su vez un agente con sus propias herramientas. Cada nivel añade llamadas al modelo: mira los tokens y piensa si compensa.

Comprueba que funciona

• agente.py responde a las tres preguntas y la segunda usa el contexto de la primera.
• La cancelación solo ocurre si respondes s.
• El paquete de 50 kg provoca un error de validación explicado por el agente.
• Con turns: 1 el stop_reason es limit_turns.
• Ves estadísticas por herramienta (llamadas, errores, tiempos).

Limpieza

Este lab no crea recursos en AWS: solo has pagado tokens. Libera espacio en CloudShell (si vas a hacer el lab 10 ahora, puedes conservar la carpeta y borrarla después):

deactivate 2>/dev/null
rm -rf ~/lab09

Si quieres comprobar el consumo, en Billing and Cost Management → Cost Explorer filtra por el servicio Amazon Bedrock (los costes aparecen con unas horas de retraso).

Preguntas para pensar como arquitecto