viclix-agent

Registro de tráfico para apps FastAPI, listo para visualizar en Viclix.

Instala un middleware ASGI que registra cada request (IP, usuario, endpoint, hora, duración, status…) en un archivo JSONL dedicado. La escritura ocurre en un hilo de fondo con el archivo siempre abierto, así que el camino del request nunca hace I/O de disco. Después, esos logs alimentan la visualización animada de Viclix: quién entró, desde qué IP, a qué páginas y en qué orden.

Instalación

pip install viclix-agent

Uso mínimo

from fastapi import FastAPI
from viclix_agent import MiddlewareLogger

app = FastAPI()
app.add_middleware(MiddlewareLogger)

Eso ya crea la carpeta .viclix-agent/middleware/ y escribe ahí traffic-YYYY-MM-DD.jsonl.

Configuración

app.add_middleware(
    MiddlewareLogger,
    log_path=".viclix-agent/middleware/traffic.jsonl",  # ruta base (se le añade la fecha)
    trust_proxy=True,                       # IP real desde X-Forwarded-For (PythonAnywhere)
    rotation="daily",                       # "daily" o None
    include_query=False,                    # incluir query string (ojo con datos sensibles)
    ignore_paths=["/static", "/health"],    # prefijos que no se registran
)

Parámetro	Default	Descripción
log_path	.viclix-agent/middleware/traffic.jsonl	Ruta base del log.
trust_proxy	True	Lee la IP de X-Forwarded-For / X-Real-IP.
user_resolver	lee request.state.user	Callable[[scope], id] para extraer el usuario.
rotation	"daily"	Rotación diaria del archivo, o None.
include_query	False	Incluye el query string en el evento.
ignore_paths	[]	Prefijos de path a ignorar.
track_sessions	True	Cookie de sesión anónima (viclix_sid) por visitante.
session_cookie	"viclix_sid"	Nombre de la cookie de sesión.
session_ttl_days	365	Duración de la cookie de sesión.

Sesiones (trazar el flujo de un visitante)

Con track_sessions=True (por defecto), cada visitante recibe una cookie anónima viclix_sid y cada evento lleva un campo session. Así puedes seguir la secuencia completa de páginas de un mismo visitante aunque no haya login — por ejemplo, para ver qué flujo llevó a un usuario a un error 4xx/5xx:

GET /            -> 200   sess=44ef64b0
GET /checkout    -> 200   sess=44ef64b0
GET /boom        -> 500   sess=44ef64b0   ← el flujo que terminó en error

Identificar al usuario

Por defecto se lee request.state.user (lo que deje tu capa de auth). Si guardas un objeto, se prueban username, email, id, pk. Para lógica propia:

def resolver(scope):
    state = scope.get("state") or {}
    user = state.get("user")
    return getattr(user, "email", None)

app.add_middleware(MiddlewareLogger, user_resolver=resolver)

Rotación y retención (tamaño del log)

Por defecto los logs rotan cada hora, los archivos cerrados se comprimen a .gz y se borran tras 24 h:

app.add_middleware(
    MiddlewareLogger,
    rotation="hourly",       # "hourly" (default) | "daily" | None
    retention_hours=24,      # borra archivos más viejos que esto (None = sin límite)
    max_total_mb=200,        # tope duro de tamaño: borra los más viejos (None = sin tope)
    compress=True,           # gzip de los archivos ya rotados
)

• retention_hours acota por tiempo; max_total_mb es la red de seguridad por tamaño (un pico de tráfico o un bot puede llenar el disco dentro del plazo de retención).
• La limpieza y la compresión corren en el hilo de fondo al rotar — nunca en el camino del request — y solo tocan archivos del propio log.
• El visor lee .jsonl y .jsonl.gz de forma transparente.

El mayor ahorro suele ser no registrar assets estáticos:

app.add_middleware(MiddlewareLogger, ignore_paths=["/static", "/assets", "/favicon.ico"])

Las background tasks aceptan los mismos parámetros:

track_background_tasks(name="checkout", rotation="hourly", retention_hours=24)

Consultas SQL (SQLAlchemy)

Desde fuera ves que un request tardó 2 s; nunca por qué. Con una línea cada evento pasa a llevar el detalle de lo que hizo contra la base de datos:

from viclix_agent import track_db_queries

track_db_queries()              # todos los engines del proceso
track_db_queries(engine)        # o solo uno

No se toca ninguna query: engancha los eventos before/after_cursor_execute de SQLAlchemy. No escribe una línea por consulta — eso multiplicaría el volumen. Agrega por request (y por background task) y lo adjunta como campo db del evento que ya se emitía:

"db": {"n": 21, "ms": 312.4, "slow": 2,
       "slowest_ms": 40.1, "slowest": "SELECT id FROM projects",
       "repeat": {"sql": "SELECT name FROM projects WHERE id = ?", "n": 20, "ms": 180.2}}

repeat es el detector de N+1: la misma consulta —normalizada, sin valores— repetida muchas veces dentro del mismo request. En el visor, los endpoints con N+1 salen marcados con ⚠ y el número medio de consultas.

Solo se guarda el texto de la sentencia, nunca los parámetros: en SQLAlchemy los valores viajan aparte (bind params), así que no se filtran datos.

Opciones: slow_ms (umbral de consulta lenta, 100 por defecto), repeat_threshold (repeticiones para marcar N+1, 5) y max_distinct (tope de consultas distintas rastreadas por request, 200).

Excepciones

Un 500 visto desde fuera es una caja negra idéntica siempre. El middleware captura las excepciones no manejadas —es el único sitio donde todavía existe el traceback— y las añade al evento, atribuidas al endpoint, la IP y el usuario:

"error": {"type": "ValueError", "msg": "algo se rompió feo",
          "where": "routes.py:142 in checkout", "traceback": "Traceback ..."}

Va activado por defecto (capture_errors=True) y no altera el flujo: la excepción se re-lanza tal cual. Con error_traceback=False se guardan solo tipo, mensaje y archivo:línea. Las background tasks que fallan registran lo mismo en su propio log (y no se duplican en el request que las creó).

No se capturan variables locales, solo el traceback formateado.

Servicios externos (httpx / requests)

Desde fuera no puedes atribuir "el checkout fue lento porque Stripe tardó 3 s": el proxy solo ve un request lento. Con una línea:

from viclix_agent import track_outbound_calls

track_outbound_calls(slow_ms=500)

Envuelve httpx.Client.send, httpx.AsyncClient.send y requests.Session.send. Igual que el SQL, agrega por request/tarea y viaja como campo out:

"out": {"n": 2, "ms": 3200.5, "slow": 1,
        "hosts": [{"host": "api.stripe.com", "n": 1, "ms": 3000.2}],
        "slowest": {"host": "api.stripe.com", "method": "POST",
                    "path": "/v1/charges", "ms": 3000.2, "status": 200}}

Del URL se guardan host y path; el query string se descarta porque suele llevar tokens y claves de API. En el visor hay un puerto (🛰) donde la personita espera el tiempo que tardó cada tercero.

Con respuestas en streaming el tiempo llega hasta las cabeceras, no hasta el último byte del cuerpo.

Salud del proceso

Esta sí va a su propio log (.viclix-agent/health/<name>.jsonl), porque no cuelga de ningún request: es un muestreo periódico.

from viclix_agent import track_health

track_health(name="www", interval=15)

{"kind": "health", "app": "www", "loop_lag_ms": 3.4,
 "rss_mb": 182.5, "threads": 12, "in_flight": 2,
 "pool": {"in_use": 3, "size": 5, "overflow": 0}}

El dato clave es loop_lag_ms: cuánto tarda el event loop en atender algo que ya estaba listo. Desde fuera solo ves "latencia alta"; el lag te dice que hay una llamada síncrona bloqueando el bucle async. Se mide desde el hilo de muestreo con loop.call_soon_threadsafe, así que no ocupa sitio en el bucle.

rss_mb usa psutil si está instalado y si no /proc/self/statm (Linux). pool sale del primer engine de SQLAlchemy que se vea usar, así que necesita track_db_queries() activo. En el visor esto es el clima de la ciudad: se nubla y llueve cuando el proceso sufre.

Requests colgados y quién bloquea el loop

Dos parámetros de track_health, sin líneas nuevas:

track_health(name="www", stuck_after=30, profile_on_lag=200)

stuck_after (segundos) reporta los requests que siguen en vuelo. Es la única forma de ver un request que nunca termina: el evento normal se escribe al acabar, así que un deadlock o un tercero sin timeout no dejaban rastro ninguno. Aparecen en la muestra de salud mientras siguen colgados:

"stuck": [{"method": "GET", "path": "/pay/charge", "ip": "1.2.3.4", "age_s": 47.2}]

profile_on_lag (ms) saca la foto de los stacks cuando el loop se atasca. El truco está en el orden: la sonda espera solo ese plazo y, si el aviso no ha vuelto, el loop está bloqueado ahora mismo — es el momento de fotografiar. Esperar a que vuelva daría el stack del loop ya en reposo, que no dice nada.

"blocked_by": "billing.py:61 in summarize",
"stacks": [{"thread": "MainThread", "stack": ["...", "billing.py:61 in summarize"]}]

Caché (Redis)

from viclix_agent import track_cache
track_cache()               # redis.Redis y redis.asyncio.Redis
track_cache(mi_cliente)     # o solo ese cliente

Campo cache en el evento: {"n": 10, "hits": 8, "misses": 2, "ms": 4.1}. Desde fuera no distingues una respuesta cacheada de una recalculada. Solo se cuenta el comando, nunca la clave ni el valor.

Autenticación y seguridad

from viclix_agent import track_security
track_security(name="www", login_paths=["/login", "/signup"])

Log propio (.viclix-agent/security/<name>.jsonl), retención de 30 días. Detecta sin tocar tu código: unauthorized (401), denied (403), login.failed y login.ok en los paths de login.

Lo importante es que se puede agrupar por cuenta: un atacante que rota IPs contra un solo usuario es invisible agrupando por IP y evidente por cuenta.

Límite honesto: desde el middleware solo se ve el código de estado. El motivo (contraseña incorrecta vs. cuenta bloqueada vs. token caducado) solo lo sabe tu capa de auth. Para eso hay una llamada opcional donde te importe:

from viclix_agent import record_auth
record_auth("login.failed", user=email, reason="bad_password")

Cuando se usa, la detección automática calla para ese request, así no se registra el mismo intento dos veces.

Arranque, parada y deploy

from viclix_agent import track_lifecycle
track_lifecycle(name="www", version=settings.release, config=settings)

Log propio y diminuto: una o dos líneas por despliegue. Desde fuera un redeploy y un crash-loop se ven idénticos; aquí se distinguen porque un start sin su stop previo significa que el proceso anterior murió sucio (SIGKILL, OOM). No se instalan manejadores de señales para no pisar los de uvicorn.

De la config se guardan los nombres de las claves no sensibles y una huella de todos los valores — nunca los valores. Si la huella cambia entre dos arranques de la misma versión, alguien tocó la configuración.

Tamaño de las respuestas

Va solo, sin configurar nada: campo bytes en cada evento. Detecta el endpoint que devuelve 4 MB de JSON sin paginar. Se desactiva con MiddlewareLogger(track_bytes=False).

A diferencia del resto de este README, esta no es exclusiva de dentro: tu proxy también ve los bytes. Está aquí porque es gratis.

Visor dentro de tu app (sin CLI)

Dos líneas y tienes el visor en tu propio dominio, sin arrancar nada:

from viclix_agent import mount_viewer
mount_viewer(app, "/_viclix", token=settings.viclix_token)

Quedan tudominio.com/_viclix y tudominio.com/_viclix/experimental.

Solo tus admins, sin token

mount_viewer(app, "/_viclix", authorize=mi_check_de_admin)

authorize recibe el scope ASGI y devuelve True/False; de ahí sacas la cookie de sesión igual que en user_resolver. Si pasas los dos, entra quien cumpla cualquiera de los dos.

Lo que hace por seguridad

• El token desaparece de la URL. Al llegar con ?token=… válido se guarda en una cookie HttpOnly; SameSite=Strict; Path=<visor> y se redirige al mismo path sin query. El token aparece una vez y no queda en el historial, ni en el Referer de las peticiones siguientes, ni en los logs del proxy.
• Responde 404, no 401. Quien no tenga la llave no sabe que el visor existe.
• Falla cerrado. Sin token ni authorize, o con un token de menos de 16 caracteres, mount_viewer lanza ValueError al arrancar en vez de servir los logs a cualquiera. Genera el token con secrets.token_urlsafe(32).
• Comparación en tiempo constante (hmac.compare_digest).
• Los intentos fallidos se registran como viewer.denied en el log de seguridad, si track_security() está activo.
• El visor no se registra a sí mismo: navegarlo no genera los eventos que estás mirando.

Lo que NO hace

No cifra nada ni limita el ritmo de intentos. Y sobre todo: estos logs llevan emails, tracebacks, sentencias SQL, IPs y nombres de claves de configuración. Móntalo solo detrás de HTTPS y trátalo como un panel de administración, porque es exactamente eso.

Varias apps (mismo venv / servidor)

Si tienes varias apps FastAPI en el mismo directorio/venv, dale a cada una un name distinto y cada una escribirá en su propio archivo (.viclix-agent/middleware/<name>.jsonl), en vez de compartir uno solo:

app.add_middleware(MiddlewareLogger, name="checkout")   # → checkout.jsonl
app.add_middleware(MiddlewareLogger, name="auth")       # → auth.jsonl
track_background_tasks(name="checkout")                 # tasks también separadas

Además, cada evento lleva un campo app. En el visor:

viclix-agent                  # agrega TODAS las apps (avisa cuáles hay)
viclix-agent --app checkout   # solo esa app

Sin name, todas comparten traffic.jsonl (comportamiento anterior).

Visor (CLI)

El paquete instala el comando viclix-agent. Al ejecutarlo en tu terminal:

viclix-agent

1. Lee los logs de .viclix-agent/middleware/ (todos los días agregados). Si no existe esa carpeta, busca recursivamente (hasta 3 niveles) como respaldo.
2. Levanta un servidor local y abre el navegador con una visualización interactiva.
3. Muestra estadísticas, línea de tiempo por sesión/IP (con animación del tráfico), endpoints e IPs más activas, y una tabla filtrable por IP, path, usuario o clase de status.

Mundo vivo (experimental)

En /experimental (enlace en el dashboard) hay una representación isométrica donde cada IP es una personita que sale de la "entrada" y camina hacia el endpoint que solicitó. Los endpoints se agrupan por clase (primer segmento del path) en distritos — /products, /products/42 → distrito products — y el mapa empieza en 0 y crece a medida que se descubren IPs y endpoints. El color del pulso indica la clase de status del último request (verde/amarillo/naranja/rojo). Controles: reproducir, velocidad, scrub temporal, arrastrar para mover, rueda para zoom, Fit para reencuadrar. El motor de personajes es el de POKER (sprites.js).

Esta vista es autocontenida y pública: es la base de lo que luego se mostrará en Viclix.

Opciones:

viclix-agent [LOG]           # ruta a un .jsonl específico (opcional)
  --dir .                    # directorio raíz de búsqueda
  --depth 3                  # profundidad máxima de búsqueda
  --port 8787                # puerto del servidor local
  --all-days                 # agregar todos los archivos diarios de la misma base
  --no-browser               # no abrir el navegador automáticamente

Formato del log (JSONL)

Una línea JSON por request:

{"v":1,"ts":"2026-07-24T16:50:55.349029+00:00","ip":"1.2.3.4","method":"GET","path":"/dashboard","status":200,"duration_ms":3.06,"user":"jp@sizth.com","ua":"Mozilla/5.0","referer":null}

Campo	Descripción
v	Versión del esquema.
ts	Timestamp ISO-8601 en UTC.
ip	IP del cliente.
method	Método HTTP.
path	Ruta solicitada.
status	Código de respuesta.
duration_ms	Duración del request en ms.
user	Identificador de usuario (o null).
session	ID de sesión anónima (cookie viclix_sid).
ua	User-Agent.
referer	Referer.
query	Query string (solo si include_query=True).

Notas de rendimiento

• Los eventos se encolan en memoria y los escribe un hilo daemon; el request no espera al disco.
• Si la cola se satura (queue_maxsize, 10k por defecto), los eventos nuevos se descartan en vez de bloquear la app. El writer lleva contadores written y dropped.
• El archivo se hace flush cada segundo y se cierra limpiamente al terminar el proceso (atexit).