fde fabric MCP

FDE织物MCP

一个MCP服务器，它将Microsoft Fabric工作区、Lakehouse、仓库、SQL端点和特定于FDE的项目编排作为MCP工具（例如Claude Desktop）公开。

它处理：

• Azure AD身份验证 azure-identity （DefaultAzureCredential或客户端机密）
• 结构工作区/项目发现
• Lakehouse/Warehouse SQL端点解析
• 对Fabric Lakehouse/Warehouse执行SQL
• 通过自定义“项目编排器”模式进行项目驱动的管道提交

______________________________________________________________________

______________________________________________________________________

特性

• 健康和身份助手(ping, whoami, reset_context)
• 工作空间发现/选择，以及湖屋/仓库解决方案
• 对活动端点执行SQL（仓库优先，湖屋回退）
• 用于工作区、湖畔小屋、仓库和项目的每个客户端内存上下文存储
• 项目工具(list_projects, set_project, get_current_project)
• 基于编排器的自定义管道提交

- run_project_pipeline （运行单个项目的路由处理器管道） - preview_project_pipeline_payload （构建有效载荷而不提交） - run_all_jobs_for_day （编排器按不同的UTC时间间隔运行）

______________________________________________________________________

安装

# 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_fabric_mcp/
  src/fde_fabric_mcp/
    clients/
      fabric.py            # REST client for Fabric API
      pipeline.py          # Pipeline submit/monitor client
      sql.py               # Fabric SQL connector via Azure AD token
    core/
      context_store.py     # Per-MCP-client in-memory context
      sql_endpoints.py     # Resolve Lakehouse/Warehouse SQL endpoints
    tools/
      identity.py          # Azure CLI identity helpers (`whoami`)
      lakehouse.py         # List/set/get current lakehouse
      notebooks.py         # Notebook list/create helpers
      pipelines.py         # Project pipeline + orchestrator helpers
      projects.py          # Project lookup/store helpers
      warehouse.py         # Set/get warehouse & run SQL queries
      workspace.py         # List/set/get current workspace
      notebook_templates/  # Built-in notebook templates (see list_notebook_templates)
    auth.py                # Azure credential + bearer token helpers
    config.py              # Settings from environment variables
    server.py              # MCP server definition & stdio entrypoint
    __init__.py            # Package metadata (__version__)

______________________________________________________________________

需求

• Python 3.10+（参见 pyproject.toml)

• 访问Microsoft Fabric/Power BI租户

• Azure AD应用程序或交互式登录 DefaultAzureCredential

• SQL Server的ODBC驱动程序，例如：

- SQL Server的ODBC驱动程序18 （首选） - SQL Server的ODBC驱动程序17

核心运行时依赖关系（非详尽）：

• azure-identity, azure-core
• mcp
• requests, pyodbc, pandas, croniter

______________________________________________________________________

配置

配置由环境变量通过以下方式驱动 config.Settings 和 auth.AuthConfig.

Auth/结构API

FABRIC_AUTH_MODE              # "auto" (default) or "client_secret"
FABRIC_TENANT_ID              # Required when FABRIC_AUTH_MODE="client_secret"
FABRIC_CLIENT_ID              # Required when FABRIC_AUTH_MODE="client_secret"
FABRIC_CLIENT_SECRET          # Required when FABRIC_AUTH_MODE="client_secret"
FABRIC_SCOPE                  # Fabric API scope for AAD tokens (default https://api.fabric.microsoft.com/.default)
FABRIC_BASE_URL               # Fabric REST base URL (default https://api.fabric.microsoft.com/v1)
FABRIC_DEFAULT_WORKSPACE_ID   # Optional default workspace GUID
FABRIC_SQL_DRIVER             # Optional ODBC driver override, default "{ODBC Driver 17 for SQL Server}"

身份验证模式：

• FABRIC_AUTH_MODE=auto

用途 DefaultAzureCredential，它可以链接：

- az login - 管理身份 - 环境凭据 - VSCode/AAzure工作负载标识

• FABRIC_AUTH_MODE=client_secret

用途 ClientSecretCredential 随着 FABRIC_TENANT_ID, FABRIC_CLIENT_ID, FABRIC_CLIENT_SECRET.

管道助手配置

项目/编排器帮助程序使用从以下位置提取的共享环境配置行：

• settings.config_workspace （默认值 "fde_core_data_config")
• settings.config_warehouse / settings.config_database （默认值 "core_dw_config")
• settings.config_table （默认值 "dbo.environment_config")

这些当前设置在 src/fde_fabric_mcp/config.py。如果您的配置仓库/表位于其他地方，请在此处进行调整。

______________________________________________________________________

SQL护栏（可选）

MCP服务器可以对特定工作区强制执行只读SQL，以防止 意外写入操作 run_sql_query.

FABRIC_READ_ONLY_SQL_ENABLED     # default true
FABRIC_READ_ONLY_WORKSPACE_NAMES # comma-separated workspace names
FABRIC_READ_ONLY_WORKSPACE_IDS   # comma-separated workspace GUIDs

默认值定义见 src/fde_fabric_mcp/config.py.

______________________________________________________________________

运行MCP服务器

您可以使用以下命令通过stdio启动服务器：

python -m fde_fabric_mcp.server

server.py 暴露a run() 初始化函数 FastMCP 实例和调用 mcp.run() （stdio传输）。

MCP配置代码片段示例

对于使用简单的基于命令的传输（伪配置）的MCP客户端：

{
  "name": "fde-fabric-mcp",
  "command": ["python", "-m", "fde_fabric_mcp.server"],
  "env": {
    "FABRIC_AUTH_MODE": "auto",
    "FABRIC_SCOPE": "https://api.fabric.microsoft.com/.default",
    "FABRIC_BASE_URL": "https://api.fabric.microsoft.com/v1",
    "FABRIC_DEFAULT_WORKSPACE_ID": ""
  }
}

______________________________________________________________________

暴露的MCP工具

效用

• ping()

健康检查。退货 "pong" 如果MCP服务器正在响应。

• whoami()

用途 az account show 返回：

  {
    "username": "",
    "email": "user@domain.com",
    "error": null
  }

或a error string如果Azure CLI不可用/未配置。

• reset_context(ctx)

清除当前MCP客户端的所有命名空间值（工作区、湖畔小屋、仓库、项目）。

______________________________________________________________________

工作区工具

• list_workspaces(ctx)

列出当前标识可见的所有结构工作区。

• set_workspace(ctx, workspace_id=None, workspace_name=None)

在上下文存储中解析并持久化活动工作区。

- 更喜欢 workspace_id 对于确定性行为。 - 回到不区分大小写的匹配 workspace_name.

• get_current_workspace(ctx)

返回活动工作区。 如果没有明确设置，但是 FABRIC_DEFAULT_WORKSPACE_ID 如果配置了，它将使用它。

______________________________________________________________________

湖屋工具

• list_lakehouses(ctx)

列出当前工作空间中的所有Lakehouse项目。

• set_lakehouse(ctx, lakehouse: str)

在活动工作区中按名称或ID解析湖屋并存储：

- server - database - workspace_id, workspace_name - item_id, item_name - type: "Lakehouse"

• get_current_lakehouse(ctx)

返回当前客户端存储的lakehouse信息。 如果未设置任何内容，则会出现错误。

______________________________________________________________________

仓库/SQL工具

• set_warehouse_from_fabric(ctx, workspace: str, warehouse: str)

使用Fabric REST按工作区+仓库（名称或GUID）解析仓库，然后存储：

- server - database - workspace_id, workspace_name - item_id, item_name - type: "Warehouse"

• get_current_warehouse(ctx)

返回当前客户端的存储仓库信息。 如果未设置任何内容，则会出现错误。

• run_sql_query(ctx, sql: str)

对“活动”SQL端点执行SQL查询，解析方式如下：

1. 仓库（首选） 1. 湖屋（后备）

返回JSON格式：

  {
    "rows": [ { "col": "value", ... } ],
    "row_count": 42,
    "server": "",
    "database": "",
    "target_kind": "warehouse" | "lakehouse",
    "target_name": "",
    "workspace_name": ""
  }

______________________________________________________________________

笔记本工具

• list_notebooks(ctx, workspace=None)

列出工作区（或活动工作区）中的笔记本项目。

• get_notebook(ctx, workspace, notebook_id)

按ID返回笔记本项目。

• create_notebook(ctx, workspace, notebook_name, content, folder_path=None, folder_id=None, create_missing_folders=False, ipynb_name=None)

从base64编码创建新笔记本 .ipynb 有效载荷。

- 使用 folder_path （例如。， "Shared/Team Notebooks/DataOps")以文件夹为目标。 - 使用 folder_id 以特定文件夹项ID为目标。 - 集 create_missing_folders=True 沿路径创建任何缺失的文件夹。

• list_notebook_templates(ctx)

列出内置模板名称和说明。 内置模板包含在以下仓库中 tools/notebook_templates.

• create_template_notebook(ctx, workspace, notebook_name, template_name, folder_path=None, folder_id=None, create_missing_folders=False, ipynb_name=None)

从内置模板创建新笔记本，并将其放置在可选文件夹中 文件夹。

可用模板：

• basic
• python_file_to_delta_template
• python_delta_or_sql_to_delta_template

______________________________________________________________________

项目工具

• list_projects(ctx, search=None, top=50)

列出来自的项目 dbo.project 在活动SQL端点中（首选仓库，湖屋回退）。

• set_project(ctx, project_id=None, project_alias=None)

按ID或别名加载项目，并将其存储在每个客户端上下文中。

• get_current_project(ctx)

返回当前客户端的存储项目信息。

______________________________________________________________________

管道工具（自定义项目编排器）

此仓库中的管道助手是 非通用的“运行任何结构管道”帮助程序他们实施了一项 自定义项目编排器模式 FDE使用的：

• 每个项目存储一个 cron_expression （评估于 America/New_York 除非 TZ=/CRON_TZ= 包括在内）。
• 项目按不同的UTC分组 run_interval_utc 每天的值。
• A. 单编排器管道 每间隔调用一次，以分派适当的项目工作负载。

必需的env_config密钥

编排器管道ID从共享环境配置行中提取：

• env_config["project_orchestrator_id"] （管道Id）
• env_config["workspace_id"] （workspaceId）

这些与以下内容一起传递到编排器调用有效负载中：

• interval （UTC分钟字符串）
• is_current （编排器逻辑标志）
• env （完整的env_config块）

工具

• run_project_pipeline(...)

提交当前选定项目的管道运行。 如果 time_utc="cron" （默认），间隔时间从项目的 cron_expression 在ET中转换为UTC。

• preview_project_pipeline_payload(...)

为当前项目构建有效负载，并在不提交的情况下返回它们。

• run_all_jobs_for_day(run_date, ...)

计算机不同 run_interval_utc 值为 run_date，然后调用 项目协调人 每间隔一次。 第一个调用返回a confirmation_required 有效载荷，并建议下一步。

注意：因为这些工具取决于您的 自定义编排器管道合同，他们假设： - 一 dbo.project 桌子与 cron_expression - 包含以下内容的环境配置行 project_orchestrator_id

______________________________________________________________________

示例提示

将这些与您的MCP客户端一起使用来驱动工具：

List all Fabric workspaces I can access.

Set the active workspace to "fde_core_data_dev" and list its lakehouses.

Set the lakehouse to "SalesLakehouse" and show the current SQL endpoint.

Use the warehouse "CoreDW" in workspace "fde_core_data_prod" and run:
SELECT TOP 10 * FROM dbo.project ORDER BY updated_at DESC;

List projects where the alias contains "sftp", then set the first one.

Preview the pipeline payload for project "daily_sales" for 2025-01-15 (use cron time).

Run the project pipeline for "daily_sales" on 2025-01-15 at 05:30 UTC (dry run).

Run the orchestrator for 2025-01-15 and show me the confirmation details.

Reset my session context so I can start over.

______________________________________________________________________

Auth和SQL片段如何组合在一起

• auth.py 构建Azure TokenCredential 使用以下任一方法：

- DefaultAzureCredential (FABRIC_AUTH_MODE=auto)，或 - ClientSecretCredential (FABRIC_AUTH_MODE=client_secret).

• clients.fabric.FabricClient 使用该凭据为Fabric REST调用获取承载令牌。

• core.sql_endpoints 使用Fabric REST来解析Lakehouse/Warehouse SQL端点。

• clients.sql.SQLServerConnection 使用SQL作用域的访问令牌并将其传递给 pyodbc 通过 SQL_COPT_SS_ACCESS_TOKEN (1256).

• tools.warehouse.run_sql_query_impl 将所有这些绑定在一起，并将JSON行返回给MCP客户端。

______________________________________________________________________

开发说明

• 上下文仅在内存中(core.context_store)并键入 ctx.client_id.

这意味着：

- 每个MCP客户都有自己的工作空间/湖畔小屋/仓库/项目选择。 - MCP服务器进程重新启动时，上下文丢失。

• HTTP调用当前是同步的(requests)但包裹着 asyncio.to_thread 因此，它们可以从异步工具中安全使用。