File size: 2,444 Bytes
c3b2fc6 a6d22cd c3b2fc6 199706a 936bec2 c3b2fc6 f37945a c3b2fc6 f37945a c3b2fc6 f37945a 26d132c c3b2fc6 f37945a c3b2fc6 f37945a a6d22cd f37945a c3b2fc6 4cf3766 f37945a c3b2fc6 f37945a 199706a c3b2fc6 936bec2 a6d22cd c3b2fc6 f37945a c3b2fc6 f37945a |
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 |
"""
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")
|