Servidor MCP — CULTIVA IA CRM

"""
CULTIVA IA CRM — Servidor MCP
Conecta agentes IA con el CRM interno de CULTIVA via stdio.
"""

import os
from typing import Optional, Literal
import httpx
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP

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

BASE_URL = "https://api.cultivaia.com/v1"


def _get_client() -> httpx.AsyncClient:
    api_key = os.environ.get("CULTIVA_API_KEY")
    if not api_key:
        raise ValueError(
            "CULTIVA_API_KEY no configurada. "
            "Añádela a tu .env o a la configuración de Claude Desktop."
        )
    return httpx.AsyncClient(
        base_url=BASE_URL,
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        timeout=10.0,
    )


def _raise_for_status(r: httpx.Response) -> None:
    if r.status_code == 401:
        raise PermissionError("API key inválida o expirada. Regenera tu token en app.cultivaia.com/settings.")
    if r.status_code == 404:
        raise LookupError(f"Recurso no encontrado (404). Comprueba el ID proporcionado.")
    r.raise_for_status()


# ─── Modelos Pydantic ─────────────────────────────────────────

LeadStatus = Literal["nuevo", "contactado", "propuesta", "negociacion", "cerrado", "perdido"]

class ListLeadsInput(BaseModel):
    status: Optional[LeadStatus] = Field(None, description="Filtrar por estado del pipeline")
    servicio: Optional[str] = Field(None, description="Ej: 'Agentes-IA', 'SEO', 'Web'")
    limit: int = Field(20, ge=1, le=100, description="Resultados por página (1–100)")
    offset: int = Field(0, ge=0, description="Desplazamiento para paginación")

class GetLeadInput(BaseModel):
    lead_id: str = Field(..., description="ID único del lead (ej: 'LEAD-0042')")

class CreateLeadInput(BaseModel):
    nombre: str = Field(..., description="Nombre de la empresa o contacto")
    email: str = Field(..., description="Email de contacto")
    servicio_interes: str = Field(..., description="Servicio de interés (ej: 'Agentes-IA')")
    presupuesto_eur: Optional[int] = Field(None, description="Presupuesto estimado en EUR")
    notas: Optional[str] = Field(None, description="Notas iniciales sobre el lead")

class UpdateStatusInput(BaseModel):
    lead_id: str = Field(..., description="ID del lead a actualizar")
    nuevo_status: LeadStatus = Field(..., description="Nuevo estado del pipeline")
    motivo: Optional[str] = Field(None, description="Motivo del cambio (opcional)")


# ─── Herramientas MCP ─────────────────────────────────────────

@mcp.tool(
    name="cultiva_list_leads",
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
async def cultiva_list_leads(params: ListLeadsInput) -> str:
    """Lista leads del CRM con filtros opcionales por estado y servicio.
    Soporta paginación via limit/offset. Devuelve JSON con items y total."""
    query = {"limit": params.limit, "offset": params.offset}
    if params.status: query["status"] = params.status
    if params.servicio: query["servicio"] = params.servicio
    async with _get_client() as client:
        r = await client.get("/leads", params=query)
        _raise_for_status(r)
        return r.text


@mcp.tool(
    name="cultiva_get_lead",
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
async def cultiva_get_lead(params: GetLeadInput) -> str:
    """Obtiene el detalle completo de un lead por su ID (ej: 'LEAD-0042')."""
    async with _get_client() as client:
        r = await client.get(f"/leads/{params.lead_id}")
        _raise_for_status(r)
        return r.text


@mcp.tool(
    name="cultiva_create_lead",
    annotations={"readOnlyHint": False, "idempotentHint": False},
)
async def cultiva_create_lead(params: CreateLeadInput) -> str:
    """Crea un nuevo lead en el CRM. Devuelve el lead creado con su ID asignado."""
    payload = {
        "nombre": params.nombre,
        "email": params.email,
        "servicio_interes": params.servicio_interes,
        "status": "nuevo",
    }
    if params.presupuesto_eur: payload["presupuesto_eur"] = params.presupuesto_eur
    if params.notas: payload["notas"] = params.notas
    async with _get_client() as client:
        r = await client.post("/leads", json=payload)
        _raise_for_status(r)
        return r.text


@mcp.tool(
    name="cultiva_update_lead_status",
    annotations={"readOnlyHint": False, "idempotentHint": True},
)
async def cultiva_update_lead_status(params: UpdateStatusInput) -> str:
    """Actualiza el estado del pipeline de un lead. 
    Estados válidos: nuevo → contactado → propuesta → negociacion → cerrado/perdido"""
    payload = {"status": params.nuevo_status}
    if params.motivo: payload["motivo_cambio"] = params.motivo
    async with _get_client() as client:
        r = await client.patch(f"/leads/{params.lead_id}/status", json=payload)
        _raise_for_status(r)
        return r.text


@mcp.tool(
    name="cultiva_list_services",
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
async def cultiva_list_services(_: None = None) -> str:
    """Devuelve el catálogo completo de servicios ofrecidos por CULTIVA IA.
    Útil para completar el campo servicio_interes al crear leads."""
    async with _get_client() as client:
        r = await client.get("/services")
        _raise_for_status(r)
        return r.text


# ─── Entrypoint ───────────────────────────────────────────────
if __name__ == "__main__":
    mcp.run(transport="stdio")