Descripción General del Proyecto

En esta lección capstone, construirás un agente de IA completo que realiza cuatro tareas reales en nombre de un usuario autenticado: verificar la identidad mediante autenticación, leer y escribir registros en una base de datos, llamar a APIs externas para obtener o enviar datos, e imponer control de acceso basado en roles (RBAC) para que los usuarios solo vean lo que tienen permitido ver. Este es el punto de integración donde se conecta cada concepto previo del curso.

El agente que construirás sigue un patrón de petición-respuesta: un usuario envía una solicitud, el agente verifica quién es, decide qué tiene permiso para hacer, luego consulta la base de datos, llama a una API, o ambas, y devuelve un resultado. Usaremos Python para la lógica del agente, una base de datos SQL para la persistencia, una API externa simulada para los datos externos y un mecanismo simple de JWT (JSON Web Token) para la autenticación.

Arquitectura y Configuración

Antes de escribir código, vamos a esbozar las cuatro piezas móviles y cómo se conectan. El usuario envía una solicitud con un token JWT. El agente valida el token, extrae el ID de usuario y el rol, y luego enruta la solicitud a un manejador de base de datos o a un llamador de API según la intención del usuario. El manejador de base de datos usa consultas SQL parametrizadas para prevenir ataques de inyección. El llamador de API adjunta el mismo JWT a las solicitudes salientes para que el servicio externo pueda verificar al llamante de forma independiente.

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: int
    role: str
    token: str

Este objeto UserContext viaja con cada llamada interna. Una vez que una solicitud entra al agente, el contexto se estampa en ella y cada función downstream recibe el mismo contexto, por lo que las comprobaciones de autorización ocurren en cada paso, no solo en el punto de entrada.

Implementando la Autenticación

La autenticación consiste en demostrar quién es el usuario. Usaremos un token estilo JWT simple: una cadena codificada en base64 que contiene el ID de usuario, el rol y una marca de tiempo de expiración, firmada con un secreto compartido. En producción usarías una librería como PyJWT, pero aquí lo mantenemos minimalista para que puedas ver la mecánica.

import base64, json, hmac, hashlib, time

SECRET = "change-me-in-production"

def create_token(user_id: int, role: str) -> str:
    payload = {"uid": user_id, "role": role, "exp": int(time.time()) + 3600}
    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
    sig = hmac.new(SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
    return f"{body}.{sig}"

def verify_token(token: str) -> UserContext:
    body, sig = token.split(".")
    expected = hmac.new(SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        raise PermissionError("Invalid token signature")
    payload = json.loads(base64.urlsafe_b64decode(body))
    if payload["exp"] < int(time.time()):
        raise PermissionError("Token expired")
    return UserContext(user_id=payload["uid"], role=payload["role"], token=token)

La función verify_token es el guardián. Cada endpoint protegido la llama primero. Si el token está manipulado, expirado o ausente, la solicitud se rechaza antes de que se ejecute cualquier código de base de datos o API.

Consultando la Base de Datos de Forma Segura

Con la autenticación lista, el siguiente paso es permitir que el agente lea registros de usuario. La regla crítica aquí es: nunca construyas SQL por concatenación de cadenas. Usa siempre consultas parametrizadas para que la entrada del usuario se trate como dato, no como SQL ejecutable.

import sqlite3

def get_user_orders(ctx: UserContext, target_user_id: int) -> list:
    conn = sqlite3.connect("app.db")
    cur = conn.cursor()
    cur.execute("SELECT id, product, total FROM orders WHERE user_id = ?", (target_user_id,))
    rows = cur.fetchall()
    conn.close()
    return rows

Observa cómo el ID de usuario se pasa como parámetro en una tupla, no se interpola dentro de la cadena. Este único hábito bloquea la inyección SQL, que es el ataque más común a bases de datos.

Llamando a APIs Externas

Muchas tareas del agente necesitan datos que la base de datos no tiene, como el estado de envío de un transportista o el clima para una ruta de entrega. El agente llama a una API HTTP externa, pasando el mismo JWT para que el servicio externo pueda verificar al llamante.

import urllib.request

def get_shipping_status(ctx: UserContext, order_id: int) -> dict:
    req = urllib.request.Request(
        f"https://api.courier.example/orders/{order_id}",
        headers={"Authorization": f"Bearer {ctx.token}"}
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())

El token se reenvía, no se genera uno nuevo. Esto permite al servicio de transporte registrar quién preguntó y aplicar sus propios límites de tasa por usuario.

Imponiendo Control de Acceso Basado en Roles

La autenticación te dice quién es el usuario. La autorización (RBAC) te dice qué tiene permiso para hacer. El lugar más limpio para imponer esto es una pequeña función de política llamada en cada punto de entrada.

def require_role(ctx: UserContext, allowed: list[str]):
    if ctx.role not in allowed:
        raise PermissionError(f"Role '{ctx.role}' not permitted")

def list_all_orders(ctx: UserContext) -> list:
    require_role(ctx, ["admin"])
    conn = sqlite3.connect("app.db")
    cur = conn.cursor()
    cur.execute("SELECT id, user_id, product, total FROM orders")
    rows = cur.fetchall()
    conn.close()
    return rows

Un usuario regular que llama a list_all_orders es rechazado. Un admin que la llama obtiene la tabla completa. La comprobación de política es una línea, pero es la diferencia entre un sistema seguro y una filtración de datos.

Uniendo Todo

La función principal del agente une la autenticación, el acceso a la base de datos, las llamadas a API y el RBAC en un solo flujo.

def handle_request(token: str, intent: str, params: dict):
    ctx = verify_token(token)
    if intent == "my_orders":
        return get_user_orders(ctx, params["user_id"])
    elif intent == "all_orders":
        return list_all_orders(ctx)
    elif intent == "shipping":
        return get_shipping_status(ctx, params["order_id"])
    else:
        raise ValueError("Unknown intent")

Cada路径 comienza con verify_token. Cada路径 que toca datos sensibles llama a require_role. Cada consulta a la base de datos está parametrizada. Cada llamada a API saliente lleva el token del usuario. Estos cuatro hábitos, repetidos en todas partes, son los que hacen que el agente esté listo para producción.

Resumen

Ahora tienes un agente end-to-end funcional. La autenticación prueba la identidad mediante tokens firmados. Las consultas SQL parametrizadas leen y escriben datos sin riesgo de inyección. Las llamadas a API externas reutilizan el mismo token para que los servicios externos puedan auditar y limitar por usuario. El control de acceso basado en roles se sitúa en cada punto de entrada, no solo en la puerta principal, por lo que una escalada de privilegios requiere eludir múltiples comprobaciones. Cuando construyas agentes reales, el patrón es el mismo: autentica una vez, autoriza en todas partes, parametriza siempre y propaga el contexto en cada llamada.