SDK Python para agente de dados de infraestrutura (pré-visualização)

O Fabric data agent Python SDK fornece acesso programático aos artefactos do Fabric data agent. Foi concebido para utilizadores que privilegiam o código e que pretendem criar, configurar, atualizar e publicar agentes de dados sem recorrer ao portal Fabric. Podes executar o SDK dentro de um portátil Microsoft Fabric, ou a partir do teu próprio ambiente depois de te autenticares no Fabric.

Plano de gestão e tempo de execução

O SDK é uma ferramenta de plano de gestão. Use-o para gerir o ciclo de vida de um agente de dados: crie o artefacto, adicione e configure fontes de dados, defina instruções e consultas de exemplo, e publique. Corre na API pública REST do Fabric, que é a mesma superfície que já usas para outros espaços de trabalho e itens, por isso autenticação e padrões de pedido são os mesmos.

O plano de gestão está separado da forma como se consulta o agente de dados em tempo de execução. Depois de publicar um agente de dados, consulte-o através do respetivo ponto final do Model Context Protocol (MCP) a partir de ferramentas, aplicações e outras experiências com agentes. Para mais informações, veja Agente de dados como servidor de Protocolo de Contexto de Modelo.

Pré-requisitos

• Um espaço de trabalho Fabric com capacidade que suporta agentes de dados.
• Uma fonte de dados suportada no espaço de trabalho, como uma casa de lago, armazém, modelo semântico Power BI ou base de dados KQL.
• Python 3.10 ou posterior.
• Para executar fora de um notebook do Fabric, é necessária uma forma de autenticação no Fabric, como a CLI do Azure ou uma entidade de serviço.

Instalação

O SDK é publicado no PyPI como fabric-data-agent-sdk. Instale-o usando pip:

%pip install fabric-data-agent-sdk

Autenticar-se no Fabric

Quando executas o SDK dentro de um notebook Fabric, o notebook trata da autenticação por ti.

Quando executares o SDK fora do Fabric, inicia sessão no Fabric primeiro. O exemplo seguinte inicia sessão com a credencial CLI do Azure e define-a como padrão para a sessão. Pode usar uma conta de utilizador ou um principal de serviço.

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

credential = AzureCliCredential()
SetFabricAnalyticsDefaultTokenCredentialsGlobally(credential)

Criar um agente de dados

Crie um agente de dados num espaço de trabalho. Substitua o ID do espaço de trabalho pelo ID do espaço de trabalho onde existe o agente de dados.

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,
)

Configure o agente de dados e adicione fontes de dados

Define as instruções do agente e depois adiciona uma fonte de dados. Use o ID de artefacto de uma casa de lago, armazém, modelo semântico ou base de dados KQL existente.

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,
)

Também pode adicionar instruções de fonte de dados e exemplos de consultas para melhorar a qualidade das respostas do agente de dados. Para saber mais, consulte os exemplos do SDK do Fabric data agent no GitHub.

Publicar o agente de dados

Publique a configuração em fases para disponibilizar o agente de dados para consultas.

agent.publish_staging(description="Initial publish")

Consultar um agente de dados publicado

Depois de publicar o agente de dados, consulte-o através do seu endpoint MCP. O endpoint MCP é a superfície de execução e consumo do agente de dados. Pode ligar-se a ela através de ferramentas, aplicações e outros agentes para consultar o agente de dados, colocar questões e receber respostas. Para passos de configuração e clientes suportados, veja Agente de dados como servidor de Protocolo de Contexto de Modelo.

Obtenha a URL do endpoint

Pode obter a URL do endpoint MCP de duas formas:

• Copie-o nas definições do agente de dados. Depois de publicar o agente de dados, abra o separador Model Context Protocol nas definições do agente e copie a URL do servidor MCP. Para mais detalhes, veja Agente de dados como servidor de Protocolo de Contexto de Modelo.

• Constrói-o manualmente. Constrói a URL a partir do ID do teu espaço de trabalho e ID do agente de dados (artefacto) usando o seguinte formato:

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

  Substitua os marcadores de posição por valores do seu agente de dados publicado:

  Marcador de posição	Description
  {WorkspaceId}	O ID do espaço de trabalho Fabric que contém o agente de dados.
  {DataAgentId}	O ID do agente de dados publicado.

  Uma URL construída manualmente só funciona depois de publicar o agente de dados. Se o agente de dados não for publicado, o endpoint devolve um erro mesmo quando a URL está correta.

Authentication

Os pedidos ao endpoint MCP devem ser autenticados no Fabric. Certifique-se de que o seu cliente inclui um token portador válido com permissões para aceder ao espaço de trabalho alvo e ao agente de dados. O token pode representar uma identidade de utilizador ou um principal de serviço (SPN).

Ferramentas disponíveis

O agente de dados Fabric expõe um servidor MCP que fornece uma única ferramenta. Os clientes recorrem a esta ferramenta para enviar uma pergunta ao agente de dados e receber a resposta gerada.

Consultar o agente de dados a partir do Python

O exemplo seguinte liga-se ao endpoint MCP, descobre a ferramenta, envia uma pergunta e imprime a resposta. Reutiliza o credential do passo Autenticar no Fabric e utiliza o MCP Python SDK. Instala primeiro o 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)

Conteúdo relacionado

• Criar um agente de dados Fabric
• Agente de dados como um servidor do Model Context Protocol
• Amostras do SDK do Fabric data agent
• Referência da API REST do Fabric