# Volkom Mesh — Referencia de la API de telemetría (para implementar) Documento técnico, autocontenido, para que un dev o un asistente de IA (Claude Code, Cursor) instrumente un agente. Dos canales, **una API, una base por tenant**. Contratos verificados contra el código. ## Modelo mental - **Tenant** = cliente, base ClickHouse aislada (`volkom_`). - **Agente** = `agent_name` (slug estable, ej. `sales-monitor`). - **Corrida** = todos los eventos con el mismo `run_id` (UUID). Secuencia: `started` → N×`in_progress` → `completed` | `failed`. - **Canal ejecución** → tabla `telemetry_events`. **Canal negocio** → tabla `business_metrics`. ## Autenticación Todos los endpoints de telemetría usan **Bearer token del agente**: ``` Authorization: Bearer ``` El token se genera en el portal (**Conectá tus agentes** → key del workspace, que sirve para toda la flota; o una key por agente desde la ficha del agente). Un token de workspace puede reportar cualquier `agent_name`; una key por-agente solo el suyo (anti-spoofing: 403 `Key bound to agent ''`). > Rate limit: 100 req/60s por token. Si tu flota lo supera, usá > `POST /telemetry/batch` (N eventos en un request) antes de pedir más cuota. Base URL: `https://api.volkom.ai/api/v1`. Swagger interactivo: `https://api.volkom.ai/docs`. --- ## 1) Registrar el agente — `POST /api/v1/agents/register` Idempotente (upsert). Declara descripción + KPIs de negocio → puebla el catálogo (`label`/`unit`/`agg`) que el front usa como diccionario. Status: **200**. Requiere `Authorization: Bearer`. **Body** (`AgentRegister`): ```json { "agent_name": "sales-monitor", "agent_version": "1.0.0", "description": "Monitorea ventas diarias.", "telemetry_doc": "Texto libre de qué reporta.", "business_kpis": [ {"key": "orders", "label": "Pedidos", "unit": "count", "agg": "sum"}, {"key": "revenue", "label": "Facturación", "unit": "money:USD", "agg": "sum"} ] } ``` `business_kpis` es `[]` para agentes sin negocio. Campos de cada KPI (`BusinessKPIDecl`): `key` (req), `label` (default `""`), `unit` (default `"count"`), `agg` (default `"sum"`). --- ## 2) Evento de ejecución — `POST /api/v1/telemetry` Un POST **por evento** (started, cada paso, cierre). Status: **202**. **Body** (`TelemetryEventIn`): ```json { "agent_name": "sales-monitor", "agent_version": "1.0.0", "client_id": "shopflow", "run_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479", "event": "in_progress", "timestamp": "2026-05-30T14:32:05Z", "message": "Leyó pedidos", "level": "info", "category": "Ingesta", "source": "store/read.py:88", "seq": 2, "metrics": {"pedidos": 42.0}, "links": ["https://oms.internal/batch/abc"], "metadata": {"region": "AR"}, "duration_ms": null, "records_processed": null, "error_type": null, "error_message": null, "output": null, "workflow_run_id": null } ``` **Campos:** | Campo | Tipo | Notas | |-------|------|-------| | `agent_name` | str req | | | `agent_version` | str req | | | `client_id` | str req | = slug del tenant | | `run_id` | UUID req | **el mismo para todos los eventos de la corrida** | | `event` | enum req | `started` \| `in_progress` \| `completed` \| `failed` | | `timestamp` | ISO-8601 req | UTC recomendado | | `message` | str | descripción humana (clave en `in_progress`) | | `level` | str | `debug`\|`info`\|`warning`\|`error`\|`success` (default `info`) | | `category` | str | agrupador/fase; el front agrupa los pasos por esto | | `source` | str | `archivo:línea` | | `seq` | int | orden dentro de la corrida (default 0) | | `metrics` | map | mediciones numéricas graficables | | `links` | list | URLs/refs externas | | `metadata` | map | atributos planos libres | | `duration_ms` | int? | normalmente en el cierre | | `records_processed` | int? | normalmente en el cierre | | `error_type` / `error_message` | str? | en `failed` | | `output` | obj? | payload rico anidado | | `workflow_run_id` | str? | si la corrida es parte de un workflow Temporal | **Patrón de una corrida (3+ POST con el mismo `run_id`):** ``` POST /telemetry {event:"started", run_id:R, timestamp:t0} POST /telemetry {event:"in_progress", run_id:R, seq:1, message:"…", category:"…", metrics:{…}} POST /telemetry {event:"in_progress", run_id:R, seq:2, message:"…", level:"warning"} POST /telemetry {event:"completed", run_id:R, duration_ms:16000, records_processed:42} ``` --- ## 3) KPIs de negocio — `POST /api/v1/telemetry/business` Un POST por corrida (los KPIs de ese run). Status: **202**. Canal separado; crea/extiende columnas on-demand en `business_metrics`. **Body** (`BusinessTelemetryIn`): ```json { "agent_name": "sales-monitor", "agent_version": "1.0.0", "client_id": "shopflow", "run_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479", "timestamp": "2026-05-30T14:32:20Z", "metrics": {"orders": 42, "revenue": 9500}, "workflow_run_id": null } ``` `metrics`: map de `key` (debe coincidir con un KPI declarado en register) → valor (number|string). `run_id` puede ser el mismo de la corrida de ejecución para correlacionar. `agent_version`/`client_id` opcionales pero recomendados. --- ## Vía carcass (volkom-core) — equivalente, sin HTTP Si el agente es Python, heredar de `BaseAgent` hace los 3 POST por vos y maneja `run_id`/`started`/cierre/tiempos automáticamente. ```python from volkom_core import BaseAgent, AgentResult, KPI, VolkomConfig, TelemetryExporter class SalesMonitor(BaseAgent): agent_name = "sales-monitor" agent_version = "1.0.0" description = "Monitorea ventas diarias." telemetry_doc = "Pedidos y facturación por corrida." business_kpis = [ KPI("orders", "Pedidos", "count", "sum"), KPI("revenue", "Facturación", "money:USD", "sum"), ] def run(self) -> AgentResult: self.event("Conectó a la tienda", category="Ingesta", source="store/conn.py:20") self.event("Leyó pedidos", category="Ingesta", metrics={"pedidos": 42}) self.warn("3 pedidos sin dirección", category="Validación", metrics={"incompletos": 3}) self.event("Cargó al panel", category="Cierre", links=["https://oms.internal/batch/abc"]) self.kpi("orders", 42) self.kpi("revenue", 9500) return AgentResult(records_processed=42) ``` **API del carcass:** - `event(message, *, level, category, attributes, metrics, links, source, kind)` — emite un `in_progress` (o `start`/`finish` con `kind`). `attributes` → `metadata`. - `step / info / warn / error (message, ...)` — wrappers por nivel. - `kpi(key, value)` — KPI de negocio (debe estar en `business_kpis`). - `emit_metric(key, value)` — metadata libre en el evento de cierre (NO es KPI). - `started`/`completed`/`failed`, `run_id`, `duration_ms` → automáticos en `execute()`. **Registro (una vez, al arrancar el cliente):** ```python exporter = TelemetryExporter(cfg) # cfg = VolkomConfig(client_id, control_plane_url, auth_token) exporter.register_agent( agent_name=SalesMonitor.agent_name, agent_version=SalesMonitor.agent_version, description=SalesMonitor.description, telemetry_doc=SalesMonitor.telemetry_doc, business_kpis=[{"key": k.key, "label": k.label, "unit": k.unit, "agg": k.agg} for k in SalesMonitor.business_kpis], ) # correr una vez (genera run_id + started + cierre): SalesMonitor(cfg, exporter).execute(run_id=, workflow_run_id=) ``` --- ## Vocabulario de KPIs **`unit`:** `count` · `money:` (ej. `money:USD`, `money:ARS`, `money:EUR`) · `percent` · `duration` (valor en segundos). **`agg`** (cómo se totaliza en el período del dashboard): `sum` (acumulables: ventas, facturas) · `last` (estados: stock, MRR, %ocupación) · `avg` (promedios: tiempo de espera) · `max` (picos). > Elegir mal el `agg` produce totales incorrectos: `sum` de un estado infla el > número; `last` de algo acumulable lo subestima. --- ## Lectura (lo que consumen los fronts — referencia) Misma API, misma DB; portal y admin leen de acá: - `GET /api/v1/tenants/{slug}/dashboard/runs?include_steps=&since=&until=&limit=` — corridas; con `include_steps=true` incluye los `in_progress`. - `GET /api/v1/tenants/{slug}/runs/{run_id}/events` — narrativa completa de una corrida. - `GET /api/v1/tenants/{slug}/business/catalog` — KPIs declarados (label/unit/agg). - `GET /api/v1/tenants/{slug}/business/summary?time_range=` — KPIs agregados por `agg`. - `GET /api/v1/tenants/{slug}/business/series?agent_name=&key=&time_range=&bucket=` — tendencia. --- ## Checklist de implementación 1. [ ] `POST /agents/register` con `business_kpis` declarados (idempotente). 2. [ ] Por corrida: un `run_id` UUID único, **reusado en todos los eventos**. 3. [ ] `started` al inicio; `in_progress` por cada paso (con `category`, `metrics`, `level` según corresponda); `completed`/`failed` al final con `duration_ms`. 4. [ ] KPIs de negocio vía `POST /telemetry/business` (keys = las declaradas). 5. [ ] Header `Authorization: Bearer ` en todo. 6. [ ] Cerrar SIEMPRE la corrida (sin cierre → "en curso" perpetuo). 7. [ ] Errores con `level:"error"` / `event:"failed"` + `error_type`/`error_message`. ## Errores y gotchas - **`run_id` distinto por evento** → cada paso se ve como corrida separada. Reusarlo. - **Corrida sin cierre** → figura "en curso" para siempre. - **KPI no declarado** → el valor entra pero sin metadata de catálogo (unidad/agg). - **Rate limit 429** → respetar 100 req/60s por token; para ráfagas, `POST /telemetry/batch`. - **Anti-spoofing 403** → una key por-agente solo puede reportar su propio `agent_name`.