Fabric データ エージェント Python SDK (プレビュー)

Fabric データ エージェント Python SDK は、Fabricデータ エージェント成果物へのプログラムによるアクセスを提供します。 これは、Fabric ポータルを使用せずにデータ エージェントを作成、構成、更新、発行するコード優先ユーザー向けに設計されています。 sdk は、Microsoft Fabric ノートブック内で実行することも、Fabricに対する認証後に独自の環境から実行することもできます。

管理プレーンとランタイム

SDK は管理プレーン ツールです。 これを使用して、成果物の作成、データ ソースの追加と構成、手順とクエリ例の設定、発行などのデータ エージェントのライフサイクルを管理します。 Fabricパブリック REST API で実行されます。これは、他のワークスペースやアイテムに既に使用しているのと同じサーフェスであるため、認証と要求パターンは同じです。

管理プレーンは、実行時にデータ エージェントを照会する方法とは別です。 データ エージェントを発行したら、ツール、アプリケーション、およびその他のエージェント エクスペリエンスから、モデル コンテキスト プロトコル (MCP) エンドポイントを使用してクエリを実行します。 詳細については、「 モデル コンテキスト プロトコル サーバーとしてのデータ エージェント」を参照してください。

前提条件

• データ エージェントをサポートする容量を持つFabric ワークスペース。
• レイクハウス、ウェアハウス、Power BI セマンティック モデル、KQL データベースなど、ワークスペースでサポートされているデータ ソース。
• Python 3.10 以降。
• Fabric ノートブックの外部で実行する場合は、Azure CLIやサービス プリンシパルなど、Fabricに対して認証する方法です。

取り付け

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 エンドポイントは、データ エージェントのランタイムと消費サーフェスです。 ツール、アプリケーション、その他のエージェントから接続して、データ エージェントのクエリを実行したり、質問をしたり、回答を受け取ることができます。 セットアップ手順とサポートされているクライアントについては、「 モデル コンテキスト プロトコル サーバーとしてのデータ エージェント」を参照してください。

エンドポイント URL を取得する

MCP エンドポイント URL は、次の 2 つの方法で取得できます。

• データ エージェントの設定からコピーします。 データ エージェントを発行したら、エージェント設定の [ モデル コンテキスト プロトコル ] タブを開き、 MCP サーバーの URL をコピーします。 詳細については、「 モデル コンテキスト プロトコル サーバーとしてのデータ エージェント」を参照してください。

• 手動でビルドします。 次の形式を使用して、ワークスペース ID とデータ エージェント (成果物) ID から URL を構築します。

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

  プレースホルダーを、公開されたデータ エージェントの値に置き換えます。

  プレースホルダー	説明
  {WorkspaceId}	データ エージェントを含むFabric ワークスペースの ID。
  {DataAgentId}	公開されたデータエージェントの ID。

  手動で構築された URL は、データ エージェントを発行した後にのみ機能します。 データ エージェントが発行されていない場合、エンドポイントは URL が正しい場合でもエラーを返します。

Authentication

MCP エンドポイントへの要求は、Fabricに対して認証する必要があります。 クライアントに、ターゲット ワークスペースとデータ エージェントにアクセスするためのアクセス許可を持つ有効なベアラー トークンが含まれていることを確認します。 トークンは、ユーザー ID またはサービス プリンシパル (SPN) を表すことができます。

使用可能なツール

Fabric データ エージェントは、1 つのツールを提供する 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 リファレンス