使用说明
Snowflake开发人员MCP服务器🚀
强大的模型上下文协议(MCP)服务器,为Claude等AI助手提供全面的Snowflake数据库操作、Cortex AI服务和数据管理工具。
🌟 特性
- 🔧 DDL操作:创建和管理数据库、模式、表和其他数据库对象
- 📊 DML操作:插入、更新、删除和查询具有完全SQL支持的数据
- ⚙️ 雪花作业:管理仓库、授权、角色和显示数据库对象
- 🔒 安全认证:支持密码和程序化访问令牌(PAT)
- 🎯 简单连接模式:每个操作连接都具有可靠性和简单性
🚀 快速开始
先决条件
- Python 3.11+
- UV包管理器(从安装https://github.com/astral-sh/uv)
- Node.js和npm(用于MCP检查器)
- 具有适当权限的Snowflake帐户
- 雪花凭据(帐户标识符、用户名、密码/PAT)
安装
-
克隆存储库
git clone https://github.com/mcp-tg/snowflake-developer.git cd snowflake-developer -
设置环境
# Copy environment template cp .env.example .env # Edit .env with your Snowflake credentials # Required: SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, SNOWFLAKE_PAT (or SNOWFLAKE_PASSWORD) -
安装UV(如果尚未安装)
# On macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # On Windows powershell -c "irm https://astral.sh/uv/install.ps1 | iex"
🧪 MCP检验员测试
测试设置的最简单方法是使用MCP检查器:
# Run the development inspector script
./dev-inspector.sh
这将:
- ✅ 创建虚拟环境(如果需要)
- ✅ 通过UV安装所有依赖项
- ✅ 从.env加载Snowflake凭据
- ✅ 启动MCP检查器web界面
- ✅ 打开浏览器以交互方式测试工具
注: 该脚本会自动处理UV包安装,因此您不需要手动安装依赖项。
第一次测试:验证连接
- 在检查员,去 工具 标签
- 找到
test_snowflake_connection然后单击 跑 - 您应该看到您的帐户详细信息,并确认连接正常
🔌 与AI助手集成
克劳德桌面
选项1:直接从GitHub(不需要本地克隆)
{
"mcpServers": {
"snowflake-developer": {
"command": "uvx",
"args": [
"--from",
"git+https://github.com/mcp-tg/snowflake-developer.git",
"main.py"
],
"env": {
"SNOWFLAKE_ACCOUNT": "your-account",
"SNOWFLAKE_USER": "your-username",
"SNOWFLAKE_PAT": "your-pat-token"
}
}
}
}
选项2:本地安装
{
"mcpServers": {
"snowflake-developer": {
"command": "uv",
"args": [
"run",
"--directory",
"/path/to/snowflake-developer",
"python",
"main.py"
],
"env": {
"SNOWFLAKE_ACCOUNT": "your-account",
"SNOWFLAKE_USER": "your-username",
"SNOWFLAKE_PAT": "your-pat-token"
}
}
}
}
安装说明:
- 克隆存储库:
git clone https://github.com/mcp-tg/snowflake-developer.git - 创建Claude Desktop配置文件:
~/Library/Application Support/Claude/claude_desktop_config.json(macOS) - 添加上述配置,替换
/path/to/snowflake-developer与你的实际路径 - 将凭据占位符替换为实际的Snowflake凭据
- 重新启动克劳德桌面
光标
备注:游标不支持MCP配置中的环境变量。您需要使用本地安装选项或在系统上全局设置环境变量。
选项1:直接从GitHub(需要全局环境变量)
{
"mcpServers": {
"snowflake-developer": {
"command": "uvx",
"args": [
"--from",
"git+https://github.com/mcp-tg/snowflake-developer.git",
"main.py"
]
}
}
}
需要设置 SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER,以及 SNOWFLAKE_PAT 作为系统环境变量。
选项2:本地安装(建议用于Cursor)
{
"mcpServers": {
"snowflake-developer": {
"command": "uv",
"args": ["run", "/path/to/snowflake-developer/main.py"]
}
}
}
使用本地 .env 使用您的凭据在项目目录中创建文件。
📚 可用工具(共22个)
🔧 DDL工具(8个工具)
管理数据库结构的工具:
| 工具 | 描述 | 检查器中的示例 | 自然语言查询 |
|---|---|---|---|
alter_database |
重命名数据库 | 数据库名称: OLD_DB新名称: NEW_DB |
“将数据库OLD_DB重命名为NEW_DB” |
alter_schema |
重命名或移动架构 | schema_name: TEST_DB.OLD_SCHEMA新名称: NEW_SCHEMA |
“在TEST_DB中将OLD_SCHEMA重命名为NEW_SCHEMA” |
alter_table |
修改表结构 | table_name: TEST_DB.PUBLIC.USERSalter_type: ADD列名称: created_at数据类型: TIMESTAMP |
“将created_at时间戳列添加到TEST_DB.PUBLIC.USERS表” |
create_database |
创建新数据库 | 数据库名称: TEST_DB |
“创建一个名为TEST_DB的新数据库” |
create_schema |
在数据库中创建架构 | database_name: TEST_DBschema_name: ANALYTICS |
“在TEST_DB数据库中创建名为ANALYTICS的架构” |
create_table |
创建包含列的表 | database_name: TEST_DBschema_name: PUBLIC表名: USERS柱: [{"name": "id", "type": "INT"}, {"name": "email", "type": "VARCHAR(255)"}] |
“在TEST_DB.PUBLIC中创建一个USERS表,id为INT,电子邮件为VARCHAR(255)” |
drop_database_object |
删除任何数据库对象 | object_type: TABLE对象名称: TEST_DB.PUBLIC.OLD_TABLE |
“删除表TEST_DB.PUBLIC.OLD_table” |
execute_ddl_statement |
运行自定义DDL SQL | DDL_statement: CREATE VIEW TEST_DB.PUBLIC.ACTIVE_USERS AS SELECT * FROM TEST_DB.PUBLIC.USERS WHERE status = 'active' |
“创建一个名为ACTIVE_USERS的视图,仅显示活动用户” |
📊 DML工具(6个工具)
处理数据的工具:
| 工具 | 描述 | 检查器中的示例 | 自然语言查询 |
|---|---|---|---|
delete_data |
从表中删除行 | table_name: TEST_DB.PUBLIC.USERS其中_原因: status = 'deleted' |
“删除所有状态为“已删除”的用户” |
execute_dml_statement |
运行自定义DML SQL | DML_语句: UPDATE TEST_DB.PUBLIC.USERS SET last_login = CURRENT_TIMESTAMP() WHERE id = 1 |
“更新id为1的用户的上次登录时间戳” |
insert_data |
在表中插入行 | table_name: TEST_DB.PUBLIC.USERS数据: {"id": 1, "email": "john@example.com", "name": "John Doe"} |
“插入id为1的新用户,电子邮件john@example.com,并将John Doe命名到USERS表中” |
merge_data |
在表之间同步数据 | target_table: TEST_DB.PUBLIC.USERS源表: TEST_DB.STAGING.NEW_USERS合并条件: target.id = source.idmatch_actions: [{"action": "UPDATE", "columns": ["email", "name"], "values": ["source.email", "source.name"]}]not_match_actions: [{"action": "INSERT", "columns": ["id", "email", "name"], "values": ["source.id", "source.email", "source.name"]}] |
将临时表中的新用户合并到生产用户表中,更新现有记录并插入新记录 |
query_data |
从表中查询数据 | table_name: TEST_DB.PUBLIC.USERS柱: ["id", "email", "name"]其中_原因: status = 'active'限制: 10 |
“显示前10个活跃用户及其id、电子邮件和姓名” |
update_data |
更新现有行 | table_name: TEST_DB.PUBLIC.USERS数据: {"status": "inactive"}其中_原因: last_login < '2023-01-01' |
“将2023年1月以来未登录的所有用户的状态设置为非活动” |
⚙️ 雪花操作工具(8个工具)
雪花特定操作工具:
| 工具 | 描述 | 检查器中的示例 | 自然语言查询 |
|---|---|---|---|
alter_warehouse |
修改仓库设置 | warehouse_name: COMPUTE_WH仓库大小: MEDIUMauto_ssuspend: 300 |
“将COMPUTE_WH更改为中等大小,并在5分钟后自动挂起” |
describe_database_object |
获取对象详细信息 | object_name: TEST_DB.PUBLIC.USERS |
“描述TEST_DB.PUBLIC.USERS表的结构” |
execute_sql_query |
运行任何SQL查询 | query: SELECT CURRENT_USER(), CURRENT_WAREHOUSE() |
“显示我当前的用户和仓库” |
grant_privileges |
授予权限 | 特权: ["SELECT", "INSERT"]_类型: TABLE_名称: TEST_DB.PUBLIC.USERSto_type: ROLEto_name: ANALYST_ROLE |
“将TEST_DB.PUBLIC.USERS表上的SELECT和INSERT权限授予ANALYST_ROLE” |
revoke_privileges |
撤销权限 | 特权: ["SELECT"]_类型: TABLE_名称: TEST_DB.PUBLIC.USERSfrom类型: ROLE发件人名称: ANALYST_ROLE |
“从ANALYST_ROLE中撤销TEST_DB.PUBLIC.USERS表上的SELECT” |
set_context |
设置数据库/模式/仓库/角色 | context_type: DATABASE上下文名称: TEST_DB |
“使用TEST_DB作为当前数据库” |
show_database_objects |
列出数据库对象 | object_type: DATABASES |
“显示所有数据库” |
test_snowflake_connection |
测试与Snowflake的连接 | (无参数) | “测试我的Snowflake连接” |
🏗️ 建筑
服务器使用简单的按操作连接模式:
- 每次工具/资源调用都会创建一个新的Snowflake连接
- 每次操作后,连接都会自动关闭
- 无需连接池或持久性
- 从环境变量中读取凭据
🛡️ 安全最佳实践
- 使用程序化访问令牌(PAT) 尽可能使用密码
- 永不承诺
.env文件 到版本控制 - 使用最低权限角色 适用于Snowflake用户
- 定期轮换凭据
- 考虑使用外部秘密管理 用于生产
🤝 贡献
欢迎投稿!请随时提交拉取请求。
📄 许可证
此项目根据MIT许可证获得许可-有关详细信息,请参阅许可证文件。
🆘 故障排除
连接问题
- 验证您的帐户标识符格式
- 检查您的用户是否具有适当的权限
- 确保您的PAT令牌尚未过期
- 测试与Snowflake的网络连接
工具错误
- 检查检查器控制台中的错误消息
- 验证是否提供了所需的参数
- 在引用数据库对象之前,确保它们存在
- 检查自定义语句的SQL语法
🚀 FastMCP框架
此MCP服务器是使用 FastMCP,一个简化构建模型上下文协议服务器的现代Python框架。FastMCP提供:
为什么选择FastMCP?
- 🎯 简单API:基于装饰器的工具和资源注册
- ⚡ 高性能:异步/等待支持,具有高效的消息处理能力
- 🔧 类型安全:完整的TypeScript风格类型提示和验证
- 📝 汽车文档:自动生成工具/资源文档
- 🛡️ 错误处理:内置异常处理和响应格式
- 🔌 MCP合规性:与MCP协议规范完全兼容
快速MCP与传统MCP
# Traditional MCP server setup
class MyMCPServer:
def __init__(self):
self.tools = {}
def register_tool(self, name, handler, schema):
# Manual registration and validation
pass
# FastMCP - Clean and Simple
from fastmcp import FastMCP
mcp = FastMCP("MyServer")
@mcp.tool()
def my_tool(param: str) -> str:
"""Tool with automatic type validation and documentation."""
return f"Result: {param}"
@mcp.resource("my://resource/{id}")
async def my_resource(id: str, ctx: Context) -> dict:
"""Resource with built-in async support and context."""
return {"data": f"Resource {id}"}
使用的关键FastMCP功能
- 装修商注册:使用简单的装饰器注册工具
- 类型验证:使用Python类型提示自动验证参数
- 上下文管理:用于进度报告和记录的内置上下文
- 资源模式:动态资源终结点的URI模板匹配
- 错误处理:自动捕获异常和标准化错误响应
FastMCP安装
# Install FastMCP
pip install fastmcp
# Or with UV (recommended)
uv add fastmcp
快速学习MCP
- 官方文件: FastMCP文档
- 例子:浏览存储库中的FastMCP示例服务器
- TypeScript MCP SDK: MCP TypeScript SDK