mcp-datahub

一个 MCP 服务和可组合的 Go 库，可将 AI 助手连接到 DataHub 元数据目录。搜索数据集、探索架构、追踪血缘关系，并访问术语表和域。

mcp-datahub.txn2.com | 安装 | 库文档

MCP 数据平台生态系统

mcp-datahub 是一套更广泛的开源 MCP 服务之一，旨在协同构建可组合的数据平台。每个组件都可以独立运行，也可以组合使用，让 AI 助手统一访问存储、查询引擎和元数据目录。

• txn2/mcp-data-platform
• txn2/mcp-s3
• txn2/mcp-trino

两种使用方式

1. 独立 MCP 服务

安装并连接到 Claude Desktop、Cursor 或任何 MCP 客户端：

Claude Desktop（最简单） - 从 releases 下载 .mcpb 安装包，然后双击安装：

• macOS Apple Silicon：mcp-datahub_X.X.X_darwin_arm64.mcpb
• macOS Intel：mcp-datahub_X.X.X_darwin_amd64.mcpb
• Windows：mcp-datahub_X.X.X_windows_amd64.mcpb

其他安装方式：

# Homebrew (macOS)
brew install txn2/tap/mcp-datahub

# Go install
go install github.com/txn2/mcp-datahub/cmd/mcp-datahub@latest

手动配置 Claude Desktop（不使用 MCPB 时）：

{
  "mcpServers": {
    "datahub": {
      "command": "/opt/homebrew/bin/mcp-datahub",
      "env": {
        "DATAHUB_URL": "https://datahub.example.com",
        "DATAHUB_TOKEN": "your_token"
      }
    }
  }
}

多服务配置

同时连接多个 DataHub 实例：

# Primary server
export DATAHUB_URL=https://prod.datahub.example.com/api/graphql
export DATAHUB_TOKEN=prod-token
export DATAHUB_CONNECTION_NAME=prod

# Additional servers (JSON)
export DATAHUB_ADDITIONAL_SERVERS='{"staging":{"url":"https://staging.datahub.example.com/api/graphql","token":"staging-token"}}'

使用 datahub_list_connections 发现可用连接，然后将 connection 参数传递给任意工具。

2. 可组合的 Go 库

将其导入自己的 MCP 服务，以实现自定义身份验证、租户隔离和审计日志：

import (
    "github.com/txn2/mcp-datahub/pkg/client"
    "github.com/txn2/mcp-datahub/pkg/tools"
)

// Create client and register tools with your MCP server
datahubClient, _ := client.NewFromEnv()
defer datahubClient.Close()

toolkit := tools.NewToolkit(datahubClient, tools.Config{})
toolkit.RegisterAll(yourMCPServer)

自定义工具描述

覆盖工具描述，以匹配你的部署：

toolkit := tools.NewToolkit(datahubClient, tools.Config{},
    tools.WithDescriptions(map[tools.ToolName]string{
        tools.ToolSearch: "Search our internal data catalog for datasets and dashboards",
    }),
)

自定义工具注释

覆盖 MCP 工具注释（面向 AI 客户端的行为提示）：

toolkit := tools.NewToolkit(datahubClient, tools.Config{},
    tools.WithAnnotations(map[tools.ToolName]*mcp.ToolAnnotations{
        tools.ToolSearch: {ReadOnlyHint: true, OpenWorldHint: boolPtr(true)},
    }),
)

所有 12 个工具均附带默认注释：读取工具标记为 ReadOnlyHint: true；datahub_create 是非破坏性且非幂等的；datahub_update 是非破坏性且幂等的；datahub_delete 是破坏性且幂等的。

扩展（日志记录、指标、错误提示）

通过 extensions 包启用可选中间件：

import "github.com/txn2/mcp-datahub/pkg/extensions"

// Load from environment variables (MCP_DATAHUB_EXT_*)
cfg := extensions.FromEnv()
opts := extensions.BuildToolkitOptions(cfg)
toolkit := tools.NewToolkit(datahubClient, toolsCfg, opts...)

// Or load from a YAML/JSON config file
serverCfg, _ := extensions.LoadConfig("config.yaml")

有关中间件、选择性工具注册和企业级模式，请参阅库文档。

与 mcp-trino 组合使用

将 DataHub 元数据与 Trino 查询执行相结合，构建统一的数据平台 MCP 服务：

import (
    datahubClient "github.com/txn2/mcp-datahub/pkg/client"
    datahubTools "github.com/txn2/mcp-datahub/pkg/tools"
    trinoClient "github.com/txn2/mcp-trino/pkg/client"
    trinoTools "github.com/txn2/mcp-trino/pkg/tools"
)

// Add DataHub tools (search, lineage, schema, glossary)
dh, _ := datahubClient.NewFromEnv()
datahubTools.NewToolkit(dh, datahubTools.Config{}).RegisterAll(server)

// Add Trino tools (query execution, catalog browsing)
tr, _ := trinoClient.NewFromEnv()
trinoTools.NewToolkit(tr, trinoTools.Config{}).RegisterAll(server)

// AI assistants can now:
// - Search DataHub for tables -> Get schema -> Query via Trino
// - Explore lineage -> Understand data flow -> Run validation queries

有关配套库，请参阅 txn2/mcp-trino。

与 QueryProvider 双向集成

该库支持双向上下文注入。mcp-trino 可以从 DataHub 拉取语义上下文，而 mcp-datahub 可以从查询引擎接收查询执行上下文：

import (
    datahubTools "github.com/txn2/mcp-datahub/pkg/tools"
    "github.com/txn2/mcp-datahub/pkg/integration"
)

// QueryProvider enables query engines to inject context into DataHub tools
type myQueryProvider struct {
    trinoClient *trino.Client
}

func (p *myQueryProvider) Name() string { return "trino" }

func (p *myQueryProvider) ResolveTable(ctx context.Context, urn string) (*integration.TableIdentifier, error) {
    // Map DataHub URN to Trino table (catalog.schema.table)
    return &integration.TableIdentifier{
        Catalog: "hive", Schema: "production", Table: "users",
    }, nil
}

func (p *myQueryProvider) GetTableAvailability(ctx context.Context, urn string) (*integration.TableAvailability, error) {
    // Check if table is queryable
    return &integration.TableAvailability{Available: true}, nil
}

func (p *myQueryProvider) GetQueryExamples(ctx context.Context, urn string) ([]integration.QueryExample, error) {
    // Return sample queries for this entity
    return []integration.QueryExample{
        {Name: "sample", SQL: "SELECT * FROM hive.production.users LIMIT 10"},
    }, nil
}

// Wire it up
toolkit := datahubTools.NewToolkit(datahubClient, config,
    datahubTools.WithQueryProvider(&myQueryProvider{trinoClient: trino}),
)

配置 QueryProvider 后，工具响应会得到增强：

• 搜索结果：包含带有表可用性信息的 query_context
• 实体详情：包含 query_table、query_examples、query_availability
• 架构：包含 query_table，可立即用于 SQL
• 血缘：包含将 URN 映射到表的 execution_context

集成中间件

访问控制和审计日志等企业功能通过中间件适配器启用：

import (
    datahubTools "github.com/txn2/mcp-datahub/pkg/tools"
    "github.com/txn2/mcp-datahub/pkg/integration"
)

// Access control - filter entities by user permissions
type myAccessFilter struct{}
func (f *myAccessFilter) CanAccess(ctx context.Context, urn string) (bool, error) { /* ... */ }
func (f *myAccessFilter) FilterURNs(ctx context.Context, urns []string) ([]string, error) { /* ... */ }

// Audit logging - track all tool invocations
type myAuditLogger struct{}
func (l *myAuditLogger) LogToolCall(ctx context.Context, tool string, params map[string]any, userID string) error { /* ... */ }

// Wire up with multiple integration options
toolkit := datahubTools.NewToolkit(datahubClient, config,
    datahubTools.WithAccessFilter(&myAccessFilter{}),
    datahubTools.WithAuditLogger(&myAuditLogger{}, func(ctx context.Context) string {
        return ctx.Value("user_id").(string)
    }),
    datahubTools.WithURNResolver(&myURNResolver{}),      // Map external IDs to URNs
    datahubTools.WithMetadataEnricher(&myEnricher{}),    // Add custom metadata
)

有关完整的集成模式，请参阅库文档。

可用工具

读取工具（始终可用）

工具	描述
datahub_search	按查询和实体类型搜索数据集、仪表板和管道
datahub_get_entity	按 URN 获取实体元数据（描述、所有者、标签、域）
datahub_get_schema	获取包含字段类型和描述的数据集架构
datahub_get_lineage	获取上游／下游血缘关系（支持列级别的 level=column）
datahub_get_queries	获取与数据集关联的 SQL 查询
datahub_browse	浏览目录：列出标签、域或数据产品
datahub_get_glossary_term	获取术语表词条定义和属性
datahub_get_data_product	获取数据产品详情（所有者、域、属性）
datahub_list_connections	列出已配置的 DataHub 服务连接（多服务模式）

写入工具（需要 DATAHUB_WRITE_ENABLED=true）

使用 what 判别模式的 3 个 CRUD 工具——共 37 项操作：

工具	操作数	描述
datahub_create	10	创建标签、域、术语表词条、数据产品、文档、应用、查询、事件、结构化属性、数据契约
datahub_update	19	更新描述（包括标签／术语表词条描述）、标签、术语表词条、链接、所有者、域、结构化属性、自定义属性、事件、查询、文档、数据契约
datahub_delete	8	删除查询、标签、域、术语表实体、数据产品、应用、文档、结构化属性

出于安全考虑，写入工具默认处于禁用状态。

DataHub 版本兼容性

最低要求：DataHub 1.3.x。完整功能集：DataHub 1.4.x。

DataHub 版本	功能
1.3.x+（最低要求）	所有读取工具；除文档外的所有写入操作（标签、域、术语表、数据产品、查询、所有者、链接、描述、事件、应用、结构化属性（包括删除）、数据契约）
1.4.x+（完整功能）	+ 文档（创建／更新／删除）

客户端会妥善处理版本差异——旧版本不支持的功能，其读取查询会返回空结果（而非错误）。

详细文档请参阅工具参考。

配置

变量	描述	默认值
DATAHUB_URL	DataHub GraphQL API URL	（必填）
DATAHUB_TOKEN	API 令牌	（必填）
DATAHUB_TIMEOUT	请求超时时间（秒）	30
DATAHUB_DEFAULT_LIMIT	默认搜索条数限制	10
DATAHUB_MAX_LIMIT	最大条数限制	100
DATAHUB_CONNECTION_NAME	主连接的显示名称	datahub
DATAHUB_ADDITIONAL_SERVERS	其他服务的 JSON 映射	（可选）
DATAHUB_WRITE_ENABLED	启用写入操作（true 或 1）	false
DATAHUB_DEBUG	启用调试日志（1 或 true）	false

扩展

变量	描述	默认值
MCP_DATAHUB_EXT_LOGGING	启用工具调用的结构化日志记录	false
MCP_DATAHUB_EXT_METRICS	启用指标收集	false
MCP_DATAHUB_EXT_METADATA	启用结果的元数据丰富处理	false
MCP_DATAHUB_EXT_ERRORS	启用错误提示丰富处理	true

配置文件

也可以通过 YAML 或 JSON 配置，而不使用环境变量：

datahub:
  url: https://datahub.example.com
  token: "${DATAHUB_TOKEN}"
  timeout: "30s"
  write_enabled: true

toolkit:
  default_limit: 20
  descriptions:
    datahub_search: "Custom search description for your deployment"

extensions:
  logging: true
  errors: true

使用 extensions.LoadConfig("config.yaml") 加载。对于敏感字段，环境变量会覆盖文件中的值。Token 值支持 $VAR / ${VAR} 展开。

如需查看所有选项，请参阅配置参考。

开发

make build     # Build binary
make test      # Run tests with race detection
make lint      # Run golangci-lint
make security  # Run gosec and govulncheck
make coverage  # Generate coverage report
make verify    # Run tidy, lint, and test
make help      # Show all targets

相关项目

• txn2/mcp-trino（文档）- 用于执行 Trino 查询的可组合 MCP 工具包
• DataHub - 开源元数据平台

参与贡献

指南请参阅 CONTRIBUTING.md。

许可证

Apache License 2.0

---

由 Craig Johnston 开源，赞助方为 Deasil Works, Inc.