Skip to content

Folders and files

NameName
Last commit message
Last commit date

Latest commit

History

17 Commits

Repository files navigation

Go High Level API Integration Template

A comprehensive template for integrating with the Go High Level API, providing pre-implemented functions for all available endpoints. This template simplifies the integration process for developers looking to work with Go High Level's CRM, marketing, and business automation platform.

Overview

This template provides ready-to-use functions for all Go High Level API endpoints, allowing developers to quickly implement Go High Level integrations in their applications. The template handles authentication, request formatting, and error handling, making it easy to interact with the Go High Level ecosystem.

Features

  • Complete endpoint coverage for Go High Level API
  • Authentication handling with API keys and OAuth
  • Proper error handling and response parsing
  • Pagination support for list endpoints
  • Webhook verification and processing

Installation

  1. Clone this repository:

    git clone https://github.com/yourusername/go-high-level-template.git
    cd go-high-level-template
  2. Set up your API credentials:

    # Set environment variablesexport GHL_API_KEY="your_api_key"export GHL_LOCATION_ID="your_location_id"# If applicableexport GHL_COMPANY_ID="your_company_id"# If applicable

Authentication

The template supports both API key authentication and OAuth 2.0:

API Key Authentication

# Import the GHL clientfromghl_clientimportGHLClient# Initialize with API keyclient=GHLClient(api_key="your_api_key")
# Or initialize with environment variablesclient=GHLClient()

OAuth 2.0 Authentication

# Initialize with OAuth credentialsclient=GHLClient(
client_id="your_client_id",
client_secret="your_client_secret",
redirect_uri="your_redirect_uri"
)
# Generate authorization URLauth_url=client.get_authorization_url()
# Exchange code for tokenstokens=client.exchange_code_for_tokens("authorization_code")
# Use refresh token to get new access tokennew_tokens=client.refresh_access_token("refresh_token")

Available Endpoints

Contacts

# Create a contactcontact=client.contacts.create({
"email": "john@example.com",
"firstName": "John",
"lastName": "Doe",
"phone": "+15551234567",
"customFields": [
{"id": "custom_field_id", "value": "custom value"}
]
})
# Get all contactscontacts=client.contacts.list(limit=100, page=1)
# Get contact by IDcontact=client.contacts.get("contact_id")
# Update contactupdated_contact=client.contacts.update("contact_id", {
"firstName": "Johnny"
})
# Delete contactclient.contacts.delete("contact_id")
# Search contactssearch_results=client.contacts.search({
"query": "john",
"filters": [
{"field": "tags", "operator": "contains", "value": "customer"}
]
})

Opportunities

# Create an opportunityopportunity=client.opportunities.create({
"name": "New Deal",
"contactId": "contact_id",
"pipelineId": "pipeline_id",
"stageId": "stage_id",
"amount": 5000,
"status": "open"
})
# Get all opportunitiesopportunities=client.opportunities.list(limit=100, page=1)
# Get opportunity by IDopportunity=client.opportunities.get("opportunity_id")
# Update opportunityupdated_opportunity=client.opportunities.update("opportunity_id", {
"amount": 7500,
"status": "won"
})
# Delete opportunityclient.opportunities.delete("opportunity_id")

Pipelines

# Get all pipelinespipelines=client.pipelines.list()
# Get pipeline by IDpipeline=client.pipelines.get("pipeline_id")
# Get pipeline stagesstages=client.pipelines.get_stages("pipeline_id")

Appointments

# Create an appointmentappointment=client.appointments.create({
"title": "Strategy Meeting",
"description": "Discuss marketing strategy",
"contactId": "contact_id",
"startTime": "2023-12-15T10:00:00Z",
"endTime": "2023-12-15T11:00:00Z",
"calendarId": "calendar_id"
})
# Get all appointmentsappointments=client.appointments.list(
startDate="2023-12-01",
endDate="2023-12-31",
limit=100,
page=1
)
# Get appointment by IDappointment=client.appointments.get("appointment_id")
# Update appointmentupdated_appointment=client.appointments.update("appointment_id", {
"title": "Updated Meeting",
"description": "Updated description"
})
# Cancel appointmentclient.appointments.cancel("appointment_id")

Tasks

# Create a tasktask=client.tasks.create({
"title": "Follow up with client",
"description": "Call to discuss proposal",
"contactId": "contact_id",
"dueDate": "2023-12-20",
"assignedTo": "user_id",
"status": "not_started"
})
# Get all taskstasks=client.tasks.list(limit=100, page=1)
# Get task by IDtask=client.tasks.get("task_id")
# Update taskupdated_task=client.tasks.update("task_id", {
"status": "completed"
})
# Delete taskclient.tasks.delete("task_id")

Calendars

# Get all calendarscalendars=client.calendars.list()
# Get calendar by IDcalendar=client.calendars.get("calendar_id")
# Get calendar availabilityavailability=client.calendars.get_availability("calendar_id", {
"startDate": "2023-12-01",
"endDate": "2023-12-31"
})

Forms

# Get all formsforms=client.forms.list()
# Get form by IDform=client.forms.get("form_id")
# Get form submissionssubmissions=client.forms.get_submissions("form_id", limit=100, page=1)

Campaigns

# Create a campaigncampaign=client.campaigns.create({
"name": "Holiday Promotion",
"type": "email",
"status": "draft"
})
# Get all campaignscampaigns=client.campaigns.list(limit=100, page=1)
# Get campaign by IDcampaign=client.campaigns.get("campaign_id")
# Update campaignupdated_campaign=client.campaigns.update("campaign_id", {
"name": "Updated Campaign Name"
})
# Delete campaignclient.campaigns.delete("campaign_id")
# Add contacts to campaignclient.campaigns.add_contacts("campaign_id", ["contact_id1", "contact_id2"])
# Remove contacts from campaignclient.campaigns.remove_contacts("campaign_id", ["contact_id1"])

Workflows

# Get all workflowsworkflows=client.workflows.list()
# Get workflow by IDworkflow=client.workflows.get("workflow_id")
# Enroll contacts in workflowclient.workflows.enroll_contacts("workflow_id", ["contact_id1", "contact_id2"])
# Remove contacts from workflowclient.workflows.remove_contacts("workflow_id", ["contact_id1"])

Conversations

# Get all conversationsconversations=client.conversations.list(limit=100, page=1)
# Get conversation by IDconversation=client.conversations.get("conversation_id")
# Send messagemessage=client.conversations.send_message({
"conversationId": "conversation_id",
"body": "Hello, how can I help you today?",
"attachments": []
})
# Get conversation messagesmessages=client.conversations.get_messages("conversation_id", limit=100, page=1)

Users

# Get all usersusers=client.users.list()
# Get user by IDuser=client.users.get("user_id")
# Create a usernew_user=client.users.create({
"firstName": "Jane",
"lastName": "Smith",
"email": "jane@example.com",
"role": "admin"
})
# Update userupdated_user=client.users.update("user_id", {
"firstName": "Janet"
})
# Delete userclient.users.delete("user_id")

Locations

# Get all locationslocations=client.locations.list()
# Get location by IDlocation=client.locations.get("location_id")

Custom Fields

# Get all custom fieldscustom_fields=client.custom_fields.list()
# Get custom field by IDcustom_field=client.custom_fields.get("custom_field_id")
# Create a custom fieldnew_custom_field=client.custom_fields.create({
"name": "Preferred Contact Method",
"type": "dropdown",
"options": ["Email", "Phone", "Text"],
"entityType": "contact"
})
# Update custom fieldupdated_custom_field=client.custom_fields.update("custom_field_id", {
"name": "Contact Preference"
})
# Delete custom fieldclient.custom_fields.delete("custom_field_id")

Email Templates

# Get all email templatestemplates=client.email_templates.list()
# Get email template by IDtemplate=client.email_templates.get("template_id")
# Create an email templatenew_template=client.email_templates.create({
"name": "Welcome Email",
"subject": "Welcome to our company!",
"body": "<p>Thank you for joining us!</p>"
})
# Update email templateupdated_template=client.email_templates.update("template_id", {
"subject": "Updated Subject"
})
# Delete email templateclient.email_templates.delete("template_id")

SMS Templates

# Get all SMS templatestemplates=client.sms_templates.list()
# Get SMS template by IDtemplate=client.sms_templates.get("template_id")
# Create an SMS templatenew_template=client.sms_templates.create({
"name": "Appointment Reminder",
"body": "Reminder: Your appointment is scheduled for {{appointment_date}}"
})
# Update SMS templateupdated_template=client.sms_templates.update("template_id", {
"body": "Updated message body"
})
# Delete SMS templateclient.sms_templates.delete("template_id")

Products

# Get all productsproducts=client.products.list(limit=100, page=1)
# Get product by IDproduct=client.products.get("product_id")
# Create a productnew_product=client.products.create({
"name": "Premium Service",
"description": "Our premium service package",
"price": 299.99,
"type": "service"
})
# Update productupdated_product=client.products.update("product_id", {
"price": 249.99
})
# Delete productclient.products.delete("product_id")

Invoices

# Create an invoiceinvoice=client.invoices.create({
"contactId": "contact_id",
"dueDate": "2023-12-31",
"items": [
{"productId": "product_id", "quantity": 1, "price": 299.99}
]
})
# Get all invoicesinvoices=client.invoices.list(limit=100, page=1)
# Get invoice by IDinvoice=client.invoices.get("invoice_id")
# Update invoiceupdated_invoice=client.invoices.update("invoice_id", {
"status": "sent"
})
# Delete invoiceclient.invoices.delete("invoice_id")
# Send invoiceclient.invoices.send("invoice_id", {
"email": "customer@example.com"
})

Payments

# Create a paymentpayment=client.payments.create({
"invoiceId": "invoice_id",
"amount": 299.99,
"method": "credit_card",
"status": "completed"
})
# Get all paymentspayments=client.payments.list(limit=100, page=1)
# Get payment by IDpayment=client.payments.get("payment_id")

Membership Sites

# Get all membership sitessites=client.membership_sites.list()
# Get membership site by IDsite=client.membership_sites.get("site_id")
# Get site membersmembers=client.membership_sites.get_members("site_id", limit=100, page=1)
# Add members to siteclient.membership_sites.add_members("site_id", ["contact_id1", "contact_id2"])
# Remove members from siteclient.membership_sites.remove_members("site_id", ["contact_id1"])

Reporting

# Get contact growth reportcontact_growth=client.reporting.contact_growth({
"startDate": "2023-01-01",
"endDate": "2023-12-31",
"interval": "month"
})
# Get revenue reportrevenue=client.reporting.revenue({
"startDate": "2023-01-01",
"endDate": "2023-12-31",
"interval": "month"
})
# Get campaign performance reportcampaign_performance=client.reporting.campaign_performance({
"campaignId": "campaign_id",
"startDate": "2023-01-01",
"endDate": "2023-12-31"
})

Webhooks

# Create a webhookwebhook=client.webhooks.create({
"url": "https://your-app.com/webhook",
"events": ["contact.created", "opportunity.updated"]
})
# Get all webhookswebhooks=client.webhooks.list()
# Get webhook by IDwebhook=client.webhooks.get("webhook_id")
# Update webhookupdated_webhook=client.webhooks.update("webhook_id", {
"events": ["contact.created", "contact.updated", "opportunity.updated"]
})
# Delete webhookclient.webhooks.delete("webhook_id")

Webhook Processing

Example of processing a Go High Level webhook:

# Flask examplefromflaskimportFlask, request, jsonifyfromghl_clientimportGHLClientapp=Flask(__name__)
client=GHLClient(api_key="your_api_key")
@app.route('/ghl-webhook', methods=['POST'])defghl_webhook():
# Verify webhook signaturesignature=request.headers.get('X-GHL-Signature')
is_valid=client.webhooks.verify_signature(
request.data,
signature,
"your_webhook_secret"
)
ifnotis_valid:
returnjsonify({"error": "Invalid signature"}), 401# Process webhook datawebhook_data=request.jsonevent_type=webhook_data.get('event')
ifevent_type=='contact.created':
contact_data=webhook_data.get('data', {})
# Process new contactprint(f"New contact created: {contact_data.get('firstName')}{contact_data.get('lastName')}")
returnjsonify({"status": "success"}), 200

Pagination

Many list endpoints support pagination:

# Example of handling paginationall_contacts= []
page=1limit=100whileTrue:
contacts_page=client.contacts.list(limit=limit, page=page)
all_contacts.extend(contacts_page.get('contacts', []))
# Check if we've reached the last pagetotal_pages=contacts_page.get('meta', {}).get('totalPages', 0)
ifpage>=total_pages:
breakpage+=1print(f"Total contacts retrieved: {len(all_contacts)}")

Error Handling

The template includes comprehensive error handling:

try:
contact=client.contacts.get("non_existent_id")
exceptGHLApiErrorase:
print(f"API Error: {e.status_code} - {e.message}")
# Handle specific error codesife.status_code==404:
print("Contact not found")
exceptGHLConnectionErrorase:
print(f"Connection Error: {str(e)}")

Multi-Location Support

For agencies or businesses with multiple locations:

# Initialize with specific locationclient=GHLClient(api_key="your_api_key", location_id="location_id")
# Or switch location during executionclient.set_location("new_location_id")
# Use company-wide API (agency level)client.set_company_mode(True)

API Rate Limiting

The template includes rate limit handling:

# Configure rate limit behaviorclient.configure_rate_limiting(
max_retries=3,
retry_delay=2, # secondsbackoff_factor=1.5
)

Resources

License

This project is licensed under the MIT License - see the LICENSE file for details.


Created and maintained by [Your Name/Organization]

About

No description, website, or topics provided.

Resources

Stars

0 stars

Watchers

1 watching

Forks

Releases

Packages

Contributors

Languages