Servidor MCP Analytics

Un servidor MCP listo para producción que expone métricas de producto desde un data warehouse interno. Este es un caso de estudio completo aplicando la jerarquía de Anthropic: primitivas → herramientas custom → MCP.

Qué Es Esto

Un ejemplo del mundo real de cuándo construir MCP en lugar de usar:

• Primitivas (ejecución de bash/código con SQL)
• Herramientas custom (definiciones de herramientas en prompts del agente)

Resultado: Una capacidad de analytics compartida, gobernada, multi-cliente.

Inicio Rápido

# 1. Instalar
pip install -r requirements.txt

# 2. Inicializar datos
python data/init_db.py

# 3. Ejecutar pruebas
python tests/test_functions.py

Salida esperada:

=== Test 1: list_tools ===
[OK] 4 herramientas recibidas:
  - query_metrics
  - get_product_summary
  - top_products_by_metric
  - event_breakdown
  
=== Test 2: call_tool (query_metrics) ===
[OK] Tipo de resultado: text
  Compras de producto 1: count=523, avg=99.45
  
=== Todas las pruebas pasaron ===

Arquitectura

El Patrón de 4 Piezas

Cada servidor MCP necesita:

1. Funciones Python normales (lógica encapsulada)
2. Objeto Server con nombre estable
3. Manejadores (list_tools, call_tool)
4. Punto de entrada (stdio_server + asyncio.run)

Ver src/server.py para la implementación completa.

Herramientas Expuestas

Herramienta	Propósito	Ejemplo
query_metrics	Filtrar eventos por producto/tipo/fecha	query_metrics(product_id=1, event_type="purchase")
get_product_summary	Resumen de producto + desglose de eventos	get_product_summary(product_id=2)
top_products_by_metric	Clasificar productos por count/sum/avg	top_products_by_metric(metric="count", limit=5)
event_breakdown	Eventos por región & tipo	event_breakdown(product_id=1)

Decisión: ¿Por Qué MCP?

El Problema

Múltiples agentes Claude necesitan consultar un data warehouse interno (eventos de producto). Los enfoques ingenuos fallan:

• Primitivas (bash/código): SQL desde agentes = riesgo de inyección, sin auditoría, duplication
• Herramientas custom: Schema duplicado en cada agente = 2k tokens desperdiciados × 10 agentes = 20k tokens duplicados

La Solución: Servidor MCP

Un servidor compartido por todos los agentes:

• ✅ Única fuente de verdad (una definición de schema)
• ✅ Acceso gobernado (rate limiting, auditoría)
• ✅ Observable (todos los queries registrados)
• ✅ Escala a cualquier cliente (agentes Claude, CLI tools, dashboards, webhooks)
• ✅ Seguro (queries parametrizadas, inputs validados)

Análisis de Costo

• Contexto: 1200 tokens (vs 2000 custom × 10 agentes = 20k) → Ahorro 40%
• Latencia: +50–80ms overhead IPC (aceptable para analytics)

Criterios MCP Cumplidos

¿Justifica este caso MCP? ✅ SÍ, todos los 4:

1. Colección de herramientas comunes → 4 patrones de query cubren 95% de requests
2. Estandarizado → JSON schema, versionado, backward-compatible
3. Requiere governance → DB compartida = rate limits, cost tracking, audit logging
4. Compartido entre clientes → Usado por agentes Claude, dashboards, reportes scheduled, monitoreo

Ver docs/decision.md para análisis completo.

Integración

Claude Code

# Añade a ~/.claude/settings.json
{
  "mcpServers": {
    "analytics": {
      "command": "python",
      "args": ["<ruta>/mcp-analytics/src/server.py"]
    }
  }
}

# En sesión: /mcp → selecciona analytics
# Los agentes pueden llamar: query_metrics, get_product_summary, etc.

Claude Desktop

{
  "mcpServers": {
    "mcp-analytics": {
      "command": "python",
      "args": ["-m", "mcp_analytics.server"]
    }
  }
}

Ver docs/setup.md para instrucciones completas.

Documentación

• docs/decision.md → El argumento completo: primitivas vs. custom vs. MCP
• docs/architecture.md → Comparación visual de flujos de datos; análisis de escalabilidad
• docs/token_analysis.md → Mediciones de costo de contexto; desglose de latencia
• docs/stockpilot_pattern.md → Ingesta de datos de alto contexto; cuándo optimizar
• docs/setup.md → Integración con Claude Code/Desktop, deployment en producción

Concepto Clave: El Patrón Middleware

MCP no reemplaza tu backend; reemplaza el middleware.

Antes:  Agente → Ejecución de SQL → Base de datos
Después: Agente → Servidor MCP (middleware) → Base de datos

El middleware:
- Traduce requests JSON a queries SQL
- Aplica rate limits & auditoría
- Cachea datos accedidos frecuentemente
- Maneja connection pooling
- Presenta interfaz uniforme

Si cambias BD (SQLite → Snowflake)?
Solo el servidor cambia. Los agentes no se entaran.

Si añades caching o rate limiting?
Solo el servidor cambia. Los agentes no se entaran.

Si añades un nuevo cliente (dashboard, CLI, webhook)?
Se conecta al mismo servidor.

Esto es por qué MCP es un estándar: desacopla clientes de detalles de implementación.

Producción: Checklist

• [ ] Base de datos: Migrar desde SQLite a PostgreSQL (concurrencia)
• [ ] Rate limiting: Presupuesto de tokens por cliente (10 queries/min)
• [ ] Caching: Capa Redis para queries repetidas
• [ ] Monitoreo: Métricas Prometheus (queries/sec, latencia)
• [ ] Auditoría: Enviar query logs a logging centralizado (ELK, DataDog)
• [ ] Control de acceso: API key por cliente; aplicar en servidor
• [ ] Testing: Load test a 10k QPS; tests de integración por tipo de cliente
• [ ] Documentación: Auto-generar OpenAPI desde schema MCP

Archivos

mcp-analytics/
├── src/
│   └── server.py          # El servidor MCP (patrón de 4 piezas)
├── data/
│   └── init_db.py         # Genera datos de ejemplo
├── tests/
│   └── test_functions.py  # Tests de integración (verificación handshake)
├── docs/
│   ├── decision.md        # ¿Por qué MCP (primitivas vs custom vs MCP)
│   ├── architecture.md    # Comparación visual & análisis de escalabilidad
│   ├── token_analysis.md  # Mediciones de costo de contexto
│   ├── stockpilot_pattern.md  # Ingesta de datos de alto contexto
│   └── setup.md           # Instrucciones de integración
├── requirements.txt       # Dependencias
└── README.md             # Este archivo

Ejecución Local

# Terminal 1: Inicia servidor
python src/server.py
# Salida: (esperando conexión del cliente)

# Terminal 2: Ejecuta tests del cliente
python tests/test_functions.py
# Salida: [OK] 4 herramientas listadas, [OK] todas las calls exitosas

Próximos Pasos

1. Lee docs/decision.md para entender cuándo usar MCP
2. Ejecuta tests para ver el handshake funcionando
3. Intégra con Claude Code (/mcp → analytics)
4. Despliega como servicio/Docker para producción
5. Añade autenticación, rate limiting, caching según docs/setup.md

Preguntas

• ¿Por qué no primitivas? Ver docs/decision.md (seguridad, governance, duplication)
• ¿Cuál es el costo de tokens? Ver docs/token_analysis.md (52% ahorro vs primitivas)
• ¿Cómo escala esto? Ver docs/architecture.md (soporta 50+ agentes vs 3 con custom tools)
• ¿Qué pasa con la latencia? Ver docs/token_analysis.md (+50–80ms overhead IPC, aceptable)
• ¿Cómo despliego? Ver docs/setup.md (Systemd, Docker, Managed Agents)

Licencia

MIT (ejemplo; usa la tuya propia)