Fabric 資料代理程式 Python SDK (預覽)

Fabric 資料代理程式 Python SDK 提供程式化存取 Fabric 資料代理產物。 它專為以程式碼為優先的使用者設計，讓他們能在不使用 Fabric 入口的情況下建立、設定、更新及發佈資料代理。 你可以在 Microsoft Fabric 筆記本中執行 SDK，或在驗證 Fabric 後從自己的環境執行。

管理平面與執行階段

SDK 是一個管理平面工具。 利用它來管理資料代理的生命週期：建立產物、新增與配置資料來源、設定指令與範例查詢，然後發佈。 它是透過 Fabric 公開 REST API 執行，而這也是你已經用於其他工作區和項目的相同介面，因此驗證和請求模式也都相同。

管理平面與執行時查詢資料代理的方式是分開的。 發布資料代理後，透過其模型情境協定（MCP）端點，透過工具、應用程式及其他代理經驗查詢資料代理。 欲了解更多資訊，請參閱 資料代理作為模型情境協定伺服器。

先決條件

• 具有支援資料代理容量的 Fabric 工作區。
• 工作空間中支援的資料來源，例如湖屋、倉庫、Power BI 語意模型或 KQL 資料庫。
• Python 3.10 或更新版本。
• 若要在 Fabric 筆記本外執行，您需要有可向 Fabric 驗證身分的方法，例如 Azure CLI 或服務主體。

安裝

SDK 以 fabric-data-agent-sdk 形式發佈在 PyPI 上。 使用 pip 安裝：

%pip install fabric-data-agent-sdk

驗證 Fabric 身分

當你在 Fabric 筆記本中執行 SDK 時，筆記本會幫你處理認證。

當你在 Fabric 外執行 SDK 時，先登入 Fabric。 以下範例是用 Azure CLI 憑證登入，並將其設為該會話的預設值。 你可以使用使用者帳號或是服務主體。

from azure.identity import AzureCliCredential
from fabric.analytics.environment.credentials import (
    SetFabricAnalyticsDefaultTokenCredentialsGlobally,
)

credential = AzureCliCredential()
SetFabricAnalyticsDefaultTokenCredentialsGlobally(credential)

建立資料代理程式

在工作區建立一個資料代理。 將工作區 ID 替換為資料代理所在工作區的 ID。

from fabric.dataagent.client import create_data_agent

workspace_id = "<your-workspace-id>"

agent = create_data_agent(
    data_agent_name="Quickstart data agent",
    workspace_id=workspace_id,
)

設定資料代理並新增資料來源

先設定代理指令，然後新增資料來源。 可以使用現有湖屋、倉庫、語意模型或 KQL 資料庫的工件 ID。

agent_instructions = "<your agent instructions>"
datasource_id = "<your-datasource-id>"

agent.update_settings(ai_instructions=agent_instructions)

agent.add_staging_datasource(
    artifact_name_or_id=datasource_id,
    workspace_id_or_name=workspace_id,
)

你也可以加入資料來源指令和範例查詢，以提升資料代理人回答的品質。 想了解更多，請參考 GitHub 上的 Fabric 資料代理 SDK 範例。

發佈資料代理

發布分階段設定，讓資料代理可供查詢。

agent.publish_staging(description="Initial publish")

查詢已發佈的資料代理

發佈資料代理後，透過其 MCP 端點查詢資料代理。 MCP 端點是資料代理的執行階段與取用介面。 你可以透過工具、應用程式及其他代理連接，查詢資料代理、提問並獲得答案。 關於設定步驟及支援的用戶端，請參閱 資料代理作為模型情境協定伺服器。

取得端點網址

你可以透過兩種方式取得 MCP 端點的 URL：

• 從資料代理設定複製。 發佈資料代理後，在代理設定中開啟 模型情境協定（Model Context Protocol ）標籤，複製 MCP 伺服器的網址。 詳情請參見 資料代理作為模型情境協定伺服器。

• 手動建置。 請根據你的工作區 ID 與資料代理（工件）ID 使用以下格式構建 URL：

  https://api.fabric.microsoft.com/v1/mcp/workspaces/{WorkspaceId}/dataagents/{DataAgentId}/agent
  

  將佔位符替換為你已發佈資料代理的值：

  Placeholder	Description
  {WorkspaceId}	包含資料代理的 Fabric 工作空間的 ID。
  {DataAgentId}	已發佈資料代理的識別碼。

  手動建立的網址只有在你發布資料代理後才有效。 如果資料代理沒有被發佈，端點即使 URL 正確也會回傳錯誤。

Authentication

對 MCP 端點的請求必須透過 Fabric 進行驗證。 確保你的客戶端包含有效的持有憑證，並授權存取目標工作空間和資料代理。 該令牌可以代表使用者身份或服務主體（SPN）。

可用工具

Fabric 資料代理會暴露一個 MCP 伺服器，提供單一工具。 客戶端呼叫此工具，將問題發送給資料代理，並接收產生的答案。

從 Python 查詢資料代理

以下範例連接 MCP 端點，發現工具，發送問題，並列印答案。 它會重複使用來自 credential 步驟的 ，並使用 MCP Python SDK。 先安裝 SDK：

%pip install mcp

import asyncio

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

workspace_id = "<your-workspace-id>"
data_agent_id = "<your-data-agent-id>"
question = "<your question>"

mcp_url = (
    f"https://api.fabric.microsoft.com/v1/mcp/workspaces/{workspace_id}"
    f"/dataagents/{data_agent_id}/agent"
)


def get_auth_headers():
    token = credential.get_token("https://api.fabric.microsoft.com/.default")
    return {"Authorization": f"Bearer {token.token}"}


async def query_data_agent(question):
    headers = get_auth_headers()

    async with streamablehttp_client(mcp_url, headers=headers) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # The data agent exposes a single tool. Discover it, then call it.
            tools = await session.list_tools()
            tool = tools.tools[0]
            question_arg = next(iter(tool.inputSchema["properties"]))

            result = await session.call_tool(tool.name, {question_arg: question})

            answers = [block.text for block in result.content if block.type == "text"]
            return "\n".join(answers)


answer = asyncio.run(query_data_agent(question))
print(answer)

相關內容

• 建立一個Fabric資料代理程式
• 資料代理作為模型情境協定伺服器
• Fabric 資料代理 SDK 範例
• Fabric REST API 參考