数据GraphQL代理
MCP（模型上下文协议）代理，能够从BigQuery SQL查询中生成生产就绪的Apollo GraphQL服务器，并支持Dataplex血缘追踪。
特点/功能
- 🚀 表情符号“🚀”在中文中通常被翻译为“火箭”或“发射”，用来表示快速前进、进步或创新等含义。所以，这个表情可以翻译为“🚀 火箭”或“🚀 发射”，具体取决于上下文和使用场景。 自动生成Apollo GraphQL服务器 来自BigQuery查询
- 📊（图表） BigQuery 集成 具有从SQL模式中推断类型的功能
- 📝 Dataplex血缘追踪 用于端到端的数据治理
- 🐳 这个表情符号通常代表“哭泣的海豚”或“哭泣的鲸鱼”，在中文中可以简单地翻译为“哭泣的海豚”或根据语境灵活表达为类似“海豚哭了”、“鲸鱼在哭泣”等意思。 Docker 支持 用于容器化部署
- 🧪 表示“试管”或“实验”。 测试客户端生成 用于API验证
- 🔌 电源插头 MCP协议 以便与Cursor及其他AI助手实现无缝集成
其工作原理
端到端流程
1. Input → 2. Schema Inference → 3. Code Generation → 4. Validation → 5. Output
BigQuery SQL Dry-run Analysis Jinja2 Templates Multi-level GCS/Local
Queries Type Mapping Apollo Server v4 Checks Files详细步骤：
- 输入您通过MCP工具提供BigQuery SQL查询
- 模式推断代理运行BigQuery预演以推断结果类型
- 代码生成使用模板生成完整的Apollo Server项目
- 验证 （可选）：在选定级别验证生成的代码
- 输出将经过验证的代码写入GCS或本地文件系统
- 部署你运行生成的 Node.js 应用程序
验证级别
根据您的需求选择验证的详尽程度：
| 级别 | 时间 | 覆盖范围 | 检查项 | 使用场景 |
|---|---|---|---|---|
| 快速 | ~1秒 | 80% | GraphQL语法，SQL干运行，文件结构 | 快速迭代，开发 |
| 标准 | ~10秒 | 95% | 快速 + TypeScript 编译，导入 | 默认，均衡方法 |
| 满的 | ~60秒 | 99% | 标准+Docker构建、服务器启动、健康检查 | 预生产、持续集成/持续交付 (CI/CD) |
建筑学
该代理生成一个包含以下内容的完整TypeScript/Node.js项目：
- Apollo 服务器 v4 - 带插件和上下文的GraphQL API服务器
- 类型安全的解析器 - 由BigQuery模式自动生成
- Dataplex集成 - 运行时血统事件追踪
- 错误处理 - 生产环境安全的错误格式化
- Docker 配置 - 用于生产的多阶段构建
- 测试套件 - 集成测试和测试客户端
安装
先决条件
- Python 3.10-3.12
- 诗歌（Python依赖管理）
- 拥有BigQuery访问权限的Google Cloud账户
设置
# Clone the repository
git clone https://github.com/opendedup/data-graphql-agent.git
cd data-graphql-agent
# Install dependencies
poetry install
# Configure environment variables
cp .env.example .env
# Edit .env with your GCP credentials配置
创建一个 .env 文件或设置环境变量：
# GCP Configuration
GCP_PROJECT_ID=your-project-id
GCP_LOCATION=us-central1
# Output Configuration
GRAPHQL_OUTPUT_DIR=gs://your-bucket/graphql-server
# Or local path: GRAPHQL_OUTPUT_DIR=/path/to/output
# MCP Server Configuration
MCP_TRANSPORT=stdio # or http
MCP_HOST=0.0.0.0
MCP_PORT=8080使用
作为MCP服务器（推荐）
在 Cursor 中进行配置 mcp.json：
{
"mcpServers": {
"data-graphql-agent": {
"command": "poetry",
"args": ["run", "python", "-m", "data_graphql_agent.mcp"],
"cwd": "/path/to/data-graphql-agent",
"env": {
"GCP_PROJECT_ID": "your-project",
"GRAPHQL_OUTPUT_DIR": "gs://your-bucket/graphql-server"
}
}
}
}直接使用Python
from data_graphql_agent.generation import ProjectGenerator
from data_graphql_agent.clients import StorageClient
from data_graphql_agent.models import QueryInput
# Define queries
queries = [
QueryInput(
query_name="trendingItems",
sql="SELECT item, SUM(sales) as total FROM `project.dataset.sales` GROUP BY item",
source_tables=["project.dataset.sales"]
)
]
# Generate project
generator = ProjectGenerator(project_id="your-project")
files = generator.generate_project("my-project", queries)
# Write to storage
storage = StorageClient(project_id="your-project")
manifests = storage.write_files("gs://bucket/output", files)作为HTTP服务器运行
# Set transport to HTTP
export MCP_TRANSPORT=http
export MCP_PORT=8080
# Start server
poetry run python -m data_graphql_agent.mcp然后通过HTTP调用工具：
curl -X POST http://localhost:8080/mcp/call-tool \
-H "Content-Type: application/json" \
-d '{
"name": "generate_graphql_api",
"arguments": {
"queries": [...],
"project_name": "my-project"
}
}'MCP 工具
generate_graphql_api
生成一个带有验证功能的完整Apollo GraphQL服务器项目。
输入：
queries查询对象的数组，其中包含queryName，sql，和source_tablesproject_name用于血统追踪的项目名称output_path可选的输出位置（默认为 GRAPHQL_OUTPUT_DIR）validation_level可选的验证详细程度 -"quick"，"standard"（默认），或"full"auto_fix可选布尔值，用于尝试自动修复错误（默认：false)
输出：
- 完整的TypeScript/Node.js项目
- Docker 配置
- 测试客户端
- 集成测试
- 验证结果：检查通过，存在警告
带验证的示例：
result = await handle_generate_graphql_api({
"queries": [
{
"queryName": "salesByRegion",
"sql": "SELECT region, SUM(amount) as total FROM `project.dataset.sales` GROUP BY region",
"source_tables": ["project.dataset.sales"]
}
],
"project_name": "analytics-api",
"output_path": "./output",
"validation_level": "standard", # Quick validation for speed
"auto_fix": false
})成功响应：
{
"success": true,
"output_path": "./output",
"files_generated": [...],
"message": "Successfully generated and validated Apollo GraphQL Server with 1 queries. Generated 15 files at ./output. Validation: 5 checks passed in 8.2s"
}验证失败响应：
{
"success": false,
"output_path": "./output",
"files_generated": [],
"message": "Code validation failed at standard level",
"error": "Validation errors: Invalid SQL in query 'salesByRegion': Table not found; TypeScript compilation failed"
}validate_graphql_schema
验证GraphQL模式文件。
输入：
schema_path模式文件的路径
输出：
- 包含错误和警告的验证结果
生成的项目结构
graphql-server/
├── src/
│ ├── server.ts # Main Apollo Server
│ ├── typeDefs.ts # GraphQL schema
│ ├── resolvers.ts # Query resolvers
│ └── lineage.ts # Dataplex integration
├── test-client/ # Test client
├── tests/ # Integration tests
├── package.json
├── tsconfig.json
├── Dockerfile
└── docker-compose.yml运行生成的服务器
cd output/graphql-server
# Install dependencies
npm install
# Development mode
npm run dev
# Production build
npm run build
npm start
# Docker
docker-compose up --build发展
运行测试
# Run all tests
poetry run pytest
# Run unit tests only
poetry run pytest tests/unit
# Run with coverage
poetry run pytest --cov=data_graphql_agent代码格式化
# Format with Black
poetry run black src tests
# Lint with Ruff
poetry run ruff check src testsBigQuery 类型映射
代理自动将 BigQuery 类型映射到 GraphQL 类型：
| BigQuery 类型 | GraphQL 类型 |
|---|---|
| 字符串 | 字符串 |
| 64位整数 | Int |
| FLOAT64 | 浮点数 |
| BOOL | 布尔（Boolean） |
| 时间戳/日期 | 字符串（ISO 8601） |
| 结构体 | 自定义对象类型 |
| 数组 | \[类型\] |
嵌套结构（STRUCTs 和 ARRAYs）得到全面支持，并自动进行类型生成。
验证的好处
为什么在编写之前要进行验证？
- 尽早发现错误 - 在部署前检测到无效的SQL、类型不匹配和语法错误
- 更快的迭代 - 无需手动调试生成的代码
- 自信 - 在运行之前确保你的代码能正常工作
npm install - 成本节约 - 避免因代码错误而浪费GCS写入和Docker构建资源
- 适合CI/CD（持续集成/持续交付） - 使用
full在流水线中进行验证以确保部署可靠
何时使用哪个级别？
快速验证（~1秒）
- ✅ 快速原型制作和实验
- ✅ 对SQL查询进行迭代优化
- ✅ 测试查询到模式的映射
- ❌ 不适用于生产环境部署
标准验证（约10秒） - 推荐默认值
- ✅ 正常的开发工作流程
- ✅ 在提交到版本控制之前
- ✅ 速度与细致程度的平衡
- ✅ 最常见的用例
完整验证（~60秒）
- ✅ 预生产部署
- ✅ 持续集成/持续交付（CI/CD）流水线
- ✅ 关键生产更新
- ✅ 当Docker兼容性至关重要时
- ❌ 快速迭代速度太慢
数据血缘
生成的GraphQL服务器会自动在Google Cloud Dataplex中追踪数据血缘：
- 过程每个解析器都被注册为一个进程
- 跑每次查询执行都会创建一个运行（带有唯一请求ID）
- “Lineage Events”可以翻译为“血统事件”或“族谱事件”，具体翻译取决于上下文和语境。在一般情况下，如果是指与家族血统或族谱相关的活动或事件，可以采用“血统事件”这一翻译。如果是指某种特定类型的活动或仪式，且与血统或族谱有直接关联，也可以采用“族谱事件”这一翻译将BigQuery数据源链接到BI报告目标
- 清理优雅关闭会移除血缘关系进程
血缘操作是异步的（即发即弃）且不会阻塞API响应。
许可证
Apache 2.0 - 查看 许可证 详情请见
做出贡献
欢迎贡献！请提交拉取请求或针对错误和功能需求打开问题。