mcp_server.py

"""
mcp_server.py – Minimal MCP tool server for:
  • query_database    – sample SQLite table
  • query_hubspot     – contacts, companies, deals from HubSpot

Env requirements:
  HUBSPOT_TOKEN
"""

import os, json, sqlite3
from mcp.server.fastmcp import FastMCP

# ─── HubSpot ─────────────────────────────
from hubspot import HubSpot
from hubspot.crm.contacts import ApiException as HSContactsError

hs_client = HubSpot(access_token=os.getenv("HUBSPOT_TOKEN"))

# ─── Setup MCP server ────────────────────
mcp = FastMCP("EnterpriseData")

# ─── In-memory SQLite setup ──────────────
conn = sqlite3.connect(":memory:", check_same_thread=False)
cur  = conn.cursor()

cur.execute("""
CREATE TABLE Customers (
    CustomerID      INTEGER PRIMARY KEY AUTOINCREMENT,
    Name            TEXT,
    Region          TEXT,
    LastOrderDate   TEXT
)
""")

cur.executemany(
    "INSERT INTO Customers (Name, Region, LastOrderDate) VALUES (?, ?, ?)",
    [
        ("Acme Corp",  "Northeast", "2024-12-01"),
        ("Beta Inc",   "West",      "2025-06-01"),
        ("Gamma Co",   "Northeast", "2023-09-15"),
        ("Delta LLC",  "South",     "2025-03-20"),
        ("Epsilon Ltd","Northeast", "2025-07-10"),
    ],
)
conn.commit()

@mcp.tool()
def query_database(sql: str) -> str:
    """Run SQL against the Customers table."""
    try:
        cur.execute(sql)
        cols = [d[0] for d in cur.description or []]
        rows = [dict(zip(cols, r)) for r in cur.fetchall()]
        return json.dumps(rows, indent=2)
    except Exception as e:
        return json.dumps({"error": str(e)})

@mcp.tool()
def query_hubspot(object_type: str = "contacts", limit: int = 100) -> str:
    """
    Fetch objects from HubSpot CRM (contacts, companies, deals).
    Example: object_type="contacts"
    """
    try:
        api = getattr(hs_client.crm, object_type)
        page = api.basic_api.get_page(limit=limit)
        items = [r.to_dict() for r in page.results]
        return json.dumps(items, indent=2)
    except HSContactsError as he:
        return json.dumps({"error": he.body})
    except Exception as e:
        return json.dumps({"error": str(e)})

# ─── Launch MCP server ───────────────────
if __name__ == "__main__":
    mcp.run(transport="stdio")