Ms Fabric MCP

Microsoft结构MCP服务器

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

🚀 特性

核心结构操作

• ✅ 工作区、湖畔小屋、仓库和桌子管理
• ✅ 增量表模式和元数据检索
• ✅ SQL查询执行和数据加载
• ✅ 报表和语义模型操作

PySpark高级开发

• 📓 智能笔记本创建 有6个专用模板
• 🔧 智能代码生成 用于常见的PySpark操作
• ✅ 全面验证 通过语法和最佳实践检查
• 🎯 特定于结构的优化 以及兼容性检查
• 📊 性能分析 提供评分和优化建议
• 🚀 实时监控 以及执行见解

LLM集成

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

🏗️ 建筑

graph TB
    subgraph "Developer Environment"
        IDE[IDE/VSCode]
        DEV[Developer]
        PROJ[Project Files]
    end
    
    subgraph "AI Layer"
        LLM[Large Language Model
Claude/GPT/etc.]
        CONTEXT[Conversation Context]
        REASONING[AI Reasoning Engine]
    end
    
    subgraph "MCP Layer"
        MCP[MCP Server]
        TOOLS[PySpark Tools]
        HELPERS[PySpark Helpers]
        TEMPLATES[Template Manager]
        VALIDATORS[Code Validators]
        GENERATORS[Code Generators]
    end
    
    subgraph "Microsoft Fabric"
        API[Fabric API]
        WS[Workspace]
        LH[Lakehouse]
        NB[Notebooks]
        TABLES[Delta Tables]
        SPARK[Spark Clusters]
    end
    
    subgraph "Operations Flow"
        CREATE[Create Notebooks]
        VALIDATE[Validate Code]
        GENERATE[Generate Code]
        ANALYZE[Analyze Performance]
        DEPLOY[Deploy to Fabric]
    end
    
    %% Developer interactions
    DEV --> IDE
    IDE --> PROJ
    
    %% LLM interactions
    IDE  LLM
    LLM  CONTEXT
    LLM --> REASONING
    
    %% MCP interactions
    LLM  MCP
    MCP --> TOOLS
    TOOLS --> HELPERS
    TOOLS --> TEMPLATES
    TOOLS --> VALIDATORS
    TOOLS --> GENERATORS
    
    %% Fabric interactions
    MCP  API
    API --> WS
    WS --> LH
    WS --> NB
    LH --> TABLES
    NB --> SPARK
    
    %% Operation flows
    TOOLS --> CREATE
    TOOLS --> VALIDATE
    TOOLS --> GENERATE
    TOOLS --> ANALYZE
    CREATE --> DEPLOY
    
    %% Data flow arrows
    REASONING -.->|"Intelligent Decisions"| TOOLS
    CONTEXT -.->|"Project Awareness"| VALIDATORS
    
    %% Styling
    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工具与结构API交互
6. 结果通过LLM以智能格式返回
7. 开发人员收到上下文相关的智能响应

📋 需求

• Python 3.12+
• Azure凭据 用于身份验证
• 紫外线 （来自星光）： 安装说明
• Azure命令行界面: 安装说明
• 可选：Node.js 对于MCP检查员： 安装说明

🔧 安装

1. 克隆存储库：

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

1. 设置虚拟环境：

   uv sync

1. 安装依赖项：

   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://:8081/mcp/",
                "headers": {
                    "Accept": "application/json,text/event-stream",
                }
            }
        }
    }
}

🛠️ 完整的工具参考

1.工作空间管理

list_workspaces

列出所有可用的结构工作区。

# Usage in LLM: "List all my Fabric workspaces"

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="Sales data lakehouse"
)

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="Sales data warehouse"
)

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

获取lakehouse或仓库的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"  # Options: basic, etl, analytics, ml
)

create_fabric_notebook

创建经过结构优化的笔记本电脑。

create_fabric_notebook(
    workspace="Analytics-Workspace",
    notebook_name="Fabric-Pipeline",
    template_type="fabric_integration"  # Options: fabric_integration, streaming
)

10.PySpark代码生成

generate_pyspark_code

为常见操作生成代码。

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

# Available operations:
# - read_table, write_table, transform, join, aggregate
# - schema_inference, data_quality, performance_optimization

generate_fabric_code

生成特定于结构的代码。

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

# Available operations:
# - 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

验证结构兼容性。

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. 基本的：基本的PySpark操作和DataFrame使用
2. 提取、转换、加载：完整的ETL管道，包括数据清理和Delta Lake
3. 分析：具有聚合和窗口函数的高级分析
4. 毫升：使用MLlib和特征工程的机器学习管道

高级模板

1. 织物集成：Lakehouse连接和特定于Fabric的实用程序
2. 流媒体：使用结构化流媒体进行实时处理

🎯 最佳实践

织物优化

# ✅ Use managed tables
df = spark.table("lakehouse.my_table")

# ✅ Use Delta Lake format
df.write.format("delta").mode("overwrite").saveAsTable("my_table")

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

性能优化

# ✅ Cache frequently used DataFrames
df.cache()

# ✅ Use broadcast for small tables
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")

# ✅ Partition large datasets
df.write.partitionBy("year", "month").saveAsTable("partitioned_table")

代码质量

# ✅ Define explicit schemas
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True)
])

# ✅ Handle null values
df.filter(col("column").isNotNull())

🔄 LLM增强工作流示例

自然语言请求

Human: "Create a PySpark notebook that reads sales data, cleans it, and optimizes performance"

LLM Response:
1. Creates Fabric-optimized notebook with ETL template
2. Generates lakehouse reading code
3. Adds data cleaning transformations
4. Includes performance optimization patterns
5. Validates code for best practices

性能分析

Human: "My PySpark notebook is slow. Help me optimize it."

LLM Response:
1. Analyzes notebook performance (scoring 0-100)
2. Identifies anti-patterns and bottlenecks
3. Suggests specific optimizations
4. Generates optimized code alternatives
5. Provides before/after comparisons

🔍 故障排除

常见问题

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

获取帮助

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

📈 性能指标

分析工具提供：

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

🤝 贡献

这个项目欢迎捐款！请参阅我们的投稿指南以了解详细信息。

📄 许可证

该项目根据MIT许可证获得许可。有关详细信息，请参阅LICENSE文件。

🙏 致谢

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

______________________________________________________________________

准备好在智能PySpark的帮助下加速您的Microsoft Fabric开发！ 🚀