ตัวแทนข้อมูลผ้า Python SDK (ตัวอย่าง)

Fabric Data Agent Python SDK ให้การเข้าถึงสิ่งประดิษฐ์ของตัวแทนข้อมูล Fabric โดยทางโปรแกรม ออกแบบมาสําหรับผู้ใช้ที่เน้นโค้ดเป็นอันดับแรกที่ต้องการสร้าง กําหนดค่า อัปเดต และเผยแพร่ตัวแทนข้อมูลโดยไม่ต้องใช้พอร์ทัล Fabric คุณสามารถเรียกใช้ SDK ภายในสมุดบันทึก Microsoft Fabric หรือจากสภาพแวดล้อมของคุณเองหลังจากที่คุณรับรองความถูกต้องกับ Fabric

ระนาบการจัดการและรันไทม์

SDK เป็นเครื่องมือระนาบการจัดการ ใช้เพื่อจัดการวงจรชีวิตของตัวแทนข้อมูล: สร้างสิ่งประดิษฐ์ เพิ่มและกําหนดค่าแหล่งข้อมูล ตั้งค่าคําแนะนําและตัวอย่างคิวรี และเผยแพร่ ทํางานบน REST API สาธารณะของ Fabric ซึ่งเป็นพื้นผิวเดียวกับที่คุณใช้สําหรับพื้นที่ทํางานและรายการอื่นๆ อยู่แล้ว ดังนั้นการรับรองความถูกต้องและรูปแบบคําขอจึงเหมือนกัน

ระนาบการจัดการจะแยกต่างหากจากวิธีที่คุณคิวรีตัวแทนข้อมูลในขณะรันไทม์ หลังจากที่คุณเผยแพร่ตัวแทนข้อมูลแล้ว ให้คิวรีผ่านจุดสิ้นสุด Model Context Protocol (MCP) จากเครื่องมือ แอปพลิเคชัน และประสบการณ์ตัวแทนอื่นๆ หากต้องการข้อมูลเพิ่มเติม โปรดดู ตัวแทนข้อมูลเป็นเซิร์ฟเวอร์โปรโตคอลบริบทแบบจําลอง

Prerequisites

• พื้นที่ทํางาน Fabric ที่มีความจุที่รองรับตัวแทนข้อมูล
• แหล่งข้อมูลที่ได้รับการสนับสนุนในพื้นที่ทํางาน เช่น เลคเฮาส์ คลังสินค้า แบบจําลองความหมายของ Power BI หรือฐานข้อมูล KQL
• Python 3.10 หรือใหม่กว่า
• สําหรับการดําเนินการภายนอกสมุดบันทึก Fabric วิธีการรับรองความถูกต้องกับ Fabric เช่น Azure CLI หรือบริการหลัก

Installation

SDK เผยแพร่บน PyPI เป็น fabric-data-agent-sdk ติดตั้งโดยใช้ pip:

%pip install fabric-data-agent-sdk

ตรวจสอบสิทธิ์กับ Fabric

เมื่อคุณเรียกใช้ SDK ภายในสมุดบันทึก Fabric สมุดบันทึกจะจัดการการรับรองความถูกต้องให้คุณ

เมื่อคุณเรียกใช้ SDK ภายนอก Fabric ให้ลงชื่อเข้าใช้ Fabric ก่อน ตัวอย่างต่อไปนี้ลงชื่อเข้าใช้ด้วยข้อมูลประจําตัว Azure CLI และตั้งค่าเป็นค่าเริ่มต้นสําหรับเซสชัน คุณสามารถใช้บัญชีผู้ใช้หรือบริการหลัก

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

credential = AzureCliCredential()
SetFabricAnalyticsDefaultTokenCredentialsGlobally(credential)

สร้างตัวแทนข้อมูล

สร้างตัวแทนข้อมูลในพื้นที่ทํางาน แทนที่รหัสพื้นที่ทํางานด้วยรหัสของพื้นที่ทํางานที่มีตัวแทนข้อมูลอยู่

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 ที่มีอยู่

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

คุณยังสามารถเพิ่มคําแนะนําแหล่งข้อมูลและตัวอย่างคิวรีเพื่อปรับปรุงคุณภาพของคําตอบของตัวแทนข้อมูล หากต้องการเรียนรู้เพิ่มเติม โปรดดูตัวอย่าง SDK ตัวแทนข้อมูล Fabric บน GitHub

เผยแพร่ตัวแทนข้อมูล

เผยแพร่การกําหนดค่าแบบลําดับขั้นเพื่อให้ตัวแทนข้อมูลพร้อมใช้งานสําหรับการสืบค้น

agent.publish_staging(description="Initial publish")

สอบถามตัวแทนข้อมูลที่เผยแพร่

หลังจากเผยแพร่ตัวแทนข้อมูลแล้ว ให้สืบค้นผ่านปลายทาง MCP จุดสิ้นสุด MCP คือรันไทม์และพื้นผิวการใช้สําหรับตัวแทนข้อมูล คุณสามารถเชื่อมต่อได้จากเครื่องมือ แอปพลิเคชัน และตัวแทนอื่นๆ เพื่อสอบถามตัวแทนข้อมูล ถามคําถาม และรับคําตอบ สําหรับขั้นตอนการตั้งค่าและไคลเอ็นต์ที่รองรับ โปรดดู ตัวแทนข้อมูลเป็นเซิร์ฟเวอร์โปรโตคอลบริบทแบบจําลอง

รับ URL ปลายทาง

คุณรับ URL ปลายทาง MCP ได้ 2 วิธี ดังนี้

• คัดลอกจากการตั้งค่าตัวแทนข้อมูล หลังจากเผยแพร่ตัวแทนข้อมูลแล้ว ให้เปิดแท็บโปรโตคอลบริบทแบบจําลองในการตั้งค่าเอเจนต์ แล้วคัดลอก URL ของเซิร์ฟเวอร์ MCP โปรดดูรายละเอียดที่หัวข้อตัวแทนข้อมูลเป็นเซิร์ฟเวอร์โปรโตคอลบริบทแบบจําลอง

• สร้างด้วยตนเอง สร้าง URL จากรหัสพื้นที่ทํางานและรหัสตัวแทนข้อมูล (สิ่งประดิษฐ์) โดยใช้รูปแบบต่อไปนี้:

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

  แทนที่ตัวยึดด้วยค่าจากตัวแทนข้อมูลที่เผยแพร่ของคุณ:

  ตัวกำหนดตำแหน่ง	คำอธิบาย
  {WorkspaceId}	รหัสของพื้นที่ทํางาน Fabric ที่มีตัวแทนข้อมูล
  {DataAgentId}	รหัสของตัวแทนข้อมูลที่เผยแพร่

  URL ที่สร้างขึ้นด้วยตนเองจะทํางานหลังจากที่คุณเผยแพร่ตัวแทนข้อมูลแล้วเท่านั้น หากไม่ได้เผยแพร่ตัวแทนข้อมูล ปลายทางจะส่งคืนข้อผิดพลาดแม้ว่า URL จะถูกต้องก็ตาม

การรับรองความถูกต้อง

คําขอไปยังปลายทาง MCP ต้องได้รับการรับรองความถูกต้องกับ Fabric ตรวจสอบให้แน่ใจว่าไคลเอ็นต์ของคุณมีโทเค็นผู้ถือที่ถูกต้องพร้อมสิทธิ์ในการเข้าถึงพื้นที่ทํางานเป้าหมายและตัวแทนข้อมูล โทเค็นสามารถแสดงข้อมูลประจําตัวของผู้ใช้หรือบริการหลัก (SPN)

เครื่องมือที่พร้อมใช้งาน

ตัวแทนข้อมูล Fabric แสดงเซิร์ฟเวอร์ MCP ที่มีเครื่องมือเดียว ไคลเอ็นต์เรียกใช้เครื่องมือนี้เพื่อส่งคําถามไปยังตัวแทนข้อมูลและรับคําตอบที่สร้างขึ้น

สืบค้นตัวแทนข้อมูลจาก Python

ตัวอย่างต่อไปนี้เชื่อมต่อกับปลายทาง MCP ค้นพบเครื่องมือ ส่งคําถาม และพิมพ์คําตอบ ใช้ขั้นตอนจากการตรวจสอบสิทธิ์เป็น Fabric ซ้ํา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
• ตัวแทนข้อมูลเป็นเซิร์ฟเวอร์ Model Context Protocol
• ตัวอย่าง SDK ตัวแทนข้อมูล Fabric
• การอ้างอิง Fabric REST API