#!/usr/bin/env python3 """ GHL MCP Helper — Drop-in functions for other agents/scripts to call GHL tools. Import this anywhere in the claudebotz system to access GHL data. Usage: from ghl_mcp_helper import ghl_get_contacts, ghl_get_pipelines, ghl_call_tool # Quick one-liners contacts = asyncio.run(ghl_get_contacts(limit=10)) pipelines = asyncio.run(ghl_get_pipelines()) # Generic tool call result = asyncio.run(ghl_call_tool("contacts_add-tags", { "path_contactId": "abc123", "body_tags": ["VIP", "New Lead"] })) """ import asyncio import sys import os # Add parent dir to path so we can import the client sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from ghl_mcp_client import GHLMCPClient # --------------------------------------------------------------------------- # Default config # --------------------------------------------------------------------------- PIT_TOKEN = os.getenv("GHL_MCP_PIT_TOKEN", os.environ.get("GHL_AGENCY_PIT", "")) LOCATION_ID = os.getenv("GHL_MCP_LOCATION_ID", "ifitVy09cFwlEVgEAFDS") async def ghl_call_tool(tool_name: str, arguments: dict = None) -> dict: """Generic: call any GHL MCP tool.""" client = GHLMCPClient(pit_token=PIT_TOKEN, location_id=LOCATION_ID) async with client.connect(): return await client.call_tool(tool_name, arguments or {}) async def ghl_list_tools() -> list[dict]: """List all available GHL MCP tools.""" client = GHLMCPClient(pit_token=PIT_TOKEN, location_id=LOCATION_ID) async with client.connect(): return await client.list_tools() # --------------------------------------------------------------------------- # Contacts # --------------------------------------------------------------------------- async def ghl_get_contacts(limit: int = 20, query: str = None) -> list: params = {"query_limit": str(limit), "query_locationId": LOCATION_ID} if query: params["query_query"] = query result = await ghl_call_tool("contacts_get-contacts", params) return result.get("results", {}).get("contacts", []) async def ghl_get_contact(contact_id: str) -> dict: result = await ghl_call_tool("contacts_get-contact", {"path_contactId": contact_id}) return result.get("results", {}).get("contact", result.get("results", {})) async def ghl_create_contact(first_name: str, last_name: str = "", email: str = "", phone: str = "", tags: list = None) -> dict: params = { "body_firstName": first_name, "body_locationId": LOCATION_ID, } if last_name: params["body_lastName"] = last_name if email: params["body_email"] = email if phone: params["body_phone"] = phone if tags: params["body_tags"] = tags return await ghl_call_tool("contacts_create-contact", params) async def ghl_add_tags(contact_id: str, tags: list[str]) -> dict: return await ghl_call_tool("contacts_add-tags", { "path_contactId": contact_id, "body_tags": tags }) async def ghl_remove_tags(contact_id: str, tags: list[str]) -> dict: return await ghl_call_tool("contacts_remove-tags", { "path_contactId": contact_id, "body_tags": tags }) # --------------------------------------------------------------------------- # Conversations # --------------------------------------------------------------------------- async def ghl_search_conversations(query: str = None, status: str = None) -> list: params = {"query_locationId": LOCATION_ID} if query: params["query_query"] = query if status: params["query_status"] = status result = await ghl_call_tool("conversations_search-conversation", params) return result.get("results", {}).get("conversations", []) async def ghl_send_message(conversation_id: str = None, contact_id: str = None, message: str = "", msg_type: str = "SMS") -> dict: params = {"body_type": msg_type, "body_message": message} if conversation_id: params["body_conversationId"] = conversation_id if contact_id: params["body_contactId"] = contact_id return await ghl_call_tool("conversations_send-a-new-message", params) # --------------------------------------------------------------------------- # Opportunities / Pipelines # --------------------------------------------------------------------------- async def ghl_get_pipelines() -> list: result = await ghl_call_tool("opportunities_get-pipelines", { "query_locationId": LOCATION_ID }) return result.get("results", {}).get("pipelines", []) async def ghl_search_opportunities(pipeline_id: str = None, status: str = "open") -> list: params = {"query_locationId": LOCATION_ID} if pipeline_id: params["query_pipelineId"] = pipeline_id if status: params["query_status"] = status result = await ghl_call_tool("opportunities_search-opportunity", params) return result.get("results", {}).get("opportunities", []) # --------------------------------------------------------------------------- # Payments # --------------------------------------------------------------------------- async def ghl_list_transactions(limit: int = 20) -> list: result = await ghl_call_tool("payments_list-transactions", { "query_locationId": LOCATION_ID, "query_limit": str(limit) }) return result.get("results", {}).get("data", []) # --------------------------------------------------------------------------- # Location # --------------------------------------------------------------------------- async def ghl_get_location() -> dict: result = await ghl_call_tool("locations_get-location", { "path_locationId": LOCATION_ID }) return result.get("results", {}) async def ghl_get_custom_fields() -> list: result = await ghl_call_tool("locations_get-custom-fields", { "path_locationId": LOCATION_ID }) return result.get("results", {}).get("customFields", []) # --------------------------------------------------------------------------- # Blogs # --------------------------------------------------------------------------- async def ghl_get_blogs() -> list: result = await ghl_call_tool("blogs_get-blogs", { "query_locationId": LOCATION_ID }) return result.get("results", {}).get("blogs", result.get("results", {}).get("data", [])) async def ghl_create_blog_post(title: str, content: str, blog_id: str, author_id: str = None, category_id: str = None) -> dict: params = { "body_title": title, "body_blogId": blog_id, "body_locationId": LOCATION_ID, "body_rawHTML": content, } if author_id: params["body_authorId"] = author_id if category_id: params["body_categoryId"] = category_id return await ghl_call_tool("blogs_create-blog-post", params) # --------------------------------------------------------------------------- # Social Media # --------------------------------------------------------------------------- async def ghl_get_social_accounts() -> list: result = await ghl_call_tool("social-media-posting_get-account", { "query_locationId": LOCATION_ID }) return result.get("results", {}).get("accounts", result.get("results", [])) async def ghl_get_social_posts(limit: int = 20) -> list: result = await ghl_call_tool("social-media-posting_get-posts", { "query_locationId": LOCATION_ID, "query_limit": str(limit) }) return result.get("results", {}).get("posts", []) # --------------------------------------------------------------------------- # Email Templates # --------------------------------------------------------------------------- async def ghl_get_email_templates() -> list: result = await ghl_call_tool("emails_fetch-template", { "query_locationId": LOCATION_ID }) return result.get("results", {}).get("templates", []) # --------------------------------------------------------------------------- # Quick test # --------------------------------------------------------------------------- if __name__ == "__main__": async def _test(): print("Testing GHL MCP Helper functions...") tools = await ghl_list_tools() print(f" Tools available: {len(tools)}") location = await ghl_get_location() print(f" Location data keys: {list(location.keys()) if isinstance(location, dict) else 'N/A'}") print(" All helpers ready!") asyncio.run(_test())