""" growthbase_mcp — Servidor MCP para GrowthBase CRM Generado con la skill constructor-servidores-mcp de CULTIVA IA

Permite a agentes LLM gestionar contactos, oportunidades y actividades del CRM GrowthBase sin salir del contexto de conversación.

Uso: export GROWTHBASE_API_KEY="your_api_key_here" python growthbase_mcp.py

Requisitos: pip install "mcp[cli]" httpx pydantic """

─────────────────────────────────────────────────────────────────────────────

IMPORTS

─────────────────────────────────────────────────────────────────────────────

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

─────────────────────────────────────────────────────────────────────────────

CONSTANTES

─────────────────────────────────────────────────────────────────────────────

API_BASE_URL = "https://api.growthbase.io/v2" CHARACTER_LIMIT = 20_000 MAX_RESULTS = 50

─────────────────────────────────────────────────────────────────────────────

SERVIDOR MCP

─────────────────────────────────────────────────────────────────────────────

mcp = FastMCP("growthbase_mcp")

─────────────────────────────────────────────────────────────────────────────

UTILIDADES COMPARTIDAS

─────────────────────────────────────────────────────────────────────────────

def _get_headers() -> dict: """Devuelve las cabeceras de autenticación para la API.""" api_key = os.environ.get("GROWTHBASE_API_KEY") if not api_key: raise ValueError( "Variable de entorno GROWTHBASE_API_KEY no encontrada. " "Configúrala con: export GROWTHBASE_API_KEY='tu_clave_api'" ) return { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json", }

async def _api_get(path: str, params: Optional[dict] = None) -> dict: """Realiza una petición GET a la API de GrowthBase.""" async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.get( f"{API_BASE_URL}{path}", headers=_get_headers(), params={k: v for k, v in (params or {}).items() if v is not None}, ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: status = e.response.status_code if status == 401: raise ValueError( "Autenticación fallida. Verifica tu GROWTHBASE_API_KEY. " "Obtén una nueva clave en app.growthbase.io/settings/api" ) elif status == 404: raise ValueError( f"Recurso no encontrado en {path}. " "Comprueba que el ID sea correcto con growthbase_search_contacts." ) elif status == 429: raise ValueError( "Límite de peticiones alcanzado. Espera 60 segundos antes de reintentar." ) raise ValueError(f"Error de API ({status}): {e.response.text[:500]}") except httpx.TimeoutException: raise ValueError( "Tiempo de espera agotado. La API de GrowthBase no respondió en 30s. " "Reintenta en unos momentos." )

async def _api_post(path: str, data: dict) -> dict: """Realiza una petición POST a la API de GrowthBase.""" async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.post( f"{API_BASE_URL}{path}", headers=_get_headers(), json=data, ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: status = e.response.status_code if status == 422: raise ValueError( f"Datos inválidos: {e.response.text[:300]}. " "Revisa que todos los campos requeridos estén presentes." ) raise ValueError(f"Error creando recurso ({status}): {e.response.text[:500]}")

async def _api_patch(path: str, data: dict) -> dict: """Realiza una petición PATCH a la API de GrowthBase.""" async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.patch( f"{API_BASE_URL}{path}", headers=_get_headers(), json=data, ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: raise ValueError(f"Error actualizando ({e.response.status_code}): {e.response.text[:500]}")

def _format_contact(c: dict, detailed: bool = False) -> str: """Formatea un contacto como texto legible para el agente.""" lines = [ f"{c.get('name', 'Sin nombre')} (ID: {c['id']})", f" Empresa: {c.get('company', '—')}", f" Email: {c.get('email', '—')} | Tel: {c.get('phone', '—')}", f" Estado: {c.get('status', '—')} | Propietario: {c.get('owner_name', '—')}", f" Tags: {', '.join(c.get('tags', [])) or '—'}", ] if detailed: lines += [ f" Fuente: {c.get('source', '—')}", f" Creado: {c.get('created_at', '—')}", f" Última actividad: {c.get('last_activity_at', '—')}", f" Notas: {c.get('notes', '—')[:200] if c.get('notes') else '—'}", ] return "\n".join(lines)

def _format_deal(d: dict) -> str: """Formatea una oportunidad como texto legible.""" return ( f"{d.get('name', 'Sin nombre')} (ID: {d['id']})\n" f" Contacto: {d.get('contact_name', '—')} | Empresa: {d.get('company', '—')}\n" f" Etapa: {d.get('stage', '—')} | Valor: {d.get('value', 0):,.0f} €\n" f" Propietario: {d.get('owner_name', '—')} | Cierre estimado: {d.get('close_date', '—')}\n" f" Probabilidad: {d.get('probability', 0)}%" )

─────────────────────────────────────────────────────────────────────────────

MODELOS DE ENTRADA (PYDANTIC v2)

─────────────────────────────────────────────────────────────────────────────

class ContactStatus(str, Enum): lead = "lead" prospect = "prospect" customer = "customer" churned = "churned"

class DealStage(str, Enum): prospecting = "prospecting" qualified = "qualified" proposal = "proposal" negotiation = "negotiation" closed_won = "closed_won" closed_lost = "closed_lost"

class ActivityType(str, Enum): call = "call" email = "email" meeting = "meeting" note = "note"

class ResponseFormat(str, Enum): markdown = "markdown" json = "json"

class SearchContactsInput(BaseModel): model_config = ConfigDict(strict=True) query: Optional[str] = Field( None, description="Texto a buscar en nombre, empresa o email. Ej: 'María López', 'Zara', 'mlopez@'" ) status: Optional[ContactStatus] = Field( None, description="Filtrar por estado del contacto: lead, prospect, customer, churned" ) tag: Optional[str] = Field( None, description="Filtrar por tag. Ej: 'ecommerce', 'agencia', 'vip'" ) owner: Optional[str] = Field( None, description="Filtrar por nombre o email del comercial propietario" ) limit: int = Field( 20, ge=1, le=MAX_RESULTS, description="Número máximo de resultados (1-50). Por defecto: 20" ) format: ResponseFormat = Field( ResponseFormat.markdown, description="Formato de respuesta: 'markdown' (legible) o 'json' (estructurado)" )

class GetContactInput(BaseModel): model_config = ConfigDict(strict=True) contact_id: str = Field( ..., description="ID único del contacto. Obtén IDs con growthbase_search_contacts. Ej: 'cnt_abc123'" ) format: ResponseFormat = Field( ResponseFormat.markdown, description="Formato de respuesta: 'markdown' (legible) o 'json' (estructurado)" )

class CreateContactInput(BaseModel): model_config = ConfigDict(strict=True) name: str = Field(..., min_length=2, max_length=100, description="Nombre completo del contacto") email: str = Field(..., description="Email del contacto. Ej: 'juan@empresa.com'") company: Optional[str] = Field(None, max_length=100, description="Nombre de la empresa") phone: Optional[str] = Field(None, description="Teléfono con prefijo. Ej: '+34 612 345 678'") status: ContactStatus = Field(ContactStatus.lead, description="Estado inicial: lead, prospect, customer") tags: Optional[List[str]] = Field(None, description="Lista de tags. Ej: ['ecommerce', 'pyme']") notes: Optional[str] = Field(None, max_length=1000, description="Notas internas sobre el contacto") source: Optional[str] = Field(None, description="Origen del lead. Ej: 'web', 'referral', 'linkedin'")

class ListDealsInput(BaseModel): model_config = ConfigDict(strict=True) stage: Optional[DealStage] = Field( None, description="Filtrar por etapa del pipeline: prospecting, qualified, proposal, negotiation, closed_won, closed_lost" ) owner: Optional[str] = Field(None, description="Filtrar por comercial propietario") min_value: Optional[float] = Field(None, ge=0, description="Valor mínimo de la oportunidad en euros") limit: int = Field(20, ge=1, le=MAX_RESULTS, description="Número máximo de resultados (1-50)") format: ResponseFormat = Field(ResponseFormat.markdown, description="Formato: 'markdown' o 'json'")

class MoveDealInput(BaseModel): model_config = ConfigDict(strict=True) deal_id: str = Field(..., description="ID de la oportunidad. Ej: 'deal_xyz789'") stage: DealStage = Field(..., description="Nueva etapa destino del pipeline") probability: Optional[int] = Field( None, ge=0, le=100, description="Probabilidad de cierre en % (0-100). Si no se indica, se usa el valor por defecto de la etapa" ) close_date: Optional[str] = Field( None, description="Nueva fecha estimada de cierre en formato YYYY-MM-DD. Ej: '2026-09-30'" ) notes: Optional[str] = Field(None, max_length=500, description="Notas sobre el movimiento de etapa")

class LogActivityInput(BaseModel): model_config = ConfigDict(strict=True) contact_id: str = Field(..., description="ID del contacto relacionado con la actividad") type: ActivityType = Field(..., description="Tipo: call, email, meeting, note") summary: str = Field( ..., min_length=5, max_length=500, description="Resumen de la actividad. Ej: 'Llamada de 15min. Interesado en plan Pro. Callback el jueves'" ) deal_id: Optional[str] = Field(None, description="ID de la oportunidad relacionada (opcional)") duration_minutes: Optional[int] = Field( None, ge=1, le=480, description="Duración en minutos (solo para calls y meetings)" ) outcome: Optional[str] = Field( None, description="Resultado: 'interested', 'not_interested', 'callback', 'no_answer', 'demo_booked'" )

class PipelineSummaryInput(BaseModel): model_config = ConfigDict(strict=True) owner: Optional[str] = Field(None, description="Filtrar resumen por comercial. Si es None, muestra el equipo completo") format: ResponseFormat = Field(ResponseFormat.markdown, description="Formato: 'markdown' o 'json'")

─────────────────────────────────────────────────────────────────────────────

HERRAMIENTAS MCP

─────────────────────────────────────────────────────────────────────────────

@mcp.tool( name="growthbase_search_contacts", annotations={ "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, } ) async def growthbase_search_contacts(params: SearchContactsInput) -> str: """ Busca contactos en GrowthBase CRM por nombre, empresa, email o filtros.

Usa esta herramienta para:
- Encontrar el ID de un contacto antes de otras operaciones
- Listar leads de una empresa o con un tag específico
- Revisar la cartera de un comercial

No uses esta herramienta para obtener el detalle completo de un contacto;
usa growthbase_get_contact con el ID obtenido aquí.

Returns:
    Lista formateada de contactos con nombre, empresa, email, estado y tags.
    Si no hay resultados, sugiere ampliar los filtros.
"""
data = await _api_get("/contacts", params={
    "q": params.query,
    "status": params.status.value if params.status else None,
    "tag": params.tag,
    "owner": params.owner,
    "limit": params.limit,
})

contacts = data.get("contacts", [])
total = data.get("total", len(contacts))

if not contacts:
    return (
        "No se encontraron contactos con los filtros indicados.\n"
        "Sugerencias:\n"
        "- Prueba sin filtro de status para ampliar la búsqueda\n"
        "- Verifica la ortografía del nombre o empresa\n"
        "- Usa growthbase_search_contacts con query=None para ver todos los contactos"
    )

if params.format == ResponseFormat.json:
    import json
    return json.dumps({"total": total, "contacts": contacts}, ensure_ascii=False, indent=2)[:CHARACTER_LIMIT]

lines = [f"## Contactos encontrados ({len(contacts)} de {total} total)\n"]
for c in contacts:
    lines.append(_format_contact(c))
    lines.append("")
if total > len(contacts):
    lines.append(f"*Mostrando {len(contacts)} de {total}. Usa limit={min(total, MAX_RESULTS)} o añade filtros para refinar.*")

result = "\n".join(lines)
return result[:CHARACTER_LIMIT]

@mcp.tool( name="growthbase_get_contact", annotations={ "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, } ) async def growthbase_get_contact(params: GetContactInput) -> str: """ Obtiene el perfil completo de un contacto por su ID.

Incluye historial de actividades recientes, oportunidades abiertas,
notas y toda la información de contacto.

Usa growthbase_search_contacts primero si no conoces el ID del contacto.

Args:
    contact_id: ID único del contacto (formato: cnt_XXXXXX)

Returns:
    Perfil detallado con datos personales, historial y oportunidades vinculadas.
"""
data = await _api_get(f"/contacts/{params.contact_id}")
contact = data.get("contact", data)

if params.format == ResponseFormat.json:
    import json
    return json.dumps(contact, ensure_ascii=False, indent=2)[:CHARACTER_LIMIT]

# Formato markdown enriquecido
lines = [
    f"# Contacto: {contact.get('name', '—')}",
    "",
    _format_contact(contact, detailed=True),
    "",
]

# Oportunidades vinculadas
deals = contact.get("deals", [])
if deals:
    lines.append(f"## Oportunidades ({len(deals)})")
    for d in deals[:5]:
        lines.append(_format_deal(d))
        lines.append("")

# Últimas actividades
activities = contact.get("recent_activities", [])
if activities:
    lines.append(f"## Actividades recientes ({len(activities)})")
    for a in activities[:5]:
        lines.append(
            f"- [{a.get('type', '?').upper()}] {a.get('date', '—')} — {a.get('summary', '—')[:120]}"
        )

result = "\n".join(lines)
return result[:CHARACTER_LIMIT]

@mcp.tool( name="growthbase_create_contact", annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True, } ) async def growthbase_create_contact(params: CreateContactInput) -> str: """ Crea un nuevo contacto/lead en GrowthBase CRM.

Usa esta herramienta cuando:
- Recibes datos de un nuevo lead (formulario, email, llamada)
- Quieres registrar un contacto antes de crear una oportunidad

No duplicar: usa growthbase_search_contacts antes para verificar
que el contacto no existe ya en el CRM.

Returns:
    Confirmación con el ID del contacto creado y un resumen de sus datos.
"""
payload = {
    "name": params.name,
    "email": params.email,
    "status": params.status.value,
}
if params.company:
    payload["company"] = params.company
if params.phone:
    payload["phone"] = params.phone
if params.tags:
    payload["tags"] = params.tags
if params.notes:
    payload["notes"] = params.notes
if params.source:
    payload["source"] = params.source

data = await _api_post("/contacts", payload)
contact = data.get("contact", data)

return (
    f"Contacto creado correctamente.\n\n"
    f"{_format_contact(contact, detailed=True)}\n\n"
    f"Próximos pasos sugeridos:\n"
    f"- Crea una oportunidad con growthbase_list_deals\n"
    f"- Registra la primera actividad con growthbase_log_activity(contact_id='{contact.get('id', '?')}', type='note', ...)"
)

@mcp.tool( name="growthbase_list_deals", annotations={ "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, } ) async def growthbase_list_deals(params: ListDealsInput) -> str: """ Lista las oportunidades del pipeline de ventas con filtros opcionales.

Ideal para revisar el estado del pipeline, ver qué oportunidades
están bloqueadas en una etapa o identificar los deals de mayor valor.

Para mover una oportunidad de etapa usa growthbase_move_deal.

Returns:
    Lista de oportunidades con etapa, valor, probabilidad y fecha de cierre.
"""
data = await _api_get("/deals", params={
    "stage": params.stage.value if params.stage else None,
    "owner": params.owner,
    "min_value": params.min_value,
    "limit": params.limit,
})

deals = data.get("deals", [])
total = data.get("total", len(deals))

if not deals:
    return (
        "No se encontraron oportunidades con los filtros indicados.\n"
        "Prueba sin filtro de stage para ver todo el pipeline."
    )

if params.format == ResponseFormat.json:
    import json
    return json.dumps({"total": total, "deals": deals}, ensure_ascii=False, indent=2)[:CHARACTER_LIMIT]

total_value = sum(d.get("value", 0) for d in deals)
lines = [
    f"## Pipeline de ventas ({len(deals)} oportunidades | Valor total: {total_value:,.0f} €)\n"
]
for d in deals:
    lines.append(_format_deal(d))
    lines.append("")

if total > len(deals):
    lines.append(f"*Mostrando {len(deals)} de {total}. Añade filtros para refinar.*")

return "\n".join(lines)[:CHARACTER_LIMIT]

@mcp.tool( name="growthbase_move_deal", annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True, } ) async def growthbase_move_deal(params: MoveDealInput) -> str: """ Mueve una oportunidad a una nueva etapa del pipeline de ventas.

Etapas en orden: prospecting → qualified → proposal → negotiation → closed_won / closed_lost

Registra automáticamente una actividad de cambio de etapa en el historial.
Si no conoces el deal_id, usa primero growthbase_list_deals para encontrarlo.

Returns:
    Confirmación del cambio con el estado actualizado de la oportunidad.
"""
payload: dict = {"stage": params.stage.value}
if params.probability is not None:
    payload["probability"] = params.probability
if params.close_date:
    payload["close_date"] = params.close_date
if params.notes:
    payload["stage_notes"] = params.notes

data = await _api_patch(f"/deals/{params.deal_id}", payload)
deal = data.get("deal", data)

return (
    f"Oportunidad movida correctamente.\n\n"
    f"{_format_deal(deal)}\n\n"
    f"Cambio registrado en el historial de actividades."
)

@mcp.tool( name="growthbase_log_activity", annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True, } ) async def growthbase_log_activity(params: LogActivityInput) -> str: """ Registra una actividad (llamada, email, reunión o nota) para un contacto.

Úsalo inmediatamente después de cualquier interacción con un contacto
para mantener el CRM actualizado y el historial completo.

El contacto debe existir en el CRM. Si no existe, primero usa
growthbase_create_contact para registrarlo.

Returns:
    Confirmación de la actividad registrada con su ID y resumen.
"""
payload = {
    "contact_id": params.contact_id,
    "type": params.type.value,
    "summary": params.summary,
}
if params.deal_id:
    payload["deal_id"] = params.deal_id
if params.duration_minutes is not None:
    payload["duration_minutes"] = params.duration_minutes
if params.outcome:
    payload["outcome"] = params.outcome

data = await _api_post("/activities", payload)
activity = data.get("activity", data)

type_emoji = {
    "call": "📞",
    "email": "📧",
    "meeting": "🤝",
    "note": "📝",
}.get(params.type.value, "•")

return (
    f"Actividad registrada (ID: {activity.get('id', '?')})\n\n"
    f"{type_emoji} **{params.type.value.upper()}** — {activity.get('created_at', 'Ahora')}\n"
    f"Contacto: {activity.get('contact_name', params.contact_id)}\n"
    f"Resumen: {params.summary}\n"
    + (f"Resultado: {params.outcome}\n" if params.outcome else "")
    + (f"Duración: {params.duration_minutes} min\n" if params.duration_minutes else "")
)

@mcp.tool( name="growthbase_pipeline_summary", annotations={ "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, } ) async def growthbase_pipeline_summary(params: PipelineSummaryInput) -> str: """ Genera un resumen ejecutivo del pipeline de ventas con totales por etapa.

Muestra el valor total, número de deals y conversiones entre etapas.
Ideal para reuniones de seguimiento o para que el agente entienda
el estado global de las ventas antes de hacer recomendaciones.

Returns:
    Resumen por etapa con valor acumulado, número de deals y tasa de conversión.
"""
api_params = {}
if params.owner:
    api_params["owner"] = params.owner

data = await _api_get("/pipeline/summary", api_params)

if params.format == ResponseFormat.json:
    import json
    return json.dumps(data, ensure_ascii=False, indent=2)[:CHARACTER_LIMIT]

stages = data.get("stages", [])
total_pipeline = data.get("total_pipeline_value", 0)
won_this_month = data.get("closed_won_this_month", 0)
conversion_rate = data.get("overall_conversion_rate", 0)
owner_label = f" — {params.owner}" if params.owner else " — Equipo completo"

lines = [
    f"# Resumen del Pipeline{owner_label}",
    "",
    f"**Valor total en pipeline:** {total_pipeline:,.0f} €",
    f"**Cerrado este mes:** {won_this_month:,.0f} €",
    f"**Tasa de conversión global:** {conversion_rate:.1f}%",
    "",
    "## Por etapa",
    "",
]

stage_names = {
    "prospecting": "Prospección",
    "qualified": "Cualificado",
    "proposal": "Propuesta",
    "negotiation": "Negociación",
    "closed_won": "Ganado",
    "closed_lost": "Perdido",
}

for s in stages:
    name = stage_names.get(s.get("stage", ""), s.get("stage", "—"))
    count = s.get("count", 0)
    value = s.get("value", 0)
    avg = value / count if count else 0
    lines.append(
        f"| **{name}** | {count} deals | {value:,.0f} € | Ticket medio: {avg:,.0f} € |"
    )

lines += [
    "",
    "---",
    "*Usa growthbase_list_deals(stage='...') para ver los deals de una etapa específica.*",
]

return "\n".join(lines)[:CHARACTER_LIMIT]

─────────────────────────────────────────────────────────────────────────────

EVALUACIONES (Fase 4 — XML de evaluación para testing del servidor)

─────────────────────────────────────────────────────────────────────────────

EVALUATION_XML = """ Busca todos los contactos con status 'prospect' y tag 'ecommerce'. ¿Cuántos hay en total? Depende de los datos reales; herramienta: growthbase_search_contacts(status='prospect', tag='ecommerce') ¿Cuál es el valor total del pipeline en etapa 'proposal' según el resumen ejecutivo? Herramienta: growthbase_pipeline_summary() → campo stages[stage='proposal'].value Registra una llamada de 20 minutos para el contacto cnt_001 con resultado 'callback'. growthbase_log_activity(contact_id='cnt_001', type='call', duration_minutes=20, outcome='callback', summary='...') Mueve la oportunidad deal_456 a etapa 'negotiation' con probabilidad 70% y cierre estimado 2026-09-30. growthbase_move_deal(deal_id='deal_456', stage='negotiation', probability=70, close_date='2026-09-30') Crea un nuevo lead: María García, mgarcia@techstartup.es, empresa TechStartup SL, fuente 'linkedin'. growthbase_create_contact(name='María García', email='mgarcia@techstartup.es', company='TechStartup SL', source='linkedin') """

─────────────────────────────────────────────────────────────────────────────

PUNTO DE ENTRADA

─────────────────────────────────────────────────────────────────────────────

if name == "main": mcp.run()