hex-mcp MCP server

hex-mcp MCP 服务

一个用于 Hex 的 MCP 服务，提供编排、监控、单元格内容访问、权限管理、集合整理和群组管理工具。

功能与限制

实际使用场景

编排与自动化：

• 从外部系统（Airflow DAG、CI/CD 流水线）触发 Hex 项目运行
• 以编程方式监控运行状态
• 取消长时间运行或卡住的执行
• 跨工作区发现和搜索项目

运维监控：

• 检查计划运行是否成功完成
• 获取运行历史以供审计
• 以编程方式访问项目元数据（所有者、上次编辑时间、描述）

单元格内容访问（新增）：

• 读取笔记本结构和单元格元数据
• 从 SQL 单元格读取 SQL 查询源代码
• 从 CODE 单元格读取 Python/R 代码
• 适用于查询迁移、代码分析和内容审计

权限管理（新增）：

• 以编程方式管理用户和群组对项目的访问权限
• 批量更新多个项目的权限
• 管理整个工作区和公开访问设置
• 在集合中添加或移除项目

集合整理（新增）：

• 创建和管理集合，以便整理项目
• 按部门、团队或主题整理项目
• 控制集合的可见性和访问权限
• 将项目批量整理到集合中

群组管理（新增）：

• 创建和管理用户群组，以便管理权限
• 批量添加或移除群组中的用户
• 使用群组简化跨项目的权限管理
• 维护井然有序的团队架构

数据连接管理（新增）：

• 列出和管理数据库及数据仓库连接
• 创建 BigQuery、Snowflake、Postgres、Redshift、Athena、Databricks 连接
• 更新连接配置和凭据
• 控制整个工作区的数据连接共享

重要限制

笔记本内容访问受限：

• ✅ 可以读取 SQL 和 CODE 单元格的源内容
• ❌ 无法读取 MARKDOWN、INPUT 或可视化单元格
• ❌ 无法创建或删除单元格
• ❌ 无法修改笔记本结构
• ❌ 无法查看查询结果或图表
• ❌ 无法管理笔记本依赖项或参数

不适用于：

• 从头开始构建或编写笔记本
• 协作开发笔记本
• 调试查询或代码执行
• 完整备份笔记本（只能访问 SQL/CODE 单元格）
• 读取 Markdown 文档或输入参数

适用场景

在以下情况下使用 hex-mcp：

• 从外部系统编排 Hex 执行
• 监控运行状态和历史记录
• 从现有笔记本中读取 SQL 查询和代码
• 跨项目审计或迁移 SQL/CODE 内容
• 跨项目管理权限和访问控制
• 为用户、群组和集合自动化批量权限更新
• 按部门、团队或主题将项目整理到集合中
• 维护用户群组，简化权限管理
• 规范化 Hex 实例中的工作区组织
• 以编程方式管理数据连接和数据库集成
• 为新工作区或环境自动化连接设置

如需全面开发和编辑笔记本，请直接使用 Hex 网页界面。

可用工具

项目操作

• list_hex_projects：列出可用的 Hex 项目
• search_hex_projects：按模式搜索 Hex 项目
• get_hex_project：获取特定项目的详细信息

项目执行

• run_hex_project：执行 Hex 项目
• 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 命令（推荐）

配置 hex-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），以便所有 hex-mcp 调用都能使用该配置。

使用环境变量

也可以通过环境变量配置 Hex MCP 服务：

• HEX_API_KEY：你的 Hex API 密钥
• HEX_API_URL：Hex API 基础 URL

为 MCP 服务设置环境变量时，需要将其设为全局变量，以便 Cursor 能读取；或者在调用服务时使用 uv 的 --env-file 标志。

在 Cursor 中使用

Cursor 支持 AI Agent 通过 MCP 协议与 Hex 交互。按照以下步骤在 Cursor 中设置并使用 hex-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（Agent）对话，并让它列出或运行 Hex 项目。

重要

MCP 服务和 CLI 仍在开发中，未来可能会有不兼容的变更。

使用示例

读取单元格内容

使用 list_hex_cells 读取 Hex 笔记本的结构和源代码：

# 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 和可视化单元格的 sqlCell 和 codeCell 均返回 null。

分页

对于包含大量单元格的项目，可以使用分页：

# 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 必须存在于你的 Hex 工作区中
• 集合 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 权限允许用户更改配置
• 可通过 connect_via_ssh 参数使用 SSH 隧道建立安全连接
• 对敏感数据源，可考虑使用连接级访问控制

关于此 Fork

此 Fork 在上游 hex-mcp 软件包的基础上扩展了错误修复和新的单元格操作功能。

错误修复（上游 v0.1.10）

• 有五个工具声明返回类型为 -> str，但实际返回的是 Python 字典／列表
• 这会导致 Claude Code MCP 集成中出现 Pydantic 验证错误
• 已在 src/hex_mcp/server.py 的返回语句中添加 json.dumps() 修复

已修复的工具（第 74、172、187、206、229 行）：

• list_hex_projects()
• get_hex_project()
• get_hex_run_status()
• get_hex_project_runs()
• run_hex_project()

新功能

单元格操作（完整 Hex API 实现的第 1 阶段）：

• 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() - 从工作区中删除群组
• 为完整的群组 CRUD 操作新增 21 项通过的测试
• 提供按部门管理群组和重组团队的完整示例
• 所有函数均包含全面的 Google 风格文档字符串

数据连接管理（完整 Hex API 实现的第 5 阶段）：

• list_hex_data_connections() - 使用分页和排序列出所有数据连接
• get_hex_data_connection() - 获取包含凭据和共享设置的连接详情
• create_hex_data_connection() - 创建 BigQuery、Snowflake、Postgres、Redshift、Athena、Databricks 连接
• update_hex_data_connection() - 更新连接配置、凭据和共享设置
• 为数据连接 CRUD 操作新增 18 项通过的测试
• 提供连接设置、凭据轮换和批量部署的完整示例
• 支持全部 6 种主要数据库/数据仓库连接类型
• 提供完整的 Google 风格文档字符串，详细说明各连接类型

测试基础设施：

• 全面的测试 fixture，使用模拟身份验证
• 105 项测试通过，0 项跳过（全部使用模拟，无需凭据）
• 随各阶段推进不断提升测试覆盖率
• 安全检查，防止在测试中使用生产凭据
• 集成 pytest、pytest-asyncio、pytest-cov
• 修复 backoff 处理程序，使其在生产和测试环境中均可运行

文档：

• 完整的 OpenAPI 规范分析（参见 OPENAPI-VALIDATION.md）
• 详细的实施计划（参见 IMPLEMENTATION-PLAN.md）
• 测试策略指南（参见 TEST-STRATEGY.md）
• 所有新功能的使用示例

上游项目：https://github.com/franccesco/hex-mcp

状态：分支 bugfix/fast-mcp-return-types 同时包含错误修复和新的单元格操作功能。若有需要，可提交上游 PR。