Azure SDK

Production patterns for Azure services in Python using the official Azure SDKs.

When to Activate

Writing code that imports azure-storage-blob, azure-search-documents, azure-ai-formrecognizer, or azure-identity
Configuring authentication for Azure services (Managed Identity, service principals, connection strings)
Designing or querying an Azure AI Search index (vector, text, hybrid)
Extracting content from documents using Azure Document Intelligence
Managing secrets with Azure Key Vault
Deploying a pipeline as an Azure Function
Troubleshooting Azure SDK errors or retry behavior

Authentication

DefaultAzureCredential (always prefer this)

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

credential = DefaultAzureCredential()
client = BlobServiceClient(account_url="https://<account>.blob.core.windows.net", credential=credential)

DefaultAzureCredential tries, in order: environment variables → Managed Identity → Azure CLI → VS Code → Interactive browser. The same code works locally (via CLI auth) and in production (via Managed Identity) without changes.

# BAD: connection string hardcoded
client = BlobServiceClient.from_connection_string("DefaultEndpointsProtocol=https;AccountName=...")

# BAD: key hardcoded
client = BlobServiceClient(account_url=url, credential="storage-account-key-here")

# GOOD: keyless auth
credential = DefaultAzureCredential()
client = BlobServiceClient(account_url=url, credential=credential)

Auth decision matrix

Environment	Credential type	How to enable
Local dev	Azure CLI	`az login`
CI/CD	Service principal (env vars)	Set `AZURE_CLIENT_ID`, `AZURE_CLIENT_SECRET`, `AZURE_TENANT_ID`
Azure VM / AKS	System-assigned Managed Identity	Enable on the resource in portal/Bicep
Azure Functions	User-assigned Managed Identity	Set `AZURE_CLIENT_ID` env var
Testing	`ClientSecretCredential`	Explicit — never use in production code

Configuration (pydantic-settings)

from pydantic_settings import BaseSettings

class AzureSettings(BaseSettings):
    azure_storage_account_url: str
    azure_search_endpoint: str
    azure_search_index_name: str
    azure_document_intelligence_endpoint: str
    azure_key_vault_url: str | None = None

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = AzureSettings()

Never store credentials in settings — let DefaultAzureCredential handle them.

Azure Blob Storage

Upload

from azure.storage.blob import BlobServiceClient, ContentSettings

def upload_file(account_url: str, container: str, blob_name: str, data: bytes, content_type: str) -> str:
    credential = DefaultAzureCredential()
    client = BlobServiceClient(account_url=account_url, credential=credential)
    blob = client.get_blob_client(container=container, blob=blob_name)
    blob.upload_blob(
        data,
        overwrite=True,
        content_settings=ContentSettings(content_type=content_type),
    )
    return blob.url

Download and list

def download_blob(account_url: str, container: str, blob_name: str) -> bytes:
    client = BlobServiceClient(account_url=account_url, credential=DefaultAzureCredential())
    blob = client.get_blob_client(container=container, blob=blob_name)
    return blob.download_blob().readall()

def list_blobs(account_url: str, container: str, prefix: str = "") -> list[str]:
    client = BlobServiceClient(account_url=account_url, credential=DefaultAzureCredential())
    container_client = client.get_container_client(container)
    return [b.name for b in container_client.list_blobs(name_starts_with=prefix)]

SAS token (time-limited read access)

from datetime import datetime, timedelta, timezone
from azure.storage.blob import generate_blob_sas, BlobSasPermissions

def get_sas_url(account_name: str, account_key: str, container: str, blob: str, expiry_hours: int = 1) -> str:
    sas = generate_blob_sas(
        account_name=account_name,
        container_name=container,
        blob_name=blob,
        account_key=account_key,
        permission=BlobSasPermissions(read=True),
        expiry=datetime.now(timezone.utc) + timedelta(hours=expiry_hours),
    )
    return f"https://{account_name}.blob.core.windows.net/{container}/{blob}?{sas}"

Azure AI Search

Index schema (with vector field)

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex, SimpleField, SearchableField, SearchFieldDataType,
    VectorSearch, HnswAlgorithmConfiguration, VectorSearchProfile,
    SearchField, SemanticConfiguration, SemanticSearch, SemanticPrioritizedFields,
    SemanticField,
)

def create_index(endpoint: str, index_name: str) -> None:
    client = SearchIndexClient(endpoint=endpoint, credential=DefaultAzureCredential())
    fields = [
        SimpleField(name="id", type=SearchFieldDataType.String, key=True),
        SearchableField(name="content", type=SearchFieldDataType.String),
        SearchableField(name="title", type=SearchFieldDataType.String),
        SimpleField(name="source", type=SearchFieldDataType.String, filterable=True),
        SimpleField(name="chunk_index", type=SearchFieldDataType.Int32, filterable=True),
        SearchField(
            name="content_vector",
            type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
            searchable=True,
            vector_search_dimensions=1536,
            vector_search_profile_name="hnsw-profile",
        ),
    ]
    vector_search = VectorSearch(
        algorithms=[HnswAlgorithmConfiguration(name="hnsw")],
        profiles=[VectorSearchProfile(name="hnsw-profile", algorithm_configuration_name="hnsw")],
    )
    semantic_search = SemanticSearch(
        configurations=[
            SemanticConfiguration(
                name="default",
                prioritized_fields=SemanticPrioritizedFields(
                    content_fields=[SemanticField(field_name="content")],
                    title_field=SemanticField(field_name="title"),
                ),
            )
        ]
    )
    index = SearchIndex(
        name=index_name,
        fields=fields,
        vector_search=vector_search,
        semantic_search=semantic_search,
    )
    client.create_or_update_index(index)

Upload documents

from azure.search.documents import SearchClient

def upload_documents(endpoint: str, index_name: str, docs: list[dict]) -> None:
    client = SearchClient(
        endpoint=endpoint,
        index_name=index_name,
        credential=DefaultAzureCredential(),
    )
    # Batch in chunks of 1000 (SDK limit)
    for i in range(0, len(docs), 1000):
        result = client.upload_documents(documents=docs[i:i + 1000])
        failed = [r for r in result if not r.succeeded]
        if failed:
            raise RuntimeError(f"{len(failed)} documents failed to index: {failed[0].key}")

Search: text / vector / hybrid

from azure.search.documents.models import VectorizedQuery

def search(
    endpoint: str,
    index_name: str,
    query: str,
    query_vector: list[float],
    top: int = 5,
    mode: str = "hybrid",  # "text" | "vector" | "hybrid"
    filter_expr: str | None = None,
) -> list[dict]:
    client = SearchClient(endpoint=endpoint, index_name=index_name, credential=DefaultAzureCredential())

    vector_query = VectorizedQuery(
        vector=query_vector,
        k_nearest_neighbors=top,
        fields="content_vector",
    ) if mode in ("vector", "hybrid") else None

    results = client.search(
        search_text=query if mode in ("text", "hybrid") else None,
        vector_queries=[vector_query] if vector_query else None,
        filter=filter_expr,
        top=top,
        query_type="semantic" if mode == "hybrid" else "simple",
        semantic_configuration_name="default" if mode == "hybrid" else None,
    )
    return [dict(r) for r in results]

Search mode comparison

Mode	When to use	Relevance	Cost
Text	Keyword lookup, exact matches	Low	Lowest
Vector	Semantic similarity, paraphrase	High	Medium
Hybrid	Production RAG (default choice)	Highest	Medium
Semantic reranking	High-precision Q&A on top of hybrid	Highest	Higher

Azure Document Intelligence

from azure.ai-formrecognizer import DocumentAnalysisClient

def analyze_document(endpoint: str, file_bytes: bytes, model_id: str = "prebuilt-read") -> dict:
    client = DocumentAnalysisClient(endpoint=endpoint, credential=DefaultAzureCredential())
    poller = client.begin_analyze_document(model_id, document=file_bytes)
    result = poller.result()
    return {
        "content": result.content,
        "pages": len(result.pages),
        "tables": [
            {
                "row_count": t.row_count,
                "column_count": t.column_count,
                "cells": [{"row": c.row_index, "col": c.column_index, "text": c.content} for c in t.cells],
            }
            for t in (result.tables or [])
        ],
    }

Model selection

Model ID	Best for
`prebuilt-read`	Text extraction from any document
`prebuilt-layout`	Tables, checkboxes, structure-aware extraction
`prebuilt-document`	Key-value pairs + tables
`prebuilt-invoice`	Invoices
`prebuilt-receipt`	Receipts
Custom model	Domain-specific forms with consistent layout

Key Vault

from azure.keyvault.secrets import SecretClient

def get_secret(vault_url: str, secret_name: str) -> str:
    client = SecretClient(vault_url=vault_url, credential=DefaultAzureCredential())
    return client.get_secret(secret_name).value

# Cache the client — don't recreate per call
_kv_client: SecretClient | None = None

def kv_client(vault_url: str) -> SecretClient:
    global _kv_client
    if _kv_client is None:
        _kv_client = SecretClient(vault_url=vault_url, credential=DefaultAzureCredential())
    return _kv_client

Retry with tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from azure.core.exceptions import HttpResponseError, ServiceRequestError

def is_retryable(exc: Exception) -> bool:
    if isinstance(exc, HttpResponseError):
        return exc.status_code in (429, 500, 502, 503, 504)
    return isinstance(exc, ServiceRequestError)

@retry(
    retry=retry_if_exception_type((HttpResponseError, ServiceRequestError)),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(5),
    reraise=True,
)
def upload_with_retry(client: SearchClient, docs: list[dict]) -> None:
    client.upload_documents(documents=docs)

Error handling

from azure.core.exceptions import (
    HttpResponseError,
    ResourceNotFoundError,
    ResourceExistsError,
    ClientAuthenticationError,
    ServiceRequestError,
)

try:
    result = client.get_document(key="doc-123")
except ResourceNotFoundError:
    # Document does not exist — handle gracefully
    return None
except ClientAuthenticationError:
    # Credential expired or RBAC role missing — fail fast
    raise
except HttpResponseError as e:
    if e.status_code == 429:
        # Throttled — tenacity will handle retry
        raise
    logger.error("azure_error", status=e.status_code, message=e.message)
    raise

Cost controls

Lever	Impact	How
AI Search tier	High	`Basic` for dev, `Standard S1` for prod; avoid `S3 HD` unless >1B docs
Semantic reranking	Medium	Enable only on queries that need it; billed per 1000 queries
Document Intelligence	Medium	Use `prebuilt-read` (cheapest) unless you need tables or KV pairs
Blob storage tier	Low-medium	`Hot` for active docs, `Cool` for archive; lifecycle policies auto-tier
Vector dimensions	Medium	1536 (ada-002) vs 3072 (text-embedding-3-large) — smaller = cheaper storage

Red Flags

Hardcoded connection strings or storage account keys — keys can be leaked or rotated; always use DefaultAzureCredential with RBAC roles, never access keys or SAS tokens in code
DefaultAzureCredential in production without pinning to ManagedIdentityCredential — the credential chain tries 6+ sources sequentially; a misconfigured chain causes 30s+ startup failures; pin to ManagedIdentityCredential in prod
SDK clients recreated per request — SDK clients are designed to be long-lived and manage connection pools; recreating them per request exhausts connections and slows every call
Uploading documents to AI Search one at a time — single-document uploads are ~100× slower than batching; always use upload_documents in batches of up to 1000
Text-only search for RAG queries — semantic/vector-only search misses exact-match terms; use hybrid search (text + vector) with semantic re-ranking for best recall across diverse queries
No retry policy on 429 or 503 responses — Azure services throttle under load; wrap all SDK calls with tenacity or the Azure SDK's built-in retry configuration
ClientAuthenticationError silently retried — auth errors must fail fast and loudly; retrying authentication failures burns through retry budget and delays surfacing the real problem

Checklist

All SDK clients use DefaultAzureCredential — no hardcoded keys or connection strings
Managed Identity enabled on compute (Function App, VM, AKS node pool)
RBAC roles assigned (Storage Blob Data Contributor, Search Index Data Contributor, etc.) — not access keys
Secrets stored in Key Vault, not env vars or config files
All long-running SDK calls wrapped with tenacity retry on 429/5xx
Document upload batched in chunks of ≤1000 for Azure AI Search
Index schema reviewed: filterable/sortable fields declared explicitly
Hybrid search enabled for RAG queries (not text-only)
Blob lifecycle policy configured to auto-tier cold data to Cool/Archive
ClientAuthenticationError caught and surfaced immediately (not retried)
SDK client instances reused per process — not recreated per request
Azure resource names follow naming convention (<service>-<env>-<region>-<suffix>)

azure

SKILL.mdskills/azure/skill.md