Snowflake数据库操作服务器

一个强大的模型上下文协议(MCP)服务器,为AI助手提供全面的Snowflake数据库操作、Cortex AI服务和数据管理工具。

  • by mcp-tg
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 mcp-tg/snowflake-developer-kit

使用说明

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)

安装

  1. 克隆存储库

    git clone https://github.com/mcp-tg/snowflake-developer.git
    cd snowflake-developer
    
  2. 设置环境

    # Copy environment template
    cp .env.example .env
    
    # Edit .env with your Snowflake credentials
    # Required: SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, SNOWFLAKE_PAT (or SNOWFLAKE_PASSWORD)
    
  3. 安装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包安装,因此您不需要手动安装依赖项。

第一次测试:验证连接

  1. 在检查员,去 工具 标签
  2. 找到 test_snowflake_connection 然后单击 跑
  3. 您应该看到您的帐户详细信息,并确认连接正常

🔌 与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"
      }
    }
  }
}

安装说明:

  1. 克隆存储库: git clone https://github.com/mcp-tg/snowflake-developer.git
  2. 创建Claude Desktop配置文件: ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
  3. 添加上述配置,替换 /path/to/snowflake-developer 与你的实际路径
  4. 将凭据占位符替换为实际的Snowflake凭据
  5. 重新启动克劳德桌面

光标

备注:游标不支持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 < '2023-01-01'
“将2023年1月以来未登录的所有用户的状态设置为非活动”

⚙️ 雪花操作工具(8个工具)

雪花特定操作工具:

工具 描述 检查器中的示例 自然语言查询
alter_warehouse 修改仓库设置 warehouse_name: COMPUTE_WH
仓库大小: MEDIUM
auto_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.USERS
to_type: ROLE
to_name: ANALYST_ROLE
“将TEST_DB.PUBLIC.USERS表上的SELECT和INSERT权限授予ANALYST_ROLE”
revoke_privileges 撤销权限 特权: ["SELECT"]
_类型: TABLE
_名称: TEST_DB.PUBLIC.USERS
from类型: 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连接
  • 每次操作后,连接都会自动关闭
  • 无需连接池或持久性
  • 从环境变量中读取凭据

🛡️ 安全最佳实践

  1. 使用程序化访问令牌(PAT) 尽可能使用密码
  2. 永不承诺 .env 文件 到版本控制
  3. 使用最低权限角色 适用于Snowflake用户
  4. 定期轮换凭据
  5. 考虑使用外部秘密管理 用于生产

🤝 贡献

欢迎投稿!请随时提交拉取请求。

📄 许可证

此项目根据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功能

  1. 装修商注册:使用简单的装饰器注册工具
  2. 类型验证:使用Python类型提示自动验证参数
  3. 上下文管理:用于进度报告和记录的内置上下文
  4. 资源模式:动态资源终结点的URI模板匹配
  5. 错误处理:自动捕获异常和标准化错误响应

FastMCP安装

# Install FastMCP
pip install fastmcp

# Or with UV (recommended)
uv add fastmcp

快速学习MCP

📚 额外资源

雪花资源

MCP协议和工具

开发工具