File size: 3,186 Bytes
e969ea3 defb2ee e969ea3 defb2ee e969ea3 defb2ee 0f53d94 defb2ee 0f53d94 defb2ee 5c279ec 0f53d94 defb2ee e969ea3 defb2ee 5c279ec defb2ee 5c279ec defb2ee 5c279ec e969ea3 defb2ee |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
import os
import json
import sqlite3
from mcp.server.fastmcp import FastMCP
# HubSpot
from hubspot import HubSpot
from hubspot.crm.contacts import ApiException as HSContactsError
# ────────────────────────────────────────────────────────────────────────────
mcp = FastMCP("EnterpriseData")
# ─── 1) In-memory SQLite sample ─────────────────────────────────────────────
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 in-memory Customers table and return JSON rows."""
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)
except Exception as e:
return json.dumps({"error": str(e)})
# ─── 2) HubSpot tools ───────────────────────────────────────────────────────
HUBSPOT_TOKEN = os.getenv("HUBSPOT_TOKEN")
hs_client = HubSpot(access_token=HUBSPOT_TOKEN)
@mcp.tool()
def query_hubspot_contacts(limit: int = 100) -> str:
"""List HubSpot contacts."""
try:
results = hs_client.crm.contacts.basic_api.get_page(limit=limit).results
return json.dumps([r.to_dict() for r in results])
except HSContactsError as he:
return json.dumps({"error": he.body})
except Exception as e:
return json.dumps({"error": str(e)})
@mcp.tool()
def query_hubspot_companies(limit: int = 100) -> str:
"""List HubSpot companies."""
try:
results = hs_client.crm.companies.basic_api.get_page(limit=limit).results
return json.dumps([r.to_dict() for r in results])
except Exception as e:
return json.dumps({"error": str(e)})
@mcp.tool()
def query_hubspot_deals(limit: int = 100) -> str:
"""List HubSpot deals."""
try:
results = hs_client.crm.deals.basic_api.get_page(limit=limit).results
return json.dumps([r.to_dict() for r in results])
except Exception as e:
return json.dumps({"error": str(e)})
# ─── Final: Start the MCP server ────────────────────────────────────────────
if __name__ == "__main__":
mcp.run(transport="stdio")
|