十六进制mcp mcp服务器
用于Hex的MCP服务器,实现编排、监控、单元内容访问、权限管理、收集组织和组管理工具。
这做什么(和不做什么)
实际使用案例
编排和自动化:
- Trigger Hex项目从外部系统运行(气流DAGs、CI/CD管道)
- 以编程方式监视运行状态
- 取消长时间运行或卡住的执行
- 跨工作区发现和搜索项目
运行监控:
- 检查计划运行是否成功完成
- 获取运行历史记录以进行审核
- 通过程序访问项目元数据(所有者、上次编辑、描述)
单元格内容访问 新
- 读取笔记本结构和单元格元数据
- 从SQL单元格读取SQL查询源代码
- 从code单元格读取Python/R代码
- 适用于查询迁移、代码分析和内容审计
权限管理 新
- 以编程方式管理用户和组对项目的访问权限
- 跨多个项目批量更新权限
- 管理工作区范围和公共访问设置
- 在集合中添加或删除项目
藏书组织 新
- 创建和管理用于组织项目的集合
- 按部门、团队或主题组织项目
- 控制集合可见性和访问
- 将项目批量组织到集合中
组管理 新
- 创建和管理用户组以进行权限管理
- 批量添加和删除组中的用户
- 使用组简化跨项目的权限管理
- 保持有组织的团队结构
数据连接管理 新
- 列出并管理数据库和仓库连接
- 为BigQuery、Snowflake、Postgres、Redshift、Athena、Databricks创建连接
- 更新连接配置和凭据
- 控制跨工作区的数据连接共享
关键限制
笔记本内容访问受限:
- ✅ 可以读取SQL和CODE单元格源内容
- ❌ 无法读取MARKDOWN、INPUT或可视化单元格
- ❌ 无法创建或删除单元格
- ❌ 无法修改笔记本结构
- ❌ 无法查看查询结果或图表
- ❌ 无法管理笔记本依赖关系或参数
不适合:
- 从头开始构建或编写笔记本
- 协作笔记本开发
- 调试查询或代码执行
- 完整笔记本备份(仅SQL/CODE单元格可访问)
- 读取标记文档或输入参数
何时使用此
需要时使用十六进制mcp:
- 编排 从外部系统执行十六进制
- 监视器 运行状态和历史记录
- 阅读 来自现有笔记本的SQL查询和代码
- 审计 或 迁移 跨项目的SQL/CODE内容
- 管理 跨项目的权限和访问控制
- 自动化 用户、组和集合的批量权限更新
- 组织 按部门、团队或主题将项目放入集合中
- 维持 简化权限管理的用户组
- 标准化 跨Hex实例的工作区组织
- 管理 以编程方式进行数据连接和数据库集成
- 自动化 新工作区或环境的连接设置
对于完整的笔记本开发和编辑,直接使用Hex web UI。
可用工具
项目运营
list_hex_projects:列出可用的十六进制项目search_hex_projects:按模式搜索十六进制项目get_hex_project:获取特定项目的详细信息
项目执行
run_hex_project:执行十六进制项目get_hex_run_status:检查项目运行的状态get_hex_project_runs:获取项目运行的历史记录cancel_hex_run:取消正在运行的项目
细胞操作(新)
list_hex_cells:使用SQL/code单元格的源代码列出项目中的所有单元格update_hex_cell:更新SQL或CODE单元格源和/或数据连接
权限管理(新增)
update_hex_project_user_sharing:授予或撤销用户对项目的访问权限update_hex_project_group_sharing:授予或撤销组对项目的访问权限update_hex_project_collection_sharing:在集合中添加或删除项目update_hex_project_workspace_sharing:更新工作区范围和公共访问
收藏管理(新)
list_hex_collections:列出工作区中的所有集合get_hex_collection:获取特定收藏的详细信息create_hex_collection:使用可选共享设置创建新收藏update_hex_collection:更新集合名称、描述或共享设置
集团管理(新)
list_hex_groups:列出工作区中的所有组get_hex_group:获取特定组的详细信息create_hex_group:创建一个具有可选初始成员的新组update_hex_group:更新组名和/或成员资格(添加/删除用户)delete_hex_group:从工作区中删除组
数据连接管理(新)
list_hex_data_connections:列出工作区中的所有数据连接get_hex_data_connection:获取特定数据连接的详细信息create_hex_data_connection:创建新的数据库/仓库连接update_hex_data_connection:更新连接配置、凭据或共享
安装
建议使用uv安装hex mcp:
uv add hex-mcp或者使用pip:
pip install hex-mcp要确认它正在工作,您可以运行:
hex-mcp --version配置
使用config命令(推荐)
配置十六进制mcp的最简单方法是使用 config 命令并传递API密钥和API URL(可选,默认为 https://app.hex.tech/api/v1):
hex-mcp config --api-key "your_hex_api_key" --api-url "https://app.hex.tech/api/v1"\[!注意\] 这会将您的配置保存到主目录中的文件中(例如。 ~/.hex-mcp/config.yml),使其可用于所有十六进制mcp调用。使用环境变量
或者,Hex MCP服务器可以配置环境变量:
HEX_API_KEY:您的十六进制API密钥HEX_API_URL:十六进制API基URL
在为MCP服务器设置环境变量时,它们需要是全局的,以便Cursor获取它们,或者使用uv --env-file 调用服务器时标记。
与光标一起使用
Cursor允许AI代理通过MCP协议与Hex交互。按照以下步骤设置和使用带有Cursor的十六进制mcp。您可以创建 .cursor/mcp.json 包含以下内容的项目根目录中的文件:
{
"mcpServers": {
"hex-mcp": {
"command": "uv",
"args": ["run", "hex-mcp", "run"]
}
}
}或者,您可以使用 hex-mcp 如果它在您的PATH中,请直接命令:
{
"mcpServers": {
"hex-mcp": {
"command": "hex-mcp",
"args": ["run"]
}
}
}一旦它启动并运行,您可以在Cursor中使用它,发起一个新的AI(代理)对话,并要求它列出或运行一个Hex项目。
\[!重要\] MCP服务器和CLI仍在开发中,可能会发生重大变化。
使用示例
读取单元格内容
使用 list_hex_cells 要阅读十六进制笔记本的结构和源代码:
# List all cells in a project
cells = list_hex_cells(project_id="your-project-uuid")
# The response includes cell metadata and source code for SQL/CODE cells
# Example response structure:
{
"values": [
{
"id": "cell-uuid-123",
"staticId": "static-id-456",
"cellType": "SQL",
"label": "Load Customer Data",
"dataConnectionId": "connection-uuid-789",
"contents": {
"sqlCell": {
"source": "SELECT * FROM customers WHERE active = true"
},
"codeCell": null
}
},
{
"id": "cell-uuid-456",
"cellType": "CODE",
"label": "Process Data",
"dataConnectionId": null,
"contents": {
"sqlCell": null,
"codeCell": {
"source": "import pandas as pd\ndf = df.dropna()"
}
}
},
{
"id": "cell-uuid-789",
"cellType": "MARKDOWN",
"label": "Documentation",
"dataConnectionId": null,
"contents": {
"sqlCell": null,
"codeCell": null
}
}
],
"pagination": {
"next": null,
"previous": null
}
}备注:只有SQL和CODE单元格包含源内容。MARKDOWN、INPUT和可视化单元格均返回null sqlCell 和 codeCell.
分页
对于包含许多单元格的项目,请使用分页:
# First page
page1 = list_hex_cells(project_id="your-project-uuid", limit=50)
# Next page using cursor
if page1["pagination"]["next"]:
page2 = list_hex_cells(
project_id="your-project-uuid",
limit=50,
after=page1["pagination"]["next"]
)
# Previous page
if page2["pagination"]["previous"]:
page1_again = list_hex_cells(
project_id="your-project-uuid",
limit=50,
before=page2["pagination"]["previous"]
)更新单元格内容
以编程方式更新SQL或CODE单元格:
# Update SQL cell source code
result = update_hex_cell(
cell_id="cell-uuid-123",
sql_source="SELECT * FROM updated_table WHERE active = true"
)
# Update CODE cell source code
result = update_hex_cell(
cell_id="cell-uuid-456",
code_source="import pandas as pd\ndf = df.dropna()"
)
# Update SQL cell data connection (without changing source)
result = update_hex_cell(
cell_id="cell-uuid-123",
data_connection_id="new-connection-uuid"
)
# Update both SQL source and connection in one call
result = update_hex_cell(
cell_id="cell-uuid-123",
sql_source="SELECT * FROM production.customers",
data_connection_id="production-connection-uuid"
)重要:
- 只能更新SQL和CODE单元格类型
- 无法更新MARKDOWN、INPUT或可视化单元格
- 需要
EDIT_PROJECT_CONTENTS许可 - 无法同时更新两者
sql_source和code_source在同一个调用中(单元格为SQL或CODE)
查询迁移示例
跨项目提取和更新SQL查询:
import json
# Get all cells from source project
cells_response = list_hex_cells(project_id="source-project-uuid")
cells = json.loads(cells_response)
# Filter for SQL cells
sql_cells = [
cell for cell in cells["values"]
if cell["cellType"] == "SQL" and cell["contents"]["sqlCell"]
]
# Migrate queries to new data connection
new_connection_id = "production-bigquery-connection"
for cell in sql_cells:
original_query = cell["contents"]["sqlCell"]["source"]
# Update table references (example: dev -> prod)
updated_query = original_query.replace("dev.schema", "prod.schema")
# Update the cell with new query and connection
result = update_hex_cell(
cell_id=cell["id"],
sql_source=updated_query,
data_connection_id=new_connection_id
)
print(f"Updated cell: {cell['label']}")
print(f"Migrated {len(sql_cells)} SQL cells to production")代码重构示例
跨多个单元格批量更新Python代码:
import json
# Get all CODE cells
cells_response = list_hex_cells(project_id="project-uuid")
cells = json.loads(cells_response)
code_cells = [
cell for cell in cells["values"]
if cell["cellType"] == "CODE" and cell["contents"]["codeCell"]
]
# Update import statements
for cell in code_cells:
original_code = cell["contents"]["codeCell"]["source"]
# Replace deprecated import
if "from old_library" in original_code:
updated_code = original_code.replace(
"from old_library import func",
"from new_library import func"
)
# Update the cell
result = update_hex_cell(
cell_id=cell["id"],
code_source=updated_code
)
print(f"Updated cell: {cell['label']}")管理项目权限
使用共享端点以编程方式控制对Hex项目的访问。
授予用户访问权限
from hex_mcp.server import update_hex_project_user_sharing
# Grant edit access to a single user
result = update_hex_project_user_sharing(
project_id="project-123",
user_permissions=[
{"user_id": "user-456", "access": "CAN_EDIT"}
]
)
# Grant access to multiple users with different permissions
result = update_hex_project_user_sharing(
project_id="project-123",
user_permissions=[
{"user_id": "user-001", "access": "CAN_VIEW"},
{"user_id": "user-002", "access": "CAN_EDIT"},
{"user_id": "user-003", "access": "FULL_ACCESS"}
]
)
# Revoke user access
result = update_hex_project_user_sharing(
project_id="project-123",
user_permissions=[
{"user_id": "user-456", "access": "NONE"}
]
)访问级别:
NONE:撤销所有访问权限APP_ONLY:仅在应用程序内访问(已发布版本)CAN_VIEW:在逻辑视图中查看项目(只读)CAN_EDIT:编辑项目内容FULL_ACCESS:编辑内容和管理项目设置/权限
授予组访问权限
from hex_mcp.server import update_hex_project_group_sharing
# Grant access to a group
result = update_hex_project_group_sharing(
project_id="project-123",
group_permissions=[
{"group_id": "group-789", "access": "CAN_VIEW"}
]
)
# Grant access to multiple groups
result = update_hex_project_group_sharing(
project_id="project-123",
group_permissions=[
{"group_id": "analytics-team", "access": "CAN_EDIT"},
{"group_id": "leadership-team", "access": "CAN_VIEW"}
]
)将项目添加到集合
from hex_mcp.server import update_hex_project_collection_sharing
# Add project to a collection
result = update_hex_project_collection_sharing(
project_id="project-123",
collection_permissions=[
{"collection_id": "collection-456", "access": "CAN_VIEW"}
]
)
# Add to multiple collections with different access levels
result = update_hex_project_collection_sharing(
project_id="project-123",
collection_permissions=[
{"collection_id": "public-dashboards", "access": "APP_ONLY"},
{"collection_id": "internal-analytics", "access": "CAN_EDIT"}
]
)
# Remove from collection
result = update_hex_project_collection_sharing(
project_id="project-123",
collection_permissions=[
{"collection_id": "collection-456", "access": "NONE"}
]
)管理工作区和公共访问
from hex_mcp.server import update_hex_project_workspace_sharing
# Make project viewable by entire workspace
result = update_hex_project_workspace_sharing(
project_id="project-123",
workspace_access="CAN_VIEW"
)
# Make project publicly accessible
result = update_hex_project_workspace_sharing(
project_id="project-123",
public_access="APP_ONLY"
)
# Update both workspace and public access
result = update_hex_project_workspace_sharing(
project_id="project-123",
workspace_access="CAN_EDIT",
public_access="APP_ONLY"
)
# Revoke public access (keep workspace access)
result = update_hex_project_workspace_sharing(
project_id="project-123",
public_access="NONE"
)批量权限管理示例
跨多个项目标准化权限:
import json
from hex_mcp.server import (
list_hex_projects,
update_hex_project_user_sharing,
update_hex_project_workspace_sharing
)
# Get all projects
projects_response = list_hex_projects()
projects = json.loads(projects_response)
# Define standard access policies
analytics_team_users = ["user-001", "user-002", "user-003"]
leadership_users = ["user-100", "user-101"]
for project in projects:
project_id = project["projectId"]
# Grant analytics team edit access
update_hex_project_user_sharing(
project_id=project_id,
user_permissions=[
{"user_id": user_id, "access": "CAN_EDIT"}
for user_id in analytics_team_users
]
)
# Grant leadership view access
update_hex_project_user_sharing(
project_id=project_id,
user_permissions=[
{"user_id": user_id, "access": "CAN_VIEW"}
for user_id in leadership_users
]
)
# Make viewable by workspace, not public
update_hex_project_workspace_sharing(
project_id=project_id,
workspace_access="CAN_VIEW",
public_access="NONE"
)
print(f"Updated permissions for: {project['name']}")重要说明:
- 所有共享操作都需要适当的工作区权限
- 用户/组ID必须存在于您的十六进制工作区中
- 集合ID必须是您有权访问的有效集合
- 工作区访问权限适用于所有工作区成员
- 公众访问(
publicWeb)允许任何有链接的人访问
管理收藏
将项目组织到集合中,以更好地组织工作空间。
列出收藏
from hex_mcp.server import list_hex_collections
import json
# List all collections
collections = list_hex_collections()
data = json.loads(collections)
for collection in data["values"]:
print(f"{collection['name']}: {collection['id']}")
print(f" Created by: {collection.get('creator', {}).get('email', 'N/A')}")
# List with pagination
page1 = list_hex_collections(limit=10)
data = json.loads(page1)
if data["pagination"]["next"]:
page2 = list_hex_collections(limit=10, after=data["pagination"]["next"])获取收藏详细信息
from hex_mcp.server import get_hex_collection
import json
# Get specific collection
collection = get_hex_collection(collection_id="collection-123")
data = json.loads(collection)
print(f"Collection: {data['name']}")
print(f"Description: {data.get('description', 'No description')}")
print(f"Sharing with {len(data['sharing'].get('users', []))} users")
print(f"Sharing with {len(data['sharing'].get('groups', []))} groups")创建收藏
from hex_mcp.server import create_hex_collection
import json
# Create simple collection
collection = create_hex_collection(
name="Q4 2026 Analytics",
description="Analytics projects for Q4 2026"
)
data = json.loads(collection)
print(f"Created collection: {data['id']}")
# Create with sharing settings
collection = create_hex_collection(
name="Executive Dashboards",
description="Dashboards for executive team",
sharing_users=[
{"id": "user-ceo", "access": "CAN_VIEW"},
{"id": "user-cfo", "access": "CAN_VIEW"}
],
workspace_access="NONE" # Only specified users can access
)
# Create with group access
collection = create_hex_collection(
name="Finance Reports",
description="Financial reporting and analysis",
sharing_groups=[
{"id": "finance-team", "access": "CAN_EDIT"}
],
workspace_access="CAN_VIEW"
)更新收藏
from hex_mcp.server import update_hex_collection
import json
# Update collection name and description
result = update_hex_collection(
collection_id="collection-123",
name="Q1 2027 Analytics",
description="Updated for Q1 2027"
)
# Update sharing settings
result = update_hex_collection(
collection_id="collection-123",
sharing_users=[
{"id": "user-456", "access": "CAN_EDIT"}
],
workspace_access="CAN_VIEW"
)
# Update multiple fields at once
result = update_hex_collection(
collection_id="collection-123",
name="Updated Collection Name",
description="Updated description",
sharing_groups=[
{"id": "analytics-team", "access": "CAN_EDIT"}
],
workspace_access="CAN_VIEW"
)收款组织示例
按部门组织所有项目:
import json
from hex_mcp.server import (
list_hex_projects,
list_hex_collections,
create_hex_collection,
update_hex_project_collection_sharing
)
# Define department structure
departments = {
"Engineering": ["user-eng-lead", "group-engineering"],
"Finance": ["user-fin-lead", "group-finance"],
"Marketing": ["user-mkt-lead", "group-marketing"]
}
# Create collections for each department
collection_ids = {}
for dept_name, members in departments.items():
# Create collection
collection = create_hex_collection(
name=f"{dept_name} Projects",
description=f"All {dept_name} department projects",
workspace_access="CAN_VIEW"
)
data = json.loads(collection)
collection_ids[dept_name] = data["id"]
print(f"Created collection for {dept_name}: {data['id']}")
# Get all projects
projects = list_hex_projects()
projects_data = json.loads(projects)
# Organize projects into collections based on name patterns
for project in projects_data:
project_id = project["projectId"]
project_name = project["name"].lower()
# Determine which collection(s) this project belongs to
if "engineering" in project_name or "technical" in project_name:
update_hex_project_collection_sharing(
project_id=project_id,
collection_permissions=[
{"collection_id": collection_ids["Engineering"], "access": "CAN_EDIT"}
]
)
print(f"Added '{project['name']}' to Engineering collection")
elif "finance" in project_name or "revenue" in project_name:
update_hex_project_collection_sharing(
project_id=project_id,
collection_permissions=[
{"collection_id": collection_ids["Finance"], "access": "CAN_EDIT"}
]
)
print(f"Added '{project['name']}' to Finance collection")
elif "marketing" in project_name or "campaign" in project_name:
update_hex_project_collection_sharing(
project_id=project_id,
collection_permissions=[
{"collection_id": collection_ids["Marketing"], "access": "CAN_EDIT"}
]
)
print(f"Added '{project['name']}' to Marketing collection")
print("Collection organization complete!")收款管理说明:
- 集合有助于按团队、部门或主题组织项目
- 集合可以拥有独立于项目的共享设置
- 一个项目可以属于多个集合
- 收藏的工作区访问权限控制谁可以查看收藏的存在
- 创建集合需要适当的工作区权限
管理组
将用户组织成组,以简化权限管理。
列出组
from hex_mcp.server import list_hex_groups
import json
# List all groups
groups = list_hex_groups()
data = json.loads(groups)
for group in data["values"]:
print(f"{group['name']}: {group['id']}")
print(f" Created: {group['createdAt']}")
# List with pagination and sorting
page1 = list_hex_groups(
limit=10,
sort_by="name",
sort_direction="asc"
)
data = json.loads(page1)
if data["pagination"]["next"]:
page2 = list_hex_groups(limit=10, after=data["pagination"]["next"])获取组详细信息
from hex_mcp.server import get_hex_group
import json
# Get specific group
group = get_hex_group(group_id="group-123")
data = json.loads(group)
print(f"Group: {data['name']}")
print(f"ID: {data['id']}")
print(f"Created: {data['createdAt']}")创建组
from hex_mcp.server import create_hex_group
import json
# Create simple group
group = create_hex_group(name="Analytics Team")
data = json.loads(group)
print(f"Created group: {data['id']}")
# Create with initial members
group = create_hex_group(
name="Finance Team",
member_user_ids=[
"user-123",
"user-456",
"user-789"
]
)
data = json.loads(group)
print(f"Created group with 3 members: {data['id']}")更新组
from hex_mcp.server import update_hex_group
import json
# Update group name
result = update_hex_group(
group_id="group-123",
name="Senior Analytics Team"
)
# Add members to group
result = update_hex_group(
group_id="group-123",
add_member_user_ids=["user-new-1", "user-new-2"]
)
# Remove members from group
result = update_hex_group(
group_id="group-123",
remove_member_user_ids=["user-old-1"]
)
# Update name and membership together
result = update_hex_group(
group_id="group-123",
name="Lead Analytics Team",
add_member_user_ids=["user-lead"],
remove_member_user_ids=["user-junior"]
)删除组
from hex_mcp.server import delete_hex_group
# Delete a group (does not delete the users, only the group)
result = delete_hex_group(group_id="group-old-team")
print(result) # "Group deleted successfully"批量组管理示例
创建基于部门的组并分配用户:
import json
from hex_mcp.server import (
create_hex_group,
update_hex_group,
update_hex_project_group_sharing
)
# Define department structure with user lists
departments = {
"Engineering": {
"leads": ["user-eng-lead-1", "user-eng-lead-2"],
"members": ["user-eng-1", "user-eng-2", "user-eng-3"]
},
"Data Science": {
"leads": ["user-ds-lead"],
"members": ["user-ds-1", "user-ds-2", "user-ds-3", "user-ds-4"]
},
"Analytics": {
"leads": ["user-analytics-lead"],
"members": ["user-analyst-1", "user-analyst-2"]
}
}
# Create groups for each department
group_ids = {}
for dept_name, users in departments.items():
# Create main department group
all_members = users["leads"] + users["members"]
group = create_hex_group(
name=f"{dept_name} Team",
member_user_ids=all_members
)
data = json.loads(group)
group_ids[dept_name] = data["id"]
print(f"Created {dept_name} Team: {data['id']} ({len(all_members)} members)")
# Create leadership sub-group
lead_group = create_hex_group(
name=f"{dept_name} Leadership",
member_user_ids=users["leads"]
)
lead_data = json.loads(lead_group)
group_ids[f"{dept_name}_Leadership"] = lead_data["id"]
print(f"Created {dept_name} Leadership: {lead_data['id']} ({len(users['leads'])} members)")
# Grant group access to relevant projects
# Engineering team gets edit access to technical projects
update_hex_project_group_sharing(
project_id="technical-project-123",
group_permissions=[
{"group_id": group_ids["Engineering"], "access": "CAN_EDIT"}
]
)
# Analytics team gets view access to all dashboards
update_hex_project_group_sharing(
project_id="dashboard-project-456",
group_permissions=[
{"group_id": group_ids["Analytics"], "access": "CAN_VIEW"}
]
)
# Leadership groups get view access to executive dashboards
for dept in ["Engineering", "Data Science", "Analytics"]:
leadership_group_id = group_ids[f"{dept}_Leadership"]
update_hex_project_group_sharing(
project_id="executive-dashboard-789",
group_permissions=[
{"group_id": leadership_group_id, "access": "CAN_VIEW"}
]
)
print("Group-based permission management complete!")团队重组示例
通过更新组成员资格来处理团队更改:
import json
from hex_mcp.server import (
list_hex_groups,
update_hex_group
)
# Get all groups
groups = list_hex_groups()
groups_data = json.loads(groups)
# Find specific groups that need updates
engineering_group = None
for group in groups_data["values"]:
if group["name"] == "Engineering Team":
engineering_group = group
break
if engineering_group:
# New hires joining
new_engineers = ["user-new-eng-1", "user-new-eng-2"]
# Engineers leaving
departing_engineers = ["user-old-eng-1"]
# Update group membership
result = update_hex_group(
group_id=engineering_group["id"],
add_member_user_ids=new_engineers,
remove_member_user_ids=departing_engineers
)
print(f"Updated {engineering_group['name']}")
print(f" Added: {len(new_engineers)} members")
print(f" Removed: {len(departing_engineers)} members")集团管理说明:
- 组允许您一次授予多个用户访问权限,从而简化了权限管理
- 一个用户可以属于多个组
- 在一次更新操作中最多可以添加或删除100个用户
- 删除组不会删除用户,只会删除组本身
- 组权限由所有组成员继承
- 使用具有项目共享端点的组授予团队范围的访问权限
管理数据连接
管理Hex项目的数据库和仓库连接。
列出数据连接
from hex_mcp.server import list_hex_data_connections
import json
# List all data connections
connections = list_hex_data_connections()
data = json.loads(connections)
for conn in data["values"]:
print(f"{conn['name']} ({conn['type']}): {conn['id']}")
if conn.get('description'):
print(f" Description: {conn['description']}")
# List with pagination and sorting
page1 = list_hex_data_connections(
limit=10,
sort_by="NAME",
sort_direction="asc"
)
data = json.loads(page1)
if data["pagination"]["next"]:
page2 = list_hex_data_connections(limit=10, after=data["pagination"]["next"])获取数据连接详细信息
from hex_mcp.server import get_hex_data_connection
import json
# Get specific connection
connection = get_hex_data_connection(connection_id="conn-123")
data = json.loads(connection)
print(f"Connection: {data['name']}")
print(f"Type: {data['type']}")
print(f"Description: {data.get('description', 'N/A')}")
print(f"Include Magic: {data.get('includeMagic', False)}")
print(f"Allow Writeback: {data.get('allowWritebackCells', False)}")
# Check sharing settings
if 'sharing' in data:
workspace = data['sharing'].get('workspace', {})
print(f"Workspace Access: {workspace.get('members', 'None')}")创建数据连接
from hex_mcp.server import create_hex_data_connection
import json
# Create BigQuery connection
bigquery_conn = create_hex_data_connection(
name="Production BigQuery",
connection_type="bigquery",
connection_details={
"bigquery": {
"serviceAccountJsonConfig": json.dumps(service_account_config),
"projectId": "my-gcp-project",
"enableStorageApi": True,
"enableDriveAccess": False
}
},
description="Production data warehouse for analytics",
include_magic=True,
allow_writeback_cells=False,
sharing={
"workspace": {
"members": "CAN_USE"
}
}
)
data = json.loads(bigquery_conn)
print(f"Created BigQuery connection: {data['id']}")
# Create Snowflake connection
snowflake_conn = create_hex_data_connection(
name="Snowflake Analytics",
connection_type="snowflake",
connection_details={
"snowflake": {
"account": "my-account",
"warehouse": "ANALYTICS_WH",
"database": "ANALYTICS_DB",
"schema": "PUBLIC",
"username": "hex_service",
"password": "secure_password"
}
},
description="Snowflake warehouse for analytics",
include_magic=True
)
data = json.loads(snowflake_conn)
print(f"Created Snowflake connection: {data['id']}")
# Create Postgres connection
postgres_conn = create_hex_data_connection(
name="Application Database",
connection_type="postgres",
connection_details={
"postgres": {
"hostname": "db.example.com",
"port": 5432,
"database": "app_db",
"username": "readonly_user",
"password": "secure_password",
"ssl": True
}
},
description="Production application database (read-only)",
connect_via_ssh=False,
include_magic=True,
allow_writeback_cells=False
)
data = json.loads(postgres_conn)
print(f"Created Postgres connection: {data['id']}")更新数据连接
from hex_mcp.server import update_hex_data_connection
import json
# Update connection name and description
result = update_hex_data_connection(
connection_id="conn-123",
name="Updated Production BigQuery",
description="Updated production data warehouse for analytics and reporting"
)
data = json.loads(result)
print(f"Updated connection: {data['name']}")
# Update connection settings
result = update_hex_data_connection(
connection_id="conn-123",
include_magic=False,
allow_writeback_cells=True
)
print("Updated connection settings")
# Update connection sharing
result = update_hex_data_connection(
connection_id="conn-123",
sharing={
"workspace": {
"members": "CAN_USE",
"guests": "NONE"
},
"groups": {
"upsert": [
{"group": {"id": "analytics-team-group-id"}, "access": "CAN_USE"},
{"group": {"id": "data-eng-group-id"}, "access": "CAN_ADMIN"}
]
}
}
)
print("Updated connection sharing")
# Rotate credentials by updating connection details
result = update_hex_data_connection(
connection_id="conn-123",
connection_details={
"bigquery": {
"serviceAccountJsonConfig": json.dumps(new_service_account),
"projectId": "my-gcp-project",
"enableStorageApi": True,
"enableDriveAccess": False
}
}
)
print("Rotated connection credentials")批量连接设置
自动化新工作区或环境的连接设置:
from hex_mcp.server import create_hex_data_connection
import json
# Define standard connections for a workspace
connection_configs = {
"Production BigQuery": {
"type": "bigquery",
"details": {
"bigquery": {
"serviceAccountJsonConfig": json.dumps(prod_bq_creds),
"projectId": "prod-project",
"enableStorageApi": True
}
},
"description": "Production BigQuery warehouse"
},
"Staging BigQuery": {
"type": "bigquery",
"details": {
"bigquery": {
"serviceAccountJsonConfig": json.dumps(staging_bq_creds),
"projectId": "staging-project",
"enableStorageApi": True
}
},
"description": "Staging BigQuery warehouse"
},
"Production Snowflake": {
"type": "snowflake",
"details": {
"snowflake": {
"account": "prod-account",
"warehouse": "PROD_WH",
"database": "PROD_DB",
"username": "hex_prod",
"password": prod_sf_password
}
},
"description": "Production Snowflake warehouse"
},
"Application DB": {
"type": "postgres",
"details": {
"postgres": {
"hostname": "prod-db.example.com",
"port": 5432,
"database": "app_db",
"username": "readonly",
"password": postgres_password,
"ssl": True
}
},
"description": "Application database (read-only)"
}
}
# Create all connections
created_connections = {}
for name, config in connection_configs.items():
connection = create_hex_data_connection(
name=name,
connection_type=config["type"],
connection_details=config["details"],
description=config["description"],
include_magic=True,
allow_writeback_cells=False,
sharing={
"workspace": {
"members": "CAN_USE"
}
}
)
data = json.loads(connection)
created_connections[name] = data["id"]
print(f"Created {name}: {data['id']}")
print(f"\nCreated {len(created_connections)} data connections")数据连接管理说明:
- 支持的连接类型:BigQuery、Snowflake、Postgres、Redshift、Athena、Databricks
- 每种连接类型都有特定的配置要求
connection_details - 连接凭据由Hex安全存储
- 使用共享设置来控制谁可以使用每个连接
CAN_USE访问允许用户查询数据,CAN_ADMIN允许配置更改- 通过支持SSH隧道
connect_via_ssh安全连接参数 - 考虑对敏感数据源使用连接级访问控制
关于这个叉子
此分支扩展了上游hex-mcp包,修复了错误并提供了新的单元操作功能。
Bug修复(v0.1.10上游版)
- 申报的五种工具
-> str返回类型但返回Python字典/列表 - 导致Claude Code MCP集成中的复制验证错误
- 通过添加修复
json.dumps()返回以下语句src/hex_mcp/server.py
固定工具 (第74、172、187、206、229行):
list_hex_projects()get_hex_project()get_hex_run_status()get_hex_project_runs()run_hex_project()
新功能
细胞操作 (完整的Hex API实现的第一阶段):
list_hex_cells()-阅读笔记本结构和单元格源代码(SQL/code单元格)update_hex_cell()-更新SQL/CODE单元格源和数据连接- 全面的测试套件,29项测试通过(全部模拟,无需证书)
- 包含查询迁移和代码重构使用示例的完整文档
权限管理 (完整的Hex API实现的第2阶段):
update_hex_project_user_sharing()-授予或撤销用户对项目的访问权限update_hex_project_group_sharing()-授予或撤销组对项目的访问权限update_hex_project_collection_sharing()-在集合中添加或删除项目update_hex_project_workspace_sharing()-管理工作区范围内的公共访问- 所有共享场景的18个额外通过测试
- 批量权限管理的完整示例
收集管理 (完成Hex API实现的第3阶段):
list_hex_collections()-列出工作区中的所有集合get_hex_collection()-使用共享设置获取收藏详细信息create_hex_collection()-创建具有可选共享功能的集合update_hex_collection()-更新收集元数据和共享- 收集CRUD操作的18个额外通过测试
- 按部门组织工作空间的完整示例
组管理 (完整的Hex API实现的第4阶段):
list_hex_groups()-列出工作区中的所有组get_hex_group()-获取组详细信息create_hex_group()-创建具有可选初始成员的组update_hex_group()-更新组名和成员资格(添加/删除最多100个用户)delete_hex_group()-从工作区删除组- 21项针对全组CRUD操作的额外通过测试
- 基于部门的团队管理和团队重组的完整示例
- 所有功能都包括全面的谷歌风格文档字符串
数据连接管理 (完整的Hex API实现的第5阶段):
list_hex_data_connections()-列出所有带分页和排序的数据连接get_hex_data_connection()-获取连接详细信息,包括凭据和共享create_hex_data_connection()-创建BigQuery、Snowflake、Postgres、Redshift、Athena、ViewModel连接update_hex_data_connection()-更新连接配置、凭据和共享设置- 18个额外的数据连接CRUD操作通过测试
- 连接设置、凭据轮换和批量部署的完整示例
- 支持所有6种主要的数据库/仓库连接类型
- 带有详细连接类型规范的完整谷歌风格文档字符串
测试基础设施:
- 具有模拟身份验证的综合测试夹具
- 105项测试通过,0项跳过(全部模拟,无需证书)
- 测试覆盖率随着每个阶段的增长而增长
- 安全检查以防止测试中的生产凭据
- pytest、pytest-asyncio、pytest-cov集成
- 修复了回退处理程序,可在生产和测试环境中工作
文档:
- 完整的OpenAPI规范分析(参见
OPENAPI-VALIDATION.md) - 详细实施计划(见
IMPLEMENTATION-PLAN.md) - 测试策略指南(见
TEST-STRATEGY.md) - 所有新功能的使用示例
上游: https://github.com/franccesco/hex-mcp
状态:分行 bugfix/fast-mcp-return-types 包括错误修复和新的单元操作功能。如果需要,准备上游PR提交。
