Highlevel MCP Server

带有n8n集成的高级OAuth2 MCP服务器

一个通过OAuth2认证（由n8n工作流管理）与HighLevel API集成的模型上下文协议（MCP）服务器。

概述

该项目提供了一个完整的解决方案，用于通过n8n作为认证协调器来管理HighLevel OAuth2认证和API交互。系统自动处理令牌的存储、刷新和检索，使得构建与HighLevel API交互的应用程序变得轻松简单。

建筑学

该系统由三个主要组件构成：

1. n8n 工作流

三个自动化工作流处理OAuth2流程：

• 授权工作流程捕获OAuth回调，并将授权码交换为访问令牌和刷新令牌
• 令牌刷新工作流程每23小时自动刷新访问令牌
• 令牌检索API为MCP服务器提供一个webhook端点，用于获取当前访问令牌

2. MCP 服务器

一个基于Python的服务器，其功能包括：

• 从n8n中检索访问令牌
• 为高级API调用提供简洁的接口
• 处理令牌过期并自动重试
• 包含常见API操作的辅助方法

3. 令牌存储

令牌以 n8n 环境变量的形式存储：

• HIGHLEVEL_ACCESS_TOKEN当前访问令牌
• HIGHLEVEL_REFRESH_TOKEN用于获取新访问令牌的刷新令牌

先决条件

在设置此集成之前，您需要：

1. n8n 实例一个可运行的n8n实例，具有API访问权限

- 实例URL（例如。， https://n8n.mikee.ai) - 程序化访问的API密钥

1. 高级OAuth应用程序在HighLevel市场上注册一个OAuth应用程序

- 客户端ID - 客户端密钥 - 重定向URI（将作为您的n8n webhook URL）

1. Python 环境Python 3.11+ 及以下包：

- requests

安装

步骤1：克隆或下载此仓库

git clone 
cd highlevel-mcp-server

步骤2：设置环境变量

创建一个 .env 导出以下环境变量：

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"

步骤3：将工作流上传到n8n

工作流已上传到您的n8n实例中：

• 高级OAuth2 - 授权 (ID: ncH51cSQXxuKy6LB)
• 高级OAuth2 - 令牌刷新 (身份/编号： Mrd1JuSoe9vKNwSn）
• 高级OAuth2 - 令牌获取API (ID: uyPHv0OQeiWFEa84)

步骤4：配置n8n环境变量

在您的 n8n 实例中，设置以下环境变量：

1. 进入设置 → 环境变量
2. 添加以下变量：

- HIGHLEVEL_CLIENT_ID您的高级OAuth应用程序客户端ID - HIGHLEVEL_CLIENT_SECRET您的高级OAuth应用客户端密钥 - HIGHLEVEL_REDIRECT_URI您的OAuth回调URL - N8N_INSTANCE_URL您的n8n实例URL（已设置） - N8N_API_KEY你的n8n API密钥（已设置）

步骤5：在n8n中激活工作流

1. 登录到您的n8n实例
2. 导航至工作流
3. 查找并激活三个高级OAuth2工作流：

- 高级OAuth2 - 授权 - 高级OAuth2 - 令牌刷新 - 高级OAuth2 - 令牌检索API

步骤6：完成OAuth授权

1. 在您的高级OAuth应用设置中，获取安装URL
2. 访问安装网址以授权该应用
3. 您将被重定向到n8n的webhook，该webhook将：

- 将授权码兑换为令牌 - 将令牌存储在 n8n 变量中 - 显示成功消息

用法

基本用法

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}")

运行测试脚本

python3.11 mcp_server_improved.py

这将：

1. 初始化MCP服务器
2. 从n8n中获取访问令牌
3. 制作一个测试API调用来获取位置信息
4. 显示结果

工作流详情

1. 授权工作流程

触发器Webhook 在 /webhook/oauth/callback/highlevel

过程:

1. 接收带有授权码的OAuth回调
2. 交换访问令牌和刷新令牌的代码
3. 将令牌存储在 n8n 变量中
4. 返回成功响应

Webhook URL（网络钩子URL）: https://your-n8n-instance.com/webhook/oauth/callback/highlevel

2. 令牌刷新工作流程

触发器时间表（每23小时一次）

过程：

1. 从n8n变量中检索当前的刷新令牌
2. 从HighLevel请求新的访问令牌和刷新令牌
3. 在n8n变量中更新令牌

注此工作流会自动运行，以确保令牌永不过期。

3. 令牌检索API

触发器Webhook位于 /webhook/highlevel/access-token

过程：

1. 接收来自MCP服务器的请求
2. 从n8n变量中检索当前访问令牌
3. 以JSON格式返回令牌

Webhook URL（网络钩子URL）： https://your-n8n-instance.com/webhook/highlevel/access-token

API 参考

HighLevelMCPServer 类

方法

get_access_token() -> Optional[str]

从n8n中检索当前的HighLevel访问令牌。

回报访问令牌字符串，或如果检索失败则为None

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

调用高级API。

参数：

• endpointAPI端点（例如，“v2/contacts/”）
• methodHTTP方法（GET、POST、PUT、DELETE）
• dataPOST/PUT 请求的请求体数据
• params查询参数

退货API响应数据，或请求失败时返回None

get_locations() -> Optional[Dict]

获取已认证用户可访问的所有位置。

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

获取特定位置的联系人。

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

创建一个新的联系人。

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

获取特定地点的机会。

故障排除

问题：“无法获取访问令牌”

可能的原因:

1. 在n8n中，工作流未被激活
2. OAuth 授权尚未完成
3. 代币不会存储在n8n变量中
4. n8n 环境变量未配置

解决方案：

1. 检查n8n中是否所有三个工作流都处于激活状态
2. 完成OAuth授权流程
3. 在n8n设置中验证环境变量
4. 检查 n8n 日志中的错误

问题：调用高级API时出现“401 未经授权”

可能的原因：

1. 访问令牌已过期
2. 无效的令牌
3. 范围不足

解决方案:

1. MCP服务器会自动使用新的令牌重试
2. 检查令牌刷新工作流是否正在运行
3. 在HighLevel中验证OAuth应用的权限范围

问题：Webhook URL 无法正常工作

可能的原因：

1. 工作流未激活
2. 错误的webhook路径
3. n8n 实例无法访问

解决方案：

1. 在n8n中激活工作流
2. 验证webhook URL是否与工作流配置匹配
3. 使用curl或浏览器测试webhook URL

安全考量

1. 环境变量永远不要将API密钥或机密提交到版本控制系统中
2. 令牌存储令牌被安全地存储在n8n的加密变量存储中
3. HTTPS（超文本传输安全协议）始终使用HTTPS进行n8n的webhooks和API调用
4. 访问控制将n8n API密钥权限限制为仅需用到的权限
5. 令牌刷新令牌会自动刷新，以最大限度地减少暴露时间

文件结构

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

贡献；助力

欢迎贡献！请随时提交问题或拉取请求。

许可证

MIT 许可证 - 详见 LICENSE 文件

支持

对于问题或疑问：

1. 查看故障排除部分
2. 审查n8n工作流执行日志
3. 检查高级API文档
4. 在这个仓库中打开一个问题（或：提交一个问题）

致谢

• 构建于 n8n - 工作流自动化平台
• 与……集成 高级（或高水平） - 一体化营销平台
• 遵循 模型上下文协议 规格；规范