HighLevel OAuth2 MCP Server with n8n Integration

A Model Context Protocol (MCP) server that integrates with the HighLevel API using OAuth2 authentication managed through n8n workflows.

Overview

This project provides a complete solution for managing HighLevel OAuth2 authentication and API interactions using n8n as the authentication orchestrator. The system automatically handles token storage, refresh, and retrieval, making it easy to build applications that interact with the HighLevel API.

Architecture

The system consists of three main components:

1. n8n Workflows

Three automated workflows handle the OAuth2 flow:

• Authorization Workflow: Captures the OAuth callback and exchanges the authorization code for access and refresh tokens
• Token Refresh Workflow: Automatically refreshes the access token every 23 hours
• Token Retrieval API: Provides a webhook endpoint for the MCP server to retrieve the current access token

2. MCP Server

A Python-based server that:

• Retrieves access tokens from n8n
• Provides a clean interface for HighLevel API calls
• Handles token expiration and automatic retry
• Includes helper methods for common API operations

3. Token Storage

Tokens are stored as n8n environment variables:

• HIGHLEVEL_ACCESS_TOKEN: Current access token
• HIGHLEVEL_REFRESH_TOKEN: Refresh token for obtaining new access tokens

Prerequisites

Before setting up this integration, you need:

1. n8n Instance: A running n8n instance with API access

   • Instance URL (e.g., https://n8n.mikee.ai)
   • API Key for programmatic access

2. HighLevel OAuth App: Register an OAuth application in HighLevel Marketplace

   • Client ID
   • Client Secret
   • Redirect URI (will be your n8n webhook URL)

3. Python Environment: Python 3.11+ with the following packages:

   • requests

Installation

Step 1: Clone or Download This Repository

git clone <repository-url>
cd highlevel-mcp-server

Step 2: Set Up Environment Variables

Create a .env file or export the following environment variables:

export N8N_INSTANCE_URL="https://your-n8n-instance.com"
export N8N_API_KEY="your-n8n-api-key"
export HIGHLEVEL_CLIENT_ID="your-highlevel-client-id"
export HIGHLEVEL_CLIENT_SECRET="your-highlevel-client-secret"
export HIGHLEVEL_REDIRECT_URI="https://your-n8n-instance.com/webhook/oauth/callback/highlevel"

Step 3: Upload Workflows to n8n

The workflows have already been uploaded to your n8n instance:

• HighLevel OAuth2 - Authorization (ID: ncH51cSQXxuKy6LB)
• HighLevel OAuth2 - Token Refresh (ID: Mrd1JuSoe9vKNwSn)
• HighLevel OAuth2 - Token Retrieval API (ID: uyPHv0OQeiWFEa84)

Step 4: Configure n8n Environment Variables

In your n8n instance, set the following environment variables:

1. Go to Settings → Environment Variables
2. Add the following variables:
   • HIGHLEVEL_CLIENT_ID: Your HighLevel OAuth app client ID
   • HIGHLEVEL_CLIENT_SECRET: Your HighLevel OAuth app client secret
   • HIGHLEVEL_REDIRECT_URI: Your OAuth callback URL
   • N8N_INSTANCE_URL: Your n8n instance URL (already set)
   • N8N_API_KEY: Your n8n API key (already set)

Step 5: Activate Workflows in n8n

1. Log in to your n8n instance
2. Navigate to Workflows
3. Find and activate the three HighLevel OAuth2 workflows:
   • HighLevel OAuth2 - Authorization
   • HighLevel OAuth2 - Token Refresh
   • HighLevel OAuth2 - Token Retrieval API

Step 6: Complete OAuth Authorization

1. In your HighLevel OAuth app settings, get the installation URL
2. Visit the installation URL to authorize the app
3. You will be redirected to the n8n webhook, which will:
   • Exchange the authorization code for tokens
   • Store the tokens in n8n variables
   • Display a success message

Usage

Basic Usage

from mcp_server_improved import HighLevelMCPServer

# Initialize the server
server = HighLevelMCPServer()

# Get access token
token = server.get_access_token()
print(f"Access Token: {token}")

# Get locations
locations = server.get_locations()
print(f"Locations: {locations}")

# Get contacts for a location
contacts = server.get_contacts(location_id="your-location-id", limit=20)
print(f"Contacts: {contacts}")

# Create a new contact
new_contact = server.create_contact(
    location_id="your-location-id",
    contact_data={
        "firstName": "John",
        "lastName": "Doe",
        "email": "john.doe@example.com",
        "phone": "+1234567890"
    }
)
print(f"New Contact: {new_contact}")

Running the Test Script

python3.11 mcp_server_improved.py

This will:

1. Initialize the MCP server
2. Retrieve the access token from n8n
3. Make a test API call to get locations
4. Display the results

Workflow Details

1. Authorization Workflow

Trigger: Webhook at /webhook/oauth/callback/highlevel

Process:

1. Receives OAuth callback with authorization code
2. Exchanges code for access and refresh tokens
3. Stores tokens in n8n variables
4. Returns success response

Webhook URL: https://your-n8n-instance.com/webhook/oauth/callback/highlevel

2. Token Refresh Workflow

Trigger: Schedule (every 23 hours)

Process:

1. Retrieves current refresh token from n8n variables
2. Requests new access and refresh tokens from HighLevel
3. Updates tokens in n8n variables

Note: This workflow runs automatically to ensure tokens never expire.

3. Token Retrieval API

Trigger: Webhook at /webhook/highlevel/access-token

Process:

1. Receives request from MCP server
2. Retrieves current access token from n8n variables
3. Returns token in JSON format

Webhook URL: https://your-n8n-instance.com/webhook/highlevel/access-token

API Reference

HighLevelMCPServer Class

Methods

get_access_token() -> Optional[str]

Retrieves the current HighLevel access token from n8n.

Returns: Access token string or None if retrieval failed

call_highlevel_api(endpoint, method="GET", data=None, params=None) -> Optional[Dict]

Makes a call to the HighLevel API.

Parameters:

• endpoint: API endpoint (e.g., "v2/contacts/")
• method: HTTP method (GET, POST, PUT, DELETE)
• data: Request body data for POST/PUT requests
• params: Query parameters

Returns: API response data or None if request failed

get_locations() -> Optional[Dict]

Gets all locations accessible to the authenticated user.

get_contacts(location_id, limit=20) -> Optional[Dict]

Gets contacts for a specific location.

create_contact(location_id, contact_data) -> Optional[Dict]

Creates a new contact.

get_opportunities(location_id, limit=20) -> Optional[Dict]

Gets opportunities for a specific location.

Troubleshooting

Issue: "Failed to retrieve access token"

Possible causes:

1. Workflows are not activated in n8n
2. OAuth authorization has not been completed
3. Tokens are not stored in n8n variables
4. n8n environment variables are not configured

Solution:

1. Check that all three workflows are active in n8n
2. Complete the OAuth authorization flow
3. Verify environment variables in n8n settings
4. Check n8n logs for errors

Issue: "401 Unauthorized" when calling HighLevel API

Possible causes:

1. Access token has expired
2. Invalid token
3. Insufficient scopes

Solution:

1. The MCP server automatically retries with a fresh token
2. Check that the token refresh workflow is running
3. Verify OAuth app scopes in HighLevel

Issue: Webhook URLs not working

Possible causes:

1. Workflows are not activated
2. Incorrect webhook path
3. n8n instance not accessible

Solution:

1. Activate workflows in n8n
2. Verify webhook URLs match the workflow configuration
3. Test webhook URLs with curl or browser

Security Considerations

1. Environment Variables: Never commit API keys or secrets to version control
2. Token Storage: Tokens are stored securely in n8n's encrypted variable storage
3. HTTPS: Always use HTTPS for n8n webhooks and API calls
4. Access Control: Limit n8n API key permissions to only what's needed
5. Token Refresh: Tokens are automatically refreshed to minimize exposure time

File Structure

highlevel-mcp-server/
├── README.md                                    # This file
├── SETUP_GUIDE.md                              # Detailed setup instructions
├── mcp_server.py                               # Original MCP server
├── mcp_server_improved.py                      # Improved MCP server with error handling
├── upload_workflows.py                         # Script to upload workflows to n8n
├── workflows/                                  # n8n workflow JSON files
│   ├── auth_workflow_simple.json              # Authorization workflow
│   ├── refresh_workflow_simple.json           # Token refresh workflow
│   └── retrieval_workflow_simple.json         # Token retrieval workflow
├── MCP Server for HighLevel OAuth2 API - Design Document.md
└── marketplace.gohighlevel.com_docs_Authorization_OAuth2.0_index.html.md

Contributing

Contributions are welcome! Please feel free to submit issues or pull requests.

License

MIT License - See LICENSE file for details

Support

For issues or questions:

1. Check the troubleshooting section
2. Review n8n workflow execution logs
3. Check HighLevel API documentation
4. Open an issue in this repository

Acknowledgments

• Built with n8n - Workflow automation platform
• Integrates with HighLevel - All-in-one marketing platform
• Follows Model Context Protocol specifications