cultivaia_crm_mcp

#!/usr/bin/env python3
"""
Servidor MCP para el CRM interno de CULTIVA IA.

Permite a Claude consultar clientes, proyectos, pipeline de leads
y métricas del negocio sin salir del contexto de conversación.
"""

from typing import Optional, List
from enum import Enum
import os, json
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP

# ── Inicialización ────────────────────────────────────────────────────────────
mcp = FastMCP("cultivaia_crm_mcp")

API_BASE_URL = "https://api.cultivaia.internal/v1"
API_KEY      = os.environ["CULTIVA_CRM_API_KEY"]

# ── Enums ─────────────────────────────────────────────────────────────────────
class ResponseFormat(str, Enum):
    MARKDOWN = "markdown"
    JSON     = "json"

class ClientStatus(str, Enum):
    ACTIVE   = "active"
    PAUSED   = "paused"
    CHURNED  = "churned"

class ProjectStatus(str, Enum):
    IN_PROGRESS = "in_progress"
    REVIEW      = "review"
    DELIVERED   = "delivered"

# ── Modelos de entrada (Pydantic v2) ──────────────────────────────────────────
class SearchClientsInput(BaseModel):
    """Input para búsqueda de clientes."""
    model_config = ConfigDict(str_strip_whitespace=True, extra='forbid')

    query:           str              = Field(..., description="Texto a buscar (nombre, sector, contacto)", min_length=2)
    status:          Optional[ClientStatus] = Field(None, description="Filtrar por estado")
    sector:          Optional[str]     = Field(None, description="Sector (ej. 'ecommerce', 'saas', 'retail')")
    limit:           int               = Field(20, ge=1, le=100)
    offset:          int               = Field(0, ge=0)
    response_format: ResponseFormat    = ResponseFormat.MARKDOWN

class GetClientInput(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra='forbid')
    client_id: str = Field(..., description="ID único del cliente (ej. 'CL-0042')", pattern=r'^CL-\d{4}$')

class ListProjectsInput(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra='forbid')
    client_id:       Optional[str]         = Field(None, description="Filtrar por cliente")
    status:          Optional[ProjectStatus] = Field(None)
    limit:           int                   = Field(20, ge=1, le=100)
    offset:          int                   = Field(0, ge=0)
    response_format: ResponseFormat        = ResponseFormat.MARKDOWN

class GetPipelineInput(BaseModel):
    model_config = ConfigDict(extra='forbid')
    min_score:       Optional[int]  = Field(None, ge=0, le=100, description="Score mínimo de prioridad (0-100)")
    stage:           Optional[str]  = Field(None, description="Etapa: 'discovery','proposal','negotiation','won','lost'")
    limit:           int            = Field(30, ge=1, le=200)
    response_format: ResponseFormat = ResponseFormat.MARKDOWN

# ── Utilidades compartidas ────────────────────────────────────────────────────
async def _api(endpoint: str, params: dict = None) -> dict:
    """Cliente HTTP reutilizable para todas las llamadas al CRM."""
    async with httpx.AsyncClient() as client:
        r = await client.get(
            f"{API_BASE_URL}/{endpoint}",
            params={k: v for k, v in (params or {}).items() if v is not None},
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=15.0,
        )
        r.raise_for_status()
        return r.json()

def _err(e: Exception) -> str:
    """Mensajes de error accionables y consistentes."""
    if isinstance(e, httpx.HTTPStatusError):
        codes = {401: "API key inválida. Verifica CULTIVA_CRM_API_KEY.",
                  403: "Sin permisos para este recurso.",
                  404: "Recurso no encontrado. Verifica el ID.",
                  429: "Rate limit. Espera unos segundos e intenta de nuevo."}
        return f"Error: {codes.get(e.response.status_code, f'HTTP {e.response.status_code}')}"
    if isinstance(e, httpx.TimeoutException):
        return "Error: Timeout. El CRM no responde. Intenta de nuevo."
    return f"Error inesperado: {type(e).__name__}: {e}"

def _paginate_meta(total: int, offset: int, count: int) -> dict:
    return {"total": total, "count": count, "offset": offset,
             "has_more": total > offset + count,
             "next_offset": offset + count if total > offset + count else None}

# ── Herramientas MCP ──────────────────────────────────────────────────────────
@mcp.tool(name="cultivaia_search_clients",
          annotations={"readOnlyHint": True, "idempotentHint": True,
                        "destructiveHint": False, "openWorldHint": False})
async def cultivaia_search_clients(params: SearchClientsInput) -> str:
    """Busca clientes en el CRM por nombre, sector o texto libre.

    Soporta filtros por estado (active/paused/churned) y sector de negocio.
    Devuelve nombre, sector, MRR, estado y fecha de inicio.
    Usa response_format='json' para integración programática.

    Returns:
        str: Lista de clientes con MRR, sector y estado.
        Ejemplo: "Clientes encontrados: 12 (mostrando 10)" + tabla markdown
    """
    try:
        data = await _api("clients/search", {
            "q": params.query, "status": params.status,
            "sector": params.sector, "limit": params.limit, "offset": params.offset,
        })
        clients = data["clients"]
        if not clients: return f"Sin resultados para '{params.query}'."

        if params.response_format == ResponseFormat.JSON:
            return json.dumps({**_paginate_meta(data["total"], params.offset, len(clients)),
                                   "clients": clients}, indent=2)

        lines = [f"## Clientes — '{params.query}' ({data['total']} resultados)\n",
                  "| ID | Nombre | Sector | MRR | Estado |",
                  "|-------|--------|--------|-----|--------|"]
        for c in clients:
            lines.append(f"| {c['id']} | {c['name']} | {c['sector']} | {c['mrr_eur']}€ | {c['status']} |")
        return "\n".join(lines)
    except Exception as e: return _err(e)


@mcp.tool(name="cultivaia_get_client",
          annotations={"readOnlyHint": True, "idempotentHint": True,
                        "destructiveHint": False, "openWorldHint": False})
async def cultivaia_get_client(params: GetClientInput) -> str:
    """Devuelve el perfil completo de un cliente: contacto, proyectos,
    historial de MRR, NPS y notas internas. ID con formato CL-XXXX."""
    try:
        c = await _api(f"clients/{params.client_id}")
        return (f"## {c['name']} ({c['id']})\n\n"
                f"- **Sector**: {c['sector']}  · **Estado**: {c['status']}\n"
                f"- **Contacto**: {c['contact_name']} <{c['contact_email']}>\n"
                f"- **MRR actual**: {c['mrr_eur']}€  · **MRR inicio**: {c['mrr_start_eur']}€\n"
                f"- **NPS**: {c['nps']} / 10  · **Desde**: {c['start_date']}\n"
                f"- **Proyectos activos**: {c['active_projects']}\n\n"
                f"### Notas\n{c.get('notes', 'Sin notas.')}")
    except Exception as e: return _err(e)


@mcp.tool(name="cultivaia_get_metrics",
          annotations={"readOnlyHint": True, "idempotentHint": True,
                        "destructiveHint": False, "openWorldHint": False})
async def cultivaia_get_metrics() -> str:
    """Resumen de métricas del negocio: MRR total, ARR, NPS medio,
    churn mensual, clientes activos y pipeline valorado."""
    try:
        m = await _api("metrics/summary")
        return ("## Métricas CULTIVA IA — resumen ejecutivo\n\n"
                f"| Métrica | Valor |\n|---------|-------|\n"
                f"| MRR total | **{m['mrr_eur']:,}€** |\n"
                f"| ARR | **{m['arr_eur']:,}€** |\n"
                f"| Clientes activos | {m['active_clients']} |\n"
                f"| NPS medio | {m['avg_nps']} / 10 |\n"
                f"| Churn mensual | {m['monthly_churn_pct']}% |\n"
                f"| Pipeline abierto | {m['pipeline_value_eur']:,}€ |")
    except Exception as e: return _err(e)


if __name__ == "__main__":
    mcp.run()  # stdio por defecto