Two Ways to Run a Peliqan MCP Server

Claude Desktop (stdio) ──────────────┐
Cursor IDE (stdio) ──────────────────┤
n8n AI Agent (HTTP) ─────────────────┤──> mcp-server-peliqan (local proxy)
Custom chatbot (HTTP/SSE) ───────────┘

POST /mcp ──> API Handler Data App (Python)
├── get_customer_health()   pq.read_sql() on DWH
├── get_deals_at_risk()    pq.connect(‘Salesforce’)
├── search_contacts()      Trino federated query
└── add_crm_note()         pq.connect(‘HubSpot’).add()

# Peliqan MCP API Handler
# Route: POST /mcp   (configured in the API endpoint settings)
# pq module is auto-available - all connected sources accessible via pq.connect()

import json

# Peliqan calls handle_request() with the parsed MCP request body
def handle_request(request):
    method = request.get("method")

    # ── MCP handshake: tell clients what tools are available ──────────
    if method == "tools/list":
        return {
            "tools": [
                {
                    "name": "get_customer_health",
                    "description": "Returns the current health score, MRR, and churn segment for a customer",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "email": {"type": "string", "description": "Customer email address"}
                        },
                        "required": ["email"]
                    }
                },
                {
                    "name": "get_deals_at_risk",
                    "description": "Lists Salesforce deals with no activity in the last N days",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "days_inactive": {"type": "integer", "description": "Inactivity threshold in days", "default": 14},
                            "min_value":     {"type": "number",  "description": "Minimum deal value in USD"}
                        }
                    }
                },
                {
                    "name": "search_contacts",
                    "description": "Searches contacts by name or email across HubSpot and the data warehouse",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Name or email fragment to search for"}
                        },
                        "required": ["query"]
                    }
                },
                {
                    "name": "add_crm_note",
                    "description": "Adds a note to a HubSpot contact record",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "email":   {"type": "string", "description": "Contact email address"},
                            "note":    {"type": "string", "description": "Note content to add"}
                        },
                        "required": ["email", "note"]
                    }
                }
            ]
        }

    # ── Tool execution ────────────────────────────────────────────────
    if method == "tools/call":
        tool_name = request["params"]["name"]
        args      = request["params"].get("arguments", {})
        return dispatch(tool_name, args)

    return {"error": f"Unknown method: {method}"}


def dispatch(tool_name, args):
    if tool_name == "get_customer_health":
        return get_customer_health(**args)
    elif tool_name == "get_deals_at_risk":
        return get_deals_at_risk(**args)
    elif tool_name == "search_contacts":
        return search_contacts(**args)
    elif tool_name == "add_crm_note":
        return add_crm_note(**args)
    return {"error": f"Unknown tool: {tool_name}"}

def get_customer_health(email: str):
    """Query the customer_health_score view built in the ELT pipeline."""
    df = pq.read_sql(f"""
        SELECT
            email,
            health_score,
            health_segment,
            mrr,
            active_days,
            features_used,
            scored_at
        FROM customer_health_score
        WHERE LOWER(email) = LOWER('{email}')
        ORDER BY scored_at DESC
        LIMIT 1
    """)

    if df.empty:
        return {"content": [{"type": "text", "text": f"No health score found for {email}"}]}

    row = df.iloc[0]
    summary = (
        f"Customer: {row['email']}\n"
        f"Health score: {row['health_score']:.1f} / 100  ({row['health_segment'].replace('_', ' ').title()})\n"
        f"MRR: ${row['mrr']:,.0f}\n"
        f"Active days (last 30): {int(row['active_days'])}\n"
        f"Features used: {int(row['features_used'])}\n"
        f"Last scored: {row['scored_at'].strftime('%Y-%m-%d')}"
    )
    return {"content": [{"type": "text", "text": summary}]}

def get_deals_at_risk(days_inactive: int = 14, min_value: float = 0):
    """Fetch Salesforce opportunities with no recent activity."""
    sf = pq.connect('Salesforce')

    # Use Peliqan's Salesforce connector to query opportunities
    deals = sf.query("""
        SELECT Id, Name, Amount, StageName, LastActivityDate, Account.Name
        FROM Opportunity
        WHERE IsClosed = false
          AND (LastActivityDate = null OR LastActivityDate < LAST_N_DAYS:{days}) ORDER BY Amount DESC NULLS LAST LIMIT 25 """.replace("{days}", str(days_inactive))) if not deals: return {"content": [{"type": "text", "text": f"No at-risk deals found (inactive {days_inactive}+ days)"}]} filtered = [d for d in deals if (d.get('Amount') or 0) >= min_value]

    lines = [f"Deals inactive for {days_inactive}+ days (min value ${min_value:,.0f}):\n"]
    for d in filtered[:10]:
        last_activity = d.get('LastActivityDate') or 'Never'
        lines.append(
            f"- {d['Name']} ({d['Account']['Name'] if d.get('Account') else 'Unknown'})\n"
            f"  Stage: {d['StageName']}  |  Value: ${d.get('Amount', 0):,.0f}  |  Last activity: {last_activity}"
        )

    return {"content": [{"type": "text", "text": "\n".join(lines)}]}

def search_contacts(query: str):
    """
    Search contacts across HubSpot and the warehouse in a single Trino query.
    Trino lets us JOIN live sources without moving data - no prior sync needed.
    """
    df = pq.read_sql(f"""
        SELECT
            h.email,
            h.firstname || ' ' || h.lastname   AS full_name,
            h.company,
            h.phone,
            h.hs_lead_status                    AS lead_status,
            hs.health_score,
            hs.health_segment
        FROM hubspot.contacts h
        LEFT JOIN customer_health_score hs ON LOWER(h.email) = LOWER(hs.email)
        WHERE
            LOWER(h.email)     LIKE LOWER('%{query}%')
            OR LOWER(h.firstname || ' ' || h.lastname) LIKE LOWER('%{query}%')
            OR LOWER(h.company) LIKE LOWER('%{query}%')
        ORDER BY hs.health_score DESC NULLS LAST
        LIMIT 10
    """)

    if df.empty:
        return {"content": [{"type": "text", "text": f"No contacts found matching '{query}'"}]}

    lines = [f"Contacts matching '{query}':\n"]
    for _, row in df.iterrows():
        score_str = f"{row['health_score']:.0f}/100 ({row['health_segment']})" if row['health_score'] else "no score"
        lines.append(
            f"- {row['full_name']} <{row['email']}>\n"
            f"  Company: {row['company'] or 'N/A'}  |  Status: {row['lead_status'] or 'N/A'}  |  Health: {score_str}"
        )

    return {"content": [{"type": "text", "text": "\n".join(lines)}]}

def add_crm_note(email: str, note: str):
    """Write a note back to a HubSpot contact - from inside the AI chat."""
    hs = pq.connect('HubSpot')

    # Look up the contact ID first
    contact = hs.get('contact', filter={'email': email})

    if not contact:
        return {"content": [{"type": "text", "text": f"Contact not found for email: {email}"}]}

    # Add the note via Peliqan's HubSpot writeback connector
    hs.add('note', {
        'associations': [{'id': contact['id'], 'type': 'contact'}],
        'body':         note,
        'timestamp':    pq.now_ms()
    })

    pq.log(f"Note added to HubSpot contact {email} via MCP")
    return {"content": [{"type": "text", "text": f"Note added to {email}'s HubSpot record."}]}

⚠️ Treat Writeback Tools with Care

Any tool that modifies data – adding notes, updating fields, sending messages – should be scoped carefully. In Claude Desktop, the AI will ask for confirmation before calling a writeback tool. In automated agent pipelines (n8n, LangChain), add explicit approval steps. Peliqan’s run logs capture every tool invocation with timestamp, arguments, and result – so you always have a full audit trail.

pip install mcp-server-peliqan

{
  "mcpServers": {
    "peliqan": {
      "command": "python",
      "args": ["-m", "mcp_server_peliqan"],
      "env": {
        "peliqan_account_id": "1234",
        "peliqan_api_token":  "your-api-token-here"
      }
    }
  }
}

🔗 Connecting Other MCP Clients

💬 Example Conversations with Claude

🔒 Security and Governance Checklist

🛠 More MCP Tools You Can Add in Minutes

• get_invoice_status – query open invoices from your accounting system (Xero, QuickBooks, Exact)
• get_support_tickets – pull open Zendesk or Freshdesk tickets for a customer
• run_sql – a general-purpose tool that executes any approved SQL query against your warehouse
• create_deal – create a new Salesforce opportunity from inside a conversation
• send_slack_message – post to a Slack channel from the AI agent
• semantic_search – RAG search over your unstructured data using Peliqan’s built-in vector store