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.USERS |
alter_type: ADD 列名称: created_at 数据类型: TIMESTAMP |“将created_at时间戳列添加到TEST_DB.PUBLIC.USERS表”| | create_database |创建新数据库|数据库名称: TEST_DB |“创建一个名为TEST_DB的新数据库”| | create_schema |在数据库中创建架构|database_name: TEST_DB schema_name: ANALYTICS |“在TEST_DB数据库中创建名为ANALYTICS的架构”| | create_table |创建包含列的表|database_name: TEST_DB schema_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.id match_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 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功能
1. **装修商注册**:使用简单的装饰器注册工具
1. **类型验证**:使用Python类型提示自动验证参数
1. **上下文管理**:用于进度报告和记录的内置上下文
1. **资源模式**:动态资源终结点的URI模板匹配
1. **错误处理**:自动捕获异常和标准化错误响应
### FastMCP安装
Install FastMCP
pip install fastmcp
Or with UV (recommended)
uv add fastmcp
### 快速学习MCP
- **官方文件**: [FastMCP文档](https://github.com/jlowin/fastmcp)
- **例子**:浏览存储库中的FastMCP示例服务器
- **TypeScript MCP SDK**: [MCP TypeScript SDK](https://github.com/modelcontextprotocol/typescript-sdk)
## 📚 额外资源
### 雪花资源
- [雪花文档](https://docs.snowflake.com/)
- [Snowflake Python连接器](https://docs.snowflake.com/en/developer-guide/python-connector/python-connector)
- [程序化访问令牌](https://docs.snowflake.com/en/user-guide/security-access-tokens)
### MCP协议和工具
- [模型上下文协议规范](https://modelcontextprotocol.io/)
- [MCP TypeScript SDK](https://github.com/modelcontextprotocol/typescript-sdk)
- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk)
- [MCP检查工具](https://github.com/modelcontextprotocol/inspector)
### 开发工具
- [UV包装管理器](https://github.com/astral-sh/uv)
- [FastMCP框架](https://github.com/jlowin/fastmcp)
- [Claude桌面配置](https://claude.ai/docs)
- [游标IDE集成](https://cursor.sh/)