agent-framework-purview

import os import asyncio from agent_framework_purview.purview_agent import PurviewAgent from agent_framework.core.processor import AgentProcessor from azure.identity.aio import DefaultAzureCredential async def main(): # Ensure AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, etc., are set # for DefaultAzureCredential to authenticate. For local development, # 'az login' or environment variables are common. For production, # Managed Identity is recommended. credential = DefaultAzureCredential() # The Purview Agent requires the Azure Purview account name. # Replace 'your_purview_account_name' or set AZURE_PURVIEW_ACCOUNT_NAME env var. purview_account_name = os.environ.get( "AZURE_PURVIEW_ACCOUNT_NAME", "your_purview_account_name" ) if purview_account_name == "your_purview_account_name": print("WARNING: Please set AZURE_PURVIEW_ACCOUNT_NAME environment variable or update the code.") return purview_agent_config = { "azure_purview_account_name": purview_account_name, "purview_scope": ["https://purview.azure.net/.default"], # Standard scope } try: purview_agent: PurviewAgent = await AgentProcessor.create_agent( PurviewAgent, credential=credential, config=purview_agent_config ) print(f"Agent created for Purview account: {purview_account_name}") # Example invocation: find assets matching a query string print("Invoking agent to find assets...") response = await purview_agent.invoke( "find assets", query_string="Sales Data" ) print("\n--- Purview Agent Response ---") print(f"Content: {response.content}") if response.citations: print(f"Citations: {response.citations}") else: print("No citations found.") except Exception as e: print(f"An error occurred: {e}") finally: await credential.close() if __name__ == "__main__": asyncio.run(main())