星战科技xingzhan.cloud
返回市场
ms-织构-mcp

ms-织构-mcp

作者：datumnova13 星标更新：2025-05-27
安装使用来源：spider==github.com

项目介绍

Microsoft Fabric MCP Server

一个基于Python的MCP（模型上下文协议）服务器，用于与Microsoft Fabric API交互，具有高级PySpark笔记本开发、测试和优化功能，并集成了大型语言模型（LLM）。

🚀 功能

核心Fabric操作

  • ✅ 工作区、湖库、仓库和表管理
  • ✅ Delta表模式和元数据检索
  • ✅ SQL查询执行和数据加载
  • ✅ 报告和语义模型操作

高级PySpark开发

  • 📓 智能笔记本创建，提供6个专用模板
  • 🔧 智能代码生成，针对常见的PySpark操作
  • ✅ 全面验证，包括语法和最佳实践检查
  • 🎯 特定于Fabric的优化和兼容性检查
  • 📊 性能分析，提供评分和优化建议
  • 🚀 实时监控和执行洞察

LLM集成

  • 🤖 自然语言接口用于PySpark开发
  • 🧠 上下文感知辅助，带有对话记忆
  • 🎨 智能代码格式化和解释
  • 📈 智能优化建议，基于项目模式

🏗️ 架构

graph TB
    subgraph "开发者环境"
        IDE[IDE/VSCode]
        DEV[开发者]
        PROJ[项目文件]
    end
    
    subgraph "AI层"
        LLM[大型语言模型<br/>Claude/GPT等]
        CONTEXT[对话上下文]
        REASONING[AI推理引擎]
    end
    
    subgraph "MCP层"
        MCP[MCP服务器]
        TOOLS[PySpark工具]
        HELPERS[PySpark助手]
        TEMPLATES[模板管理器]
        VALIDATORS[代码验证器]
        GENERATORS[代码生成器]
    end
    
    subgraph "Microsoft Fabric"
        API[Fabric API]
        WS[工作区]
        LH[湖库]
        NB[笔记本]
        TABLES[Delta表]
        SPARK[Spark集群]
    end
    
    subgraph "操作流程"
        CREATE[创建笔记本]
        VALIDATE[验证代码]
        GENERATE[生成代码]
        ANALYZE[分析性能]
        DEPLOY[部署到Fabric]
    end
    
    %% 开发者交互
    DEV --> IDE
    IDE --> PROJ
    
    %% LLM交互
    IDE <--> LLM
    LLM <--> CONTEXT
    LLM --> REASONING
    
    %% MCP交互
    LLM <--> MCP
    MCP --> TOOLS
    TOOLS --> HELPERS
    TOOLS --> TEMPLATES
    TOOLS --> VALIDATORS
    TOOLS --> GENERATORS
    
    %% Fabric交互
    MCP <--> API
    API --> WS
    WS --> LH
    WS --> NB
    LH --> TABLES
    NB --> SPARK
    
    %% 操作流程
    TOOLS --> CREATE
    TOOLS --> VALIDATE
    TOOLS --> GENERATE
    TOOLS --> ANALYZE
    CREATE --> DEPLOY
    
    %% 数据流箭头
    REASONING -.->|"智能决策"| TOOLS
    CONTEXT -.->|"项目意识"| VALIDATORS
    
    %% 样式
    classDef devEnv fill:#e1f5fe
    classDef aiLayer fill:#fff9c4
    classDef mcpLayer fill:#f3e5f5
    classDef fabricLayer fill:#e8f5e8
    classDef operations fill:#fff3e0
    
    class IDE,DEV,PROJ devEnv
    class LLM,CONTEXT,REASONING aiLayer
    class MCP,TOOLS,HELPERS,TEMPLATES,VALIDATORS,GENERATORS mcpLayer
    class API,WS,LH,NB,TABLES,SPARK fabricLayer
    class CREATE,VALIDATE,GENERATE,ANALYZE,DEPLOY operations

交互流程

  1. 开发者在IDE中请求帮助
  2. IDE与LLM（Claude/GPT）通信
  3. LLM通过上下文和推理进行分析
  4. LLM智能调用MCP服务器工具
  5. MCP工具与Fabric API交互
  6. 结果通过LLM以智能格式返回
  7. 开发者收到上下文相关的智能响应

📋 要求

🔧 安装

  1. 克隆仓库：

    git clone https://github.com/your-repo/fabric-mcp.git
cd fabric-mcp

  2. 设置虚拟环境：

    uv sync

  3. 安装依赖项：

    pip install -r requirements.txt

🚀 使用

  1. 使用STDIO

连接到Microsoft Fabric

az login --scope https://api.fabric.microsoft.com/.default

使用MCP检查器运行

uv run --with mcp mcp dev fabric_mcp.py

这将在http://localhost:6274启动服务器并附带检查器。

VSCode集成

添加到你的launch.json：

{
    "mcp": {
        "servers": {
            "ms-fabric-mcp": {
                "type": "stdio",
                "command": "<项目文件夹完整路径>\\.venv\\Scripts\\python.exe",
                "args": ["<项目文件夹完整路径>\\fabric_mcp.py"]
            }
        }
    }
}
  1. 使用HTTP

启动MCP服务器

uv run python .\fabric_mcp.py --port 8081

VSCode集成

添加到你的launch.json：

{
    "mcp": {
        "servers": {
            "ms-fabric-mcp": {
                "type": "http",
                "url": "http://<localhost或远程IP>:8081/mcp/",
                "headers": {
                    "Accept": "application/json,text/event-stream",
                }
            }
        }
    }
}

🛠️ 完整工具参考

1. 工作区管理

list_workspaces

列出所有可用的Fabric工作区。

# 在LLM中的使用："列出我的所有Fabric工作区"

set_workspace

设置当前会话的工作区上下文。

set_workspace(workspace="Analytics-Workspace")

2. 湖库操作

list_lakehouses

列出工作区中的所有湖库。

list_lakehouses(workspace="Analytics-Workspace")

create_lakehouse

创建新的湖库。

create_lakehouse(
    name="Sales-Data-Lake",
    workspace="Analytics-Workspace",
    description="销售数据湖库"
)

set_lakehouse

设置当前湖库上下文。

set_lakehouse(lakehouse="Sales-Data-Lake")

3. 仓库操作

list_warehouses

列出工作区中的所有仓库。

list_warehouses(workspace="Analytics-Workspace")

create_warehouse

创建新的仓库。

create__warehouse(
    name="Sales-DW",
    workspace="Analytics-Workspace", 
    description="销售数据仓库"
)

set_warehouse

设置当前仓库上下文。

set_warehouse(warehouse="Sales-DW")

4. 表操作

list_tables

列出湖库中的所有表。

list_tables(workspace="Analytics-Workspace", lakehouse="Sales-Data-Lake")

get_lakehouse_table_schema

获取特定表的模式。

get_lakehouse_table_schema(
    workspace="Analytics-Workspace",
    lakehouse="Sales-Data-Lake",
    table_name="transactions"
)

get_all_lakehouse_schemas

获取湖库中所有表的模式。

get_all_lakehouse_schemas(
    workspace="Analytics-Workspace",
    lakehouse="Sales-Data-Lake"
)

set_table

设置当前表上下文。

set_table(table_name="transactions")

5. SQL操作

get_sql_endpoint

获取湖库或仓库的SQL端点。

get_sql_endpoint(
    workspace="Analytics-Workspace",
    lakehouse="Sales-Data-Lake",
    type="lakehouse"
)

run_query

执行SQL查询。

run_query(
    workspace="Analytics-Workspace",
    lakehouse="Sales-Data-Lake",
    query="SELECT COUNT(*) FROM transactions",
    type="lakehouse"
)

6. 数据加载

load_data_from_url

从URL加载数据到表中。

load_data_from_url(
    url="https://example.com/data.csv",
    destination_table="new_data",
    workspace="Analytics-Workspace",
    lakehouse="Sales-Data-Lake"
)

7. 报告&模型

list_reports

列出工作区中的所有报告。

list_reports(workspace="Analytics-Workspace")

get_report

获取特定报告的详细信息。

get_report(workspace="Analytics-Workspace", report_id="report-id")

list_semantic_models

列出工作区中的语义模型。

list_semantic_models(workspace="Analytics-Workspace")

get_semantic_model

获取特定语义模型。

get_semantic_model(workspace="Analytics-Workspace", model_id="model-id")

8. 基本笔记本操作

list_notebooks

列出工作区中的所有笔记本。

list_notebooks(workspace="Analytics-Workspace")

get_notebook_content

获取笔记本内容。

get_notebook_content(
    workspace="Analytics-Workspace",
    notebook_id="notebook-id"
)

update_notebook_cell

更新特定笔记本单元格。

update_notebook_cell(
    workspace="Analytics-Workspace",
    notebook_id="notebook-id",
    cell_index=0,
    cell_content="print('Hello, Fabric!')",
    cell_type="code"
)

9. 高级PySpark笔记本创建

create_pyspark_notebook

从基本模板创建笔记本。

create_pyspark_notebook(
    workspace="Analytics-Workspace",
    notebook_name="Data-Analysis",
    template_type="analytics"  # 选项：basic, etl, analytics, ml
)

create_fabric_notebook

创建优化过的Fabric笔记本。

create_fabric_notebook(
    workspace="Analytics-Workspace",
    notebook_name="Fabric-Pipeline",
    template_type="fabric_integration"  # 选项：fabric_integration, streaming
)

10. PySpark代码生成

generate_pyspark_code

为常见操作生成代码。

generate_pyspark_code(
    operation="read_table",
    source_table="sales.transactions",
    columns="id,amount,date"
)

# 可用操作：
# - read_table, write_table, transform, join, aggregate
# - schema_inference, data_quality, performance_optimization

generate_fabric_code

生成特定于Fabric的代码。

generate_fabric_code(
    operation="read_lakehouse",
    lakehouse_name="Sales-Data-Lake",
    table_name="transactions"
)

# 可用操作：
# - read_lakehouse, write_lakehouse, merge_delta, performance_monitor

11. 代码验证&分析

validate_pyspark_code

验证PySpark代码的语法和最佳实践。

validate_pyspark_code(code="""
df = spark.table('transactions')
df.show()
""")

validate_fabric_code

验证Fabric兼容性。

validate_fabric_code(code="""
df = spark.table('lakehouse.transactions')
df.write.format('delta').saveAsTable('summary')
""")

analyze_notebook_performance

全面性能分析。

analyze_notebook_performance(
    workspace="Analytics-Workspace",
    notebook_id="notebook-id"
)

12. 上下文管理

clear_context

清除当前会话上下文。

clear_context()

📊 PySpark模板

基础模板

  1. basic：基本的PySpark操作和DataFrame使用
  2. etl：完整的ETL管道，包括数据清洗和Delta Lake
  3. analytics：高级分析，包括聚合和窗口函数
  4. ml：机器学习管道，包括MLlib和特征工程

高级模板

  1. fabric_integration：湖库连接性和特定于Fabric的实用程序
  2. streaming：实时处理，使用Structured Streaming

🎯 最佳实践

Fabric优化

# ✅ 使用托管表
df = spark.table("lakehouse.my_table")

# ✅ 使用Delta Lake格式
df.write.format("delta").mode("overwrite").saveAsTable("my_table")

# ✅ 利用notebookutils
import notebookutils as nbu
workspace_id = nbu.runtime.context.workspaceId

性能优化

# ✅ 缓存频繁使用的DataFrame
df.cache()

# ✅ 对小表使用广播
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")

# ✅ 分区大数据集
df.write.partitionBy("year", "month").saveAsTable("partitioned_table")

代码质量

# ✅ 定义明确的模式
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True)
])

# ✅ 处理空值
df.filter(col("column").isNotNull())

🔄 示例LLM增强工作流程

自然语言请求

Human: "创建一个PySpark笔记本，读取销售数据，清理它，并优化性能"

LLM Response:
1. 创建优化过的Fabric笔记本，使用ETL模板
2. 生成读取湖库的代码
3. 添加数据清洗转换
4. 包括性能优化模式
5. 验证代码的最佳实践

性能分析

Human: "我的PySpark笔记本很慢。帮我优化它。"

LLM Response:
1. 分析笔记本性能（评分0-100）
2. 识别反模式和瓶颈
3. 提供具体的优化建议
4. 生成优化代码替代方案
5. 提供前后比较

🔍 故障排除

常见问题

  • 认证：确保az login使用正确的范围
  • 上下文：使用clear_context()重置会话状态
  • 工作区：验证工作区名称和权限
  • 模板：检查文档中的可用模板类型

获取帮助

  • 使用验证工具解决代码问题
  • 检查性能分析以寻找优化机会
  • 利用LLM自然语言界面获取指导

📈 性能指标

分析工具提供：

  • 操作计数每笔记本单元格
  • 性能问题检测和标记
  • 优化机会识别
  • 评分系统（0-100）评估代码质量
  • Fabric兼容性评估

🤝 贡献

欢迎贡献！请参阅我们的贡献指南了解详情。

📄 许可

该项目采用MIT许可。详见LICENSE文件。

🙏 致谢

灵感来源：https://github.com/Augustab/microsoft_fabric_mcp/tree/main

准备好使用智能PySpark辅助来加速您的Microsoft Fabric开发！ 🚀