""" mcp_server.py – MCP tool server exposing: • query_database – SQLite sample • query_salesforce – Salesforce via simple-salesforce • query_hubspot – HubSpot via hubspot-api-client • query_quickbooks – QuickBooks Online via python-quickbooks • query_stripe – Stripe via stripe • query_googleads – Google Ads via google-ads • query_sharepoint – SharePoint via Office365-REST-Python-Client Secrets expected in env: SF_USER, SF_PASS, SF_TOKEN HUBSPOT_TOKEN QB_CLIENT_ID, QB_CLIENT_SECRET, QB_REFRESH_TOKEN, QB_REALM_ID STRIPE_API_KEY GOOGLE_ADS_DEVELOPER_TOKEN, GOOGLE_ADS_LOGIN_CUSTOMER_ID, GOOGLE_ADS_REFRESH_TOKEN, GOOGLE_ADS_CLIENT_ID, GOOGLE_ADS_CLIENT_SECRET SHAREPOINT_SITE_URL, SHAREPOINT_CLIENT_ID, SHAREPOINT_CLIENT_SECRET, SHAREPOINT_TENANT_ID """ import os import json import sqlite3 from mcp.server.fastmcp import FastMCP # Salesforce from simple_salesforce import Salesforce # HubSpot from hubspot import HubSpot from hubspot.crm.contacts import ApiException as HSContactsError # QuickBooks from quickbooks import QuickBooks from quickbooks.objects.customer import Customer # example object # Stripe import stripe # Google Ads from google.ads.googleads.client import GoogleAdsClient from google.ads.googleads.errors import GoogleAdsException # SharePoint from office365.runtime.auth.client_credential import ClientCredential from office365.sharepoint.client_context import ClientContext # ──────────────────────────────────────────────────────────────────────────── 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) Salesforce tool ───────────────────────────────────────────────────── sf = Salesforce( username=os.getenv("SF_USER"), password=os.getenv("SF_PASS"), security_token=os.getenv("SF_TOKEN"), ) @mcp.tool() def query_salesforce(soql: str) -> str: """ Run SOQL query via simple-salesforce. Example: SELECT Id, Name FROM Account WHERE Industry = 'Technology' """ try: result = sf.query_all(soql) records = result.get("records", []) # remove attributes block for r in records: r.pop("attributes", None) return json.dumps(records) except Exception as e: return json.dumps({"error": str(e)}) # ─── 3) HubSpot tool ──────────────────────────────────────────────────────── hs_client = HubSpot(access_token=os.getenv("HUBSPOT_TOKEN")) @mcp.tool() def query_hubspot(object_type: str, limit: int = 100) -> str: """ Fetch up to `limit` objects of type `contacts`, `companies`, or `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) except HSContactsError as he: return json.dumps({"error": he.body}) except Exception as e: return json.dumps({"error": str(e)}) # ─── 4) QuickBooks tool ──────────────────────────────────────────────────── qb = QuickBooks( client_id=os.getenv("QB_CLIENT_ID"), client_secret=os.getenv("QB_CLIENT_SECRET"), refresh_token=os.getenv("QB_REFRESH_TOKEN"), company_id=os.getenv("QB_REALM_ID"), ) @mcp.tool() def query_quickbooks(entity: str, max_results: int = 50) -> str: """ Query QuickBooks Online entity via python-quickbooks. entity="Customer" or "Invoice", etc. """ try: cls = globals().get(entity, None) if cls is None: raise ValueError(f"Unknown entity '{entity}'") objs = cls.where("", qb=qb, max_results=max_results) return json.dumps([o.to_dict() for o in objs]) except Exception as e: return json.dumps({"error": str(e)}) # ─── 5) Stripe tool ───────────────────────────────────────────────────────── stripe.api_key = os.getenv("STRIPE_API_KEY") @mcp.tool() def query_stripe(obj: str, limit: int = 10) -> str: """ List Stripe objects, e.g. obj="customers", "charges", "invoices" """ try: method = getattr(stripe, obj) items = method.list(limit=limit).data return json.dumps([i.to_dict() for i in items]) except Exception as e: return json.dumps({"error": str(e)}) # ─── 6) Google Ads tool ──────────────────────────────────────────────────── ga_config = { "developer_token": os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN"), "login_customer_id": os.getenv("GOOGLE_ADS_LOGIN_CUSTOMER_ID"), "refresh_token": os.getenv("GOOGLE_ADS_REFRESH_TOKEN"), "client_id": os.getenv("GOOGLE_ADS_CLIENT_ID"), "client_secret": os.getenv("GOOGLE_ADS_CLIENT_SECRET"), } ga_client = GoogleAdsClient.load_from_dict(ga_config) @mcp.tool() def query_googleads(customer_id: str, query: str) -> str: """ Run a GAQL query against a Google Ads customer. Example: query="SELECT campaign.id, campaign.name FROM campaign ORDER BY campaign.id" """ try: service = ga_client.get_service("GoogleAdsService") response = service.search(customer_id=customer_id, query=query) rows = [] for row in response: # convert protobuf row to dict rows.append({k: getattr(row, k) for k in row._pb.DESCRIPTOR.fields_by_name}) return json.dumps(rows) except GoogleAdsException as ge: return json.dumps({"error": ge.error.message}) except Exception as e: return json.dumps({"error": str(e)}) # ─── 7) SharePoint tool ──────────────────────────────────────────────────── site_url = os.getenv("SHAREPOINT_SITE_URL") client_id = os.getenv("SHAREPOINT_CLIENT_ID") client_secret = os.getenv("SHAREPOINT_CLIENT_SECRET") tenant_id = os.getenv("SHAREPOINT_TENANT_ID") auth_ctx = ClientContext(site_url).with_credentials( ClientCredential(client_id, client_secret) ) @mcp.tool() def query_sharepoint(list_title: str, top: int = 50) -> str: """ Fetch items from a SharePoint list. Example: list_title="ContactsList" """ try: sp_list = auth_ctx.web.lists.get_by_title(list_title) items = sp_list.items.top(top).get().execute_query() out = [dict(i.properties) for i in items] return json.dumps(out, ensure_ascii=False) except Exception as e: return json.dumps({"error": str(e)}) # ─── 8) Start the MCP server ─────────────────────────────────────────────── if __name__ == "__main__": mcp.run(transport="stdio")