microsoft-kiota-bundle

import asyncio import os from azure.identity import ClientSecretCredential from microsoft_kiota_authentication_azure.azure_identity_authentication_provider import AzureIdentityAuthenticationProvider from microsoft_kiota_http.requests_client import RequestsClient from microsoft_kiota_abstractions.request_information import RequestInformation from microsoft_kiota_abstractions.method import Method from microsoft_kiota_abstractions.base_request_configuration import BaseRequestConfiguration # --- Environment variables for Azure authentication --- # Replace with your actual values or set as environment variables TENANT_ID = os.environ.get('AZURE_TENANT_ID', 'YOUR_TENANT_ID') CLIENT_ID = os.environ.get('AZURE_CLIENT_ID', 'YOUR_CLIENT_ID') CLIENT_SECRET = os.environ.get('AZURE_CLIENT_SECRET', 'YOUR_CLIENT_SECRET') # Define the scopes required for your API calls SCOPES = [os.environ.get('AZURE_SCOPES', 'https://graph.microsoft.com/.default')] async def main(): if not all([TENANT_ID, CLIENT_ID, CLIENT_SECRET]): print("Please set AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET environment variables (or hardcode for testing).") print("Using dummy values for demonstration, this will likely fail.") # Use dummy values to allow execution for demonstration purposes if env vars are missing credential = ClientSecretCredential("dummy_tenant", "dummy_client", "dummy_secret") else: # 1. Initialize Azure Identity credential credential = ClientSecretCredential( tenant_id=TENANT_ID, client_id=CLIENT_ID, client_secret=CLIENT_SECRET ) # 2. Create an Azure Identity Authentication Provider auth_provider = AzureIdentityAuthenticationProvider( credential=credential, scopes=SCOPES ) # 3. Create an HTTP Client using the authentication provider # The microsoft-kiota-bundle includes RequestsClient and HttpxClient http_client = RequestsClient(authentication_provider=auth_provider) # 4. Construct a Request Information object (typically done by a generated Kiota client) request_info = RequestInformation() request_info.http_method = Method.GET request_info.url_template = "https://graph.microsoft.com/v1.0/me" request_info.set_header("Accept", "application/json") print(f"Attempting to make a GET request to: {request_info.url_template}") try: # 5. Send the request and get the response response = await http_client.send(request_info, BaseRequestConfiguration()) if response: print(f"Response Status: {response.status_code}") # For a real Kiota generated client, the response would be deserialized automatically. # For demonstration, you might read the text content if successful. if 200 <= response.status_code < 300: body_content = await response.text() print(f"Response Body (truncated): {body_content[:500]}...") else: print(f"Error Response: {await response.text()}") else: print("No response received.") except Exception as e: print(f"An error occurred during the request: {e}") if __name__ == '__main__': asyncio.run(main())