Agente de dados de tecido Python SDK (versão prévia)

O SDK do Python para o agente de dados do Fabric fornece acesso programático aos artefatos do agente de dados do Fabric. Ele foi projetado para usuários de primeiro código que desejam criar, configurar, atualizar e publicar agentes de dados sem usar o portal Fabric. Você pode executar o SDK dentro de um bloco de anotações Microsoft Fabric ou em seu próprio ambiente depois de autenticar para Fabric.

Plano de gerenciamento e tempo de execução

O SDK é uma ferramenta de plano de gerenciamento. Use-o para gerenciar o ciclo de vida de um agente de dados: crie o artefato, adicione e configure fontes de dados, defina instruções e consultas de exemplo e publique. Ele funciona com base na API REST pública do Fabric, que é a mesma interface que você já usa para outros workspaces e itens, portanto, a autenticação e os padrões de requisição são os mesmos.

O plano de gerenciamento é separado de como você consulta o agente de dados em runtime. Depois de publicar um agente de dados, consulte-o por meio do endpoint MCP (Protocolo de Contexto de Modelo) em ferramentas, aplicativos e outras experiências agênticas. Para obter mais informações, consulte o agente de dados como um servidor de protocolo de contexto de modelo.

Pré-requisitos

• Um workspace do Fabric com uma capacidade que oferece suporte a agentes de dados.
• Uma fonte de dados compatível no workspace, como um lakehouse, um warehouse, um modelo semântico do Power BI ou um banco de dados KQL.
• Python 3.10 ou posterior.
• Para execução fora de um notebook do Fabric, uma forma de se autenticar 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 você executa o SDK dentro de um notebook do Fabric, o notebook cuida da autenticação para você.

Quando você executar o SDK fora do Fabric, entre no Fabric primeiro. O exemplo a seguir entra com a credencial CLI do Azure e a define como o padrão para a sessão. Você pode usar uma conta de usuário ou uma entidade 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 em um workspace. Substitua o ID do espaço de trabalho pelo ID do espaço de trabalho onde o agente de dados está.

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

Configurar o agente de dados e adicionar fontes de dados

Defina as instruções do agente e adicione uma fonte de dados. Use o ID do artefato de um lakehouse, warehouse, modelo semântico ou banco 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,
)

Você também pode adicionar instruções de fonte de dados e consultas de exemplo para melhorar a qualidade das respostas do agente de dados. Para saber mais, confira os exemplos do SDK do agente de dados Fabric no GitHub.

Publicar o agente de dados

Publique a configuração em etapas para disponibilizar o agente de dados para consulta.

agent.publish_staging(description="Initial publish")

Consultar um agente de dados publicado

Depois de publicar o agente de dados, consulte-o por seu endpoint MCP. O endpoint MCP é a interface de tempo de execução e consumo para o agente de dados. Você pode se conectar a ele por meio de ferramentas, aplicativos e outros agentes para consultar o agente de dados, fazer perguntas e receber respostas. Para ver as etapas de configuração e os clientes compatíveis, consulte o agente de dados como um servidor do Model Context Protocol.

Obter a URL do ponto de extremidade

Você pode obter a URL do ponto de extremidade MCP de duas maneiras:

• Copie-o das configurações do agente de dados. Depois de publicar o agente de dados, abra a guia Protocolo de Contexto do Modelo nas configurações do agente e copie a URL do servidor MCP. Para obter detalhes, consulte o agente de dados como um servidor de protocolo de contexto de modelo.

• Compile-o manualmente. Construa a URL a partir da ID do seu workspace e da ID do agente de dados (artefato), usando o seguinte formato:

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

  Substitua os marcadores pelos valores do seu agente de dados publicado:

  Espaço reservado	Description
  {WorkspaceId}	O ID do workspace do Fabric que contém o agente de dados.
  {DataAgentId}	O identificador do agente de dados publicado.

  Uma URL criada manualmente só funciona depois que você publica o agente de dados. Se o agente de dados não estiver publicado, o endpoint retornará um erro mesmo quando a URL estiver correta.

Authentication

As solicitações ao endpoint MCP devem ser autenticadas no Fabric. Verifique se o cliente inclui um token de portador válido com permissões para acessar o workspace de destino e o agente de dados. O token pode representar uma identidade de usuário ou um SPN (entidade de serviço).

Ferramentas disponíveis

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

Consultar o agente de dados de Python

O exemplo a seguir conecta-se ao ponto de extremidade MCP, descobre a ferramenta, envia uma pergunta e imprime a resposta. Reutiliza o credential da etapa Autenticação no Fabric e usa o SDK do MCP para Python. Instale o SDK primeiro:

%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
• exemplos do SDK do agente de dados Fabric
• Referência da API REST do Fabric