Microsoft Kiota Bundle

1.10.1 · active · verified Thu Apr 16

microsoft-kiota-bundle is a convenience meta-package that installs all core Microsoft Kiota Python libraries. Kiota is a command-line tool for generating API clients based on OpenAPI descriptions. This bundle provides the necessary runtime components, including abstractions, authentication via Azure Identity, HTTP client implementations (based on Requests and Httpx), and serialization for JSON, Text, Form, and Multipart formats. The current version is 1.10.1, with releases typically synchronized across the Kiota Python ecosystem.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to set up core Kiota components using the `microsoft-kiota-bundle`. It initializes an `AzureIdentityAuthenticationProvider` for authentication (using `azure-identity` credentials), creates a `RequestsClient` for HTTP communication, and then uses Kiota's `RequestInformation` abstractions to make a simple GET request to Microsoft Graph. Note that in a real application, you would typically use a Kiota-generated client, which wraps these low-level components.

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

view raw JSON →