FDE SQL MCP
使用Windows身份验证的本地SQL server的MCP服务器。它从一个列出数据库的工具开始,并为添加更多SQL工具提供了一个干净的基础。
______________________________________________________________________
目的
当您需要MCP工具通过Windows身份验证(Trusted_Connection)与本地SQL server实例交互时,请使用此服务器。它连接到配置的服务器,并为MCP客户端返回JSON格式的结果。
______________________________________________________________________
特性
- 简单的stdio MCP服务器
- Windows身份验证SQL Server连接(Trusted_connection)
- 用于列出数据库、模式对象和每个数据库的索引的工具
______________________________________________________________________
需求
- Python 3.10+
- 访问通过配置的SQL Server主机
fde_sql_mcp.config.json或SQL_SERVER_HOST - 运行MCP服务器的帐户的Windows身份验证权限
- 已安装SQL Server的ODBC驱动程序17或18
______________________________________________________________________
安装
# create & activate a venv (required so Windows auth works predictably)
python -m venv .venv
.venv\Scripts\activate
# install the MCP client/programming helpers so the MCP runtime is available
pip install mcp
# install the project dependencies
pip install -r requirements.txt
# install this package in editable mode so imports work from the repo root
pip install -e .______________________________________________________________________
配置
创建本地 fde_sql_mcp.config.json repo根目录下的文件(复制 fde_sql_mcp.config.template.json)并用实际的SQL Server端点和任何覆盖来填充它。模板已提交,工作文件被忽略,加载程序更喜欢本地文件,因此私有IP永远不会进入版本控制。
{
"sql_server": "your.sql.server.address",
"sql_server_port": 1433,
"sql_database": "master",
"sql_driver": "{ODBC Driver 17 for SQL Server}",
"sql_application_intent": "ReadOnly",
"sql_encrypt": true,
"sql_trust_server_certificate": true,
"sql_connection_timeout": 30,
"sql_query_timeout": 30,
"sql_max_rows": 200,
"sql_max_query_chars": 10000,
"sql_enforce_readonly": true,
"fabric_enabled": false,
"fabric_tenant_id": "",
"fabric_client_id": "",
"fabric_client_secret": "",
"fabric_auth_fallback_mode": "default_browser",
"fabric_default_workspace": "",
"fabric_default_database": "",
"fabric_workspace_id_map": {
"fde_core_data_dev": "",
"fde_core_data_stg": "",
"fde_core_data_prod": ""
},
"fabric_sql_endpoint_map": {
"fde_core_data_dev": {
"warehouse": {
"server": "dev-warehouse.sql.fabric.microsoft.com",
"database": "core_dw",
"user": "",
"password": ""
},
"lakehouse": {
"server": "dev-lakehouse.sql.fabric.microsoft.com",
"database": "core_lh",
"user": "",
"password": ""
}
}
},
"fabric_allowed_workspaces": ["fde_core_data_dev", "fde_core_data_stg", "fde_core_data_prod"],
"fabric_allowed_databases": ["core_dw", "core_lh"]
}如果本地文件丢失, SQL_SERVER_HOST 必须在环境中设置(以下其他名称仍可以覆盖文件值或独立工作):
SQL_SERVER_HOST=
SQL_SERVER_PORT= # optional
SQL_SERVER_DATABASE=master
SQL_DRIVER={ODBC Driver 17 for SQL Server}
SQL_APPLICATION_INTENT=ReadOnly
SQL_ENCRYPT=true
SQL_TRUST_SERVER_CERTIFICATE=true
SQL_CONNECTION_TIMEOUT=30
SQL_QUERY_TIMEOUT=30
SQL_MAX_ROWS=200
SQL_MAX_QUERY_CHARS=10000
SQL_ENFORCE_READONLY=true
FABRIC_ENABLED=false
FABRIC_TENANT_ID=
FABRIC_CLIENT_ID=
FABRIC_CLIENT_SECRET=
FABRIC_AUTH_FALLBACK_MODE=default_browser
FABRIC_DEFAULT_WORKSPACE=
FABRIC_DEFAULT_DATABASE=
FABRIC_WORKSPACE_ID_MAP= # JSON object or comma pairs (workspace:id)
FABRIC_SQL_ENDPOINT_MAP= # JSON object keyed by workspace with warehouse/lakehouse server+database (+ optional user/password)
FABRIC_ALLOWED_WORKSPACES=fde_core_data_dev,fde_core_data_stg,fde_core_data_prod
FABRIC_ALLOWED_DATABASES=core_dw,core_lh笔记:
- 本地SQL使用Windows身份验证(
Trusted_Connection=yes). - 结构SQL端点身份验证源自
fabric_sql_endpoint_map:
- 如果 user 和 password 提供,使用SQL身份验证; - 如果省略,则使用结构身份验证模式: - client_secret 使用服务主体凭据(FABRIC_CLIENT_ID + FABRIC_CLIENT_SECRET) - default_browser / browser 当不存在缓存会话时,触发ODBC交互式浏览器身份验证。
SQL_TRUST_SERVER_CERTIFICATE=true符合您的可信证书要求。- 结构身份验证模式优先级是确定的:
- client_secret 模式时 FABRIC_TENANT_ID, FABRIC_CLIENT_ID,以及 FABRIC_CLIENT_SECRET 都已配置。 - 回退模式(FABRIC_AUTH_FALLBACK_MODE,默认值 default_browser)当缺少任何客户端机密值时。
- 结构工作区路由可以使用将规范工作区名称映射到ID
fabric_workspace_id_map/FABRIC_WORKSPACE_ID_MAP. - 结构端点执行需要
fabric_sql_endpoint_map/FABRIC_SQL_ENDPOINT_MAP您路由到的每个已分配工作区+端点对的条目。
______________________________________________________________________
运行MCP服务器
python -m fde_sql_mcp.serverMCP配置代码片段示例
{
"name": "fde-sql-mcp",
"command": ["python", "-m", "fde_sql_mcp.server"],
"env": {
"SQL_SERVER_HOST": "your.sql.server.address",
"SQL_SERVER_DATABASE": "master",
"SQL_ENCRYPT": "true",
"SQL_TRUST_SERVER_CERTIFICATE": "true"
}
}______________________________________________________________________
操作员指南(Prem+面料)
1) 双环境设置检查表
- 配置本地SQL默认值(
sql_server,sql_database)infde_sql_mcp.config.json. - 配置结构配置列表:
- fabric_allowed_workspaces:仅 fde_core_data_dev, fde_core_data_stg, fde_core_data_prod - fabric_allowed_databases:仅 core_dw, core_lh
- 为您计划查询的每个工作区/端点对配置端点映射条目:
- fabric_sql_endpoint_map..warehouse -> server + database=core_dw - fabric_sql_endpoint_map..lakehouse -> server + database=core_lh
- 保持安全限制配置:
- sql_enforce_readonly=true - sql_max_rows - sql_max_query_chars - sql_query_timeout
2) 结构认证优先级
身份验证模式是确定的:
client_secret设置所有三个变量时的模式:
- FABRIC_TENANT_ID - FABRIC_CLIENT_ID - FABRIC_CLIENT_SECRET
- 否则,请使用回退模式
FABRIC_AUTH_FALLBACK_MODE(默认值default_browser).
使用 get_auth_info() 以确认有效的身份验证模式和配置的输入存在。
3) 路线选择示例
显式目标切换:
set_query_target("onprem")
set_query_target("fabric workspace=fde_core_data_dev endpoint=warehouse database=core_dw")
set_query_target("fabric workspace=fde_core_data_prod endpoint=lakehouse database=core_lh")自然语言目标选择:
set_query_target("query fabric prod lakehouse")
set_query_target("run this in fabric dev warehouse")查询前验证活动路由:
get_query_target()4) 安全护栏行为
run_readonly_query 对于本地和织物目标,始终执行护栏:
- 非只读SQL被拒绝:
- 例子: DELETE FROM dbo.table_name - 预期错误:仅 SELECT/WITH 允许发表声明。
- 超大SQL被拒绝:
- 示例:查询文本长度超过 sql_max_query_chars - 预期错误:超过了只读查询的最大长度。
- 超过配置最大值的行请求被限制:
- 例子: max_rows=1000 随着 sql_max_rows=200 - 预期行为: row_limit 回报 200, truncated 反映了上限结果。
- 超时适用于查询和元数据操作的SQL执行游标:
- 来源: sql_query_timeout
如果查询失败,请检查错误是否为:
- 路由/治理 (工作区/数据库不匹配,缺少端点映射),或
- 安全确认 (只读/尺寸/行限护栏)。
______________________________________________________________________
工具
get_auth_info()
返回非秘密结构身份验证诊断:
- 解析身份验证模式(
client_secret,default_browser或配置回退), - 凭证来源,
- 解析租户上下文,
- 布尔值指示配置了哪些结构身份验证输入。
set_query_target(target: str)
使用显式或自然语言提示设置查询执行的活动路由目标。
示例:
onpremfabric workspace=fde_core_data_dev endpoint=warehouse database=core_dwquery fabric prod lakehouse
get_query_target()
返回当前活动的路由目标上下文:
- 环境,
- 工作空间/workspace_id(用于结构),
- 端点类型,
- 数据库。
list_fabric_workspaces()
仅列出已分配的结构工作区和已配置的工作区ID映射。 超出所有列表的工作区将永远不会返回。
list_databases()
列出当前路由的SQL目标可见的数据库。
笔记:
- 本地目标:使用配置的SQL Server(
Trusted_Connection). - 结构目标:使用路由结构端点/数据库映射
fabric_sql_endpoint_map.
示例响应:
[
{
"name": "master",
"database_id": 1,
"state_desc": "ONLINE",
"recovery_model_desc": "SIMPLE"
}
]list_tables(database: str)
使用模式名称、创建/修改时间戳和时间类型元数据枚举所提供数据库中的表。
笔记:
- 对于织物目标,
database必须与路由目标数据库匹配(core_dw对于仓库,core_lh湖屋)。
list_views(database: str)
枚举所提供数据库中的视图以及模式和时间戳。
list_stored_procedures(database: str)
枚举所提供数据库中的存储过程,包括创建元数据以及对象是否随SQL Server一起提供。
list_indexes(database: str)
枚举所提供数据库中表范围内的非空索引定义(包括唯一性、主键标志、禁用状态和填充因子)。
run_readonly_query(database: str, query: str, max_rows: int | None)
执行具有服务器端行上限的已验证只读查询(仅限SELECT/CTE)。响应包括行、row_count、row_limit、截断标志和 target_context.
笔记:
- 活动结构目标针对配置的结构SQL端点映射执行,同时保留相同的只读合约。
target_context始终反映用于执行的已解析活动路由目标。
list_table_columns(database: str, schema: str, table: str)
返回表的列级元数据(类型、可空性、标识/计算标志、默认值、排序规则)。
list_view_columns(database: str, schema: str, view: str)
返回视图的列级元数据。
list_table_constraints(database: str, schema: str, table: str)
返回指定表的主键和唯一约束。
list_foreign_keys(database: str, schema: str, table: str)
返回指定表的外键关系(包括操作和引用的列)。
list_index_details(database: str, schema: str, table: str)
返回指定表的详细索引定义,包括键/包含列和筛选器。
list_view_definition(database: str, schema: str, view: str)
返回指定视图的SQL定义。
list_stored_procedure_definition(database: str, schema: str, procedure: str)
返回指定存储过程的SQL定义。
list_stored_procedure_parameters(database: str, schema: str, procedure: str)
返回指定存储过程的参数元数据。
list_object_dependencies(database: str, schema: str, object_name: str)
基于以下内容返回视图或存储过程的引用对象 sys.sql_expression_dependencies.
______________________________________________________________________
示例提示
List the databases on the SQL Server instance.
Show me the available SQL Server databases for my Windows login.