使用说明
ms结构mcp服务器
Microsoft Fabric的模型上下文协议(MCP)服务器。将结构操作(工作区、笔记本电脑、SQL、Livy、管道、作业)作为人工智能代理可以调用的MCP工具公开。
⚠️ 警告:此包装适用于 仅限开发环境 不应用于生产。它包括可以执行破坏性操作的工具(例如。,
delete_item,delete_lakehouse_file,delete_activity_from_pipeline)并通过Livy Spark会话执行任意代码。执行前始终查看AI生成的工具调用。
快速开始
使用此MCP服务器的最快方法是 uvx:
uvx ms-fabric-mcp-server
安装
# Using uv (recommended)
uv pip install ms-fabric-mcp-server
# Using pip
pip install ms-fabric-mcp-server
# With SQL support (requires pyodbc)
pip install ms-fabric-mcp-server[sql]
# With OpenTelemetry tracing
pip install ms-fabric-mcp-server[sql,telemetry]
认证
用途 DefaultAzureCredential 从 azure-identity -不需要显式的凭据配置。这会自动尝试多种身份验证方法:
- 环境凭据(
AZURE_CLIENT_ID,AZURE_TENANT_ID,AZURE_CLIENT_SECRET) - 托管身份(在Azure上运行时)
- Azure CLI凭据(
az login) - VS代码凭据
- Azure PowerShell凭据
不需要特定于结构的身份验证环境变量 -只有通过上述任何方法进行身份验证,它才有效。
用法
VS代码集成
添加到您的VS代码MCP设置(.vscode/mcp.json 或用户设置):
{
"servers": {
"MS Fabric MCP Server": {
"type": "stdio",
"command": "uvx",
"args": ["ms-fabric-mcp-server"]
}
}
}
Claude桌面集成
添加到您的 claude_desktop_config.json:
{
"mcpServers": {
"fabric": {
"command": "uvx",
"args": ["ms-fabric-mcp-server"]
}
}
}
食品法典整合
添加到您的Codex config.toml:
[mcp_servers.ms_fabric_mcp]
command = "uvx"
args = ["ms-fabric-mcp-server"]
独立运行
# Using uvx (no installation needed)
uvx ms-fabric-mcp-server
# Direct execution (if installed)
ms-fabric-mcp-server
# Via Python module
python -m ms_fabric_mcp_server
# With MCP Inspector (development)
npx @modelcontextprotocol/inspector uvx ms-fabric-mcp-server
日志记录和调试(可选)
MCP stdio服务器必须将协议流量保持在stdout上,因此重定向 标准错误 以捕获日志。
赋予代理对日志文件的读取权限是调试故障的有效方法。
您还可以设置 AZURE_LOG_LEVEL (Azure SDK)和 MCP_LOG_LEVEL (服务器)控制冗长。
VS代码(Bash):
{
"servers": {
"MS Fabric MCP Server": {
"type": "stdio",
"command": "bash",
"args": [
"-lc",
"LOG_DIR=\"$HOME/mcp_logs\"; LOG_FILE=\"$LOG_DIR/ms-fabric-mcp-$(date +%Y%m%d_%H%M%S).log\"; uvx ms-fabric-mcp-server 2> \"$LOG_FILE\""
],
"env": {
"AZURE_LOG_LEVEL": "info",
"MCP_LOG_LEVEL": "INFO"
}
}
}
}
VS代码(PowerShell):
{
"servers": {
"MS Fabric MCP Server": {
"type": "stdio",
"command": "powershell",
"args": [
"-NoProfile",
"-Command",
"$logDir=\"$env:USERPROFILE\\mcp_logs\"; New-Item -ItemType Directory -Force -Path $logDir | Out-Null; $ts=Get-Date -Format yyyyMMdd_HHmmss; $logFile=\"$logDir\\ms-fabric-mcp-$ts.log\"; uvx ms-fabric-mcp-server 2> $logFile"
],
"env": {
"AZURE_LOG_LEVEL": "info",
"MCP_LOG_LEVEL": "INFO"
}
}
}
}
程序化使用(库模式)
from fastmcp import FastMCP
from ms_fabric_mcp_server import register_fabric_tools
# Create your own server
mcp = FastMCP("my-custom-server")
# Register all Fabric tools
register_fabric_tools(mcp)
# Add your own customizations...
mcp.run()
配置
环境变量(所有可选变量都有合理的默认值):
| 变量 | 默认值 | 描述 |
|---|---|---|
FABRIC_BASE_URL |
https://api.fabric.microsoft.com/v1 |
结构API基本URL |
FABRIC_SCOPES |
https://api.fabric.microsoft.com/.default |
OAuth作用域 |
FABRIC_API_CALL_TIMEOUT |
30 |
API超时(秒) |
FABRIC_MAX_RETRIES |
3 |
最大重试次数 |
FABRIC_RETRY_BACKOFF |
2.0 |
退避因子 |
LIVY_API_CALL_TIMEOUT |
120 |
Livy超时(秒) |
LIVY_POLL_INTERVAL |
2.0 |
Livy轮询间隔 |
LIVY_STATEMENT_WAIT_TIMEOUT |
10 |
Livy语句等待超时 |
LIVY_SESSION_WAIT_TIMEOUT |
240 |
Livy会话等待超时 |
MCP_SERVER_NAME |
ms-fabric-mcp-server |
MCP的服务器名称 |
MCP_LOG_LEVEL |
INFO |
日志记录级别 |
AZURE_LOG_LEVEL |
info |
Azure SDK日志记录级别 |
复制 .env.example 到 .env 并根据需要进行定制。
可用工具
服务器提供 57个核心工具,与 3个附加SQL工具 安装时 [sql] 额外(共60个)。
| 工具组 | 计数 | 工具 |
|---|---|---|
| 工作区 | 1 | list_workspaces |
| 物品 | 9 | list_items, get_item, list_folders, create_folder, move_folder, delete_folder, delete_item, rename_item, move_item_to_folder |
| 湖仓 | 4 | create_lakehouse, list_lakehouse_files, upload_lakehouse_file, delete_lakehouse_file |
| 笔记本 | 6 | create_notebook, get_notebook_definition, update_notebook_definition, get_notebook_run_details, list_notebook_runs, get_notebook_driver_logs |
| 工作 | 4 | run_on_demand_job, get_job_status, get_job_status_by_url, get_operation_result |
| 李维 | 8 | livy_create_session, livy_list_sessions, livy_get_session_status, livy_close_session, livy_run_statement, livy_get_statement_status, livy_cancel_statement, livy_get_session_log |
| 管道 | 11 | create_pipeline, add_copy_activity_to_pipeline, add_notebook_activity_to_pipeline, add_dataflow_activity_to_pipeline, add_activity_to_pipeline, delete_activity_from_pipeline, remove_activity_dependency, add_activity_dependency, get_pipeline_definition, update_pipeline_definition, get_pipeline_activity_runs |
| 数据流 | 3 | create_dataflow, get_dataflow_definition, run_dataflow |
| 语义模型 | 9 | create_semantic_model, add_table_to_semantic_model, add_relationship_to_semantic_model, get_semantic_model_details, get_semantic_model_definition, add_measures_to_semantic_model, delete_measures_from_semantic_model, delete_table_from_semantic_model, delete_relationship_from_semantic_model |
| Power BI | 2 | refresh_semantic_model, execute_dax_query |
| 结构化查询语言 (可选) | 3 | get_sql_endpoint, execute_sql_query, execute_sql_statement |
SQL工具(可选)
SQL工具需要 pyodbc 以及用于SQL Server的Microsoft ODBC驱动程序(驱动程序18或17——服务会自动检测已安装的驱动程序,并首选驱动程序18;设置 FABRIC_ODBC_DRIVER 覆盖):
# Install with SQL support
pip install ms-fabric-mcp-server[sql]
# On Ubuntu/Debian, install the ODBC driver first:
curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
sudo apt-get update
sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18 # or msodbcsql17
如果 pyodbc 如果不可用,服务器将启动57个工具(SQL工具已禁用)。
发展
# Clone and install with dev dependencies
git clone https://github.com/your-org/ms-fabric-mcp-server.git
cd ms-fabric-mcp-server
pip install -e ".[dev,sql,telemetry]"
# Run tests
pytest
# Run with coverage
pytest --cov
# Format code
black src tests
isort src tests
# Type checking
mypy src
集成测试
集成测试针对实时结构资源运行,并且是可选的。它们需要预先配置的结构工作区、Lakehouse、Warehouse,以及(对于复制活动测试)至少一个外部源连接。看 设置先决条件 在你第一次跑步之前。
要在本地开始,请复制示例env文件:
cp .env.integration.example .env.integration
然后填写与您配置的资源匹配的值。全套变量(带按目的分组的内联注释)存在于 .env.integration.example;此列表用于定向:
所有集成测试都需要:
FABRIC_INTEGRATION_TESTS=1FABRIC_TEST_WORKSPACE_NAME--测试工作区的显示名称FABRIC_TEST_LAKEHOUSE_NAME--工作区中的Lakehouse项目(用作复制活动的目标)FABRIC_TEST_LAKEHOUSE_SQL_DATABASE--Lakehouse的SQL端点数据库名称(通常与Lakehouse名称相同)FABRIC_TEST_WAREHOUSE_NAME--工作区中的仓库项(供SQLDML测试使用)
复制活动测试(管道流)所需:
FABRIC_TEST_DEST_CONNECTION_ID--目标Lakehouse的结构连接ID(用于所有复制活动测试)
每个发动机源输入 --设置一个块的所有5个变量,以启用该块的复制活动测试;如果缺少任何值,测试将跳过并记录原因。
PostgreSQL源代码(例如,通过网关可访问的VM托管的Postgres):
FABRIC_TEST_POSTGRES_CONNECTION_IDFABRIC_TEST_POSTGRES_SOURCE_TYPE(默认值PostgreSqlSource)FABRIC_TEST_POSTGRES_SCHEMAFABRIC_TEST_POSTGRES_TABLEFABRIC_TEST_POSTGRES_DEST_TABLE_NAME
SQL Server源(例如,VM托管的SQL Server、Azure SQL DB):
FABRIC_TEST_SQLSERVER_CONNECTION_IDFABRIC_TEST_SQLSERVER_SOURCE_TYPE(默认值SqlServerSource)FABRIC_TEST_SQLSERVER_SCHEMAFABRIC_TEST_SQLSERVER_TABLEFABRIC_TEST_SQLSERVER_DEST_TABLE_NAME
可选输入 (其他流在缺失时会干净地跳过):
- 语义模型:
FABRIC_TEST_SEMANTIC_MODEL_TABLE,FABRIC_TEST_SEMANTIC_MODEL_COLUMNS,FABRIC_TEST_SEMANTIC_MODEL_TABLE_2,FABRIC_TEST_SEMANTIC_MODEL_COLUMNS_2,FABRIC_TEST_SEMANTIC_MODEL_SCHEMA - 数据流:
FABRIC_TEST_DATAFLOW_NAME - Azure SPN身份验证(不使用时
az login):AZURE_TENANT_ID,AZURE_CLIENT_ID,AZURE_CLIENT_SECRET - Power BI调整:
POWERBI_BASE_URL,POWERBI_SCOPES,POWERBI_API_CALL_TIMEOUT,POWERBI_REFRESH_POLL_INTERVAL,POWERBI_REFRESH_WAIT_TIMEOUT
运行集成测试:
FABRIC_INTEGRATION_TESTS=1 pytest -m integration
或过滤到特定测试:
FABRIC_INTEGRATION_TESTS=1 pytest -m integration -k "copy_activity"
笔记:
- SQL测试需要
pyodbc以及SQL Server ODBC驱动程序(微软msodbcsql18推荐)。CI工作流通过以下方式安装它apt-get install msodbcsql18. - 当缺少可选依赖项或环境变量时,测试可能会跳过——这是故意的,而不是失败的。
- 这些测试使用实时结构资源,可能会产生容量使用和存储成本。在非生产工作区运行。
设置集成测试的先决条件
测试假设以下Fabric/Azure基础架构已经就绪。此设置是针对每个环境的一次性设置,不是测试运行本身的一部分:
-
专用的Fabric工作区 积极的能力。做 不 使用生产工作区——测试创建、修改和删除项目。
-
一个湖屋和一个仓库 在该工作区中,显示名称匹配
FABRIC_TEST_LAKEHOUSE_NAME和FABRIC_TEST_WAREHOUSE_NAME。将输入Lakehouse的SQL端点数据库名称(通常与Lakehouse名称匹配)FABRIC_TEST_LAKEHOUSE_SQL_DATABASE. -
可从Fabric访问的源数据库 (用于复制活动测试)。选项:
- Azure托管服务 (Azure SQL DB、Azure PostgreSQL数据库):直接创建;结构通过公共Azure骨干网到达它们。
- VM托管或本地数据库:需要 本地数据网关 安装在可以访问源数据库(在源网络上)和结构(通过公共互联网)的Windows主机上。
-
织物连接 到这些源数据库(每个要测试的源引擎一个)。通过Fabric门户创建→ 设置→ 管理连接和网关 → 新连接。使用 结构REST API 要读回的GUID
FABRIC_TEST_*_CONNECTION_ID:TOKEN=$(az account get-access-token --resource https://api.fabric.microsoft.com --query accessToken -o tsv) curl -sS -H "Authorization: Bearer $TOKEN" https://api.fabric.microsoft.com/v1/connections \ | python3 -c "import json,sys for c in json.load(sys.stdin).get('value',[]): print(f\"{c['displayName']:40s} {c['id']}\")" -
湖屋目的地Fabric连接 (云>湖屋)。用于
FABRIC_TEST_DEST_CONNECTION_ID.管道复制活动在写入目标Lakehouse时针对此连接。 -
认证:
- 本地开发:
az login因为结构工作区成员就足够了(服务器使用DefaultAzureCredential). - CI/无人值守:创建Azure服务主体,将其添加到测试工作区 成员 (织物门户→ 工作区→ 管理访问权限),授予它 “可以使用” 在每个结构连接上(设置→ 管理连接→ 选择连接→ 分享→ 添加SP),并通过提供其凭据
AZURE_TENANT_ID/AZURE_CLIENT_ID/AZURE_CLIENT_SECRET两个ACL授权(工作区成员+每个连接“可以使用”)都是必需的——连接访问权限不从工作区成员身份继承。
- 本地开发:
-
GitHub操作 (如果运行捆绑的工作流):创建一个名为
Integration在fork的存储库设置中添加每个FABRIC_TEST_*和AZURE_*上面的变量作为同名的环境机密。.github/workflows/integration-tests.yml列出规范秘密名称集。
许可证
麻省理工学院
