azure-monitor-ingestion-py

OverviewHistoryStatsSecurity

npx skills add microsoft/agent-skills --skill azure-monitor-ingestion-py

Documentation

SKILL.md

npx skills add ...

Documentation

SKILL.md

Azure Monitor Ingestion SDK for Python. Use for sending custom logs to Log Analytics workspace via Logs Ingestion API. Triggers: "azure-monitor-ingestion", "LogsIngestionClient", "custom logs", "DCR", "data collection rule", "Log Analytics".

This repo is now called microsoft/skills. Both names install the same content, but the install count here only covers this one.

npx skills add microsoft/agent-skills --skill azure-monitor-ingestion-py

Documentation

SKILL.mdView source

Azure Monitor Ingestion SDK for Python

Send custom logs to Azure Monitor Log Analytics workspace using the Logs Ingestion API.

Installation

pip install azure-monitor-ingestion
pip install azure-identity

Environment Variables

# Data Collection Endpoint (DCE)
AZURE_DCE_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com  # Required for all auth methods
 
# Data Collection Rule (DCR) immutable ID
AZURE_DCR_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # Required for all auth methods
 
# Stream name from DCR
AZURE_DCR_STREAM_NAME=Custom-MyTable_CL  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Prerequisites

Before using this SDK, you need:

Log Analytics Workspace — Target for your logs
Data Collection Endpoint (DCE) — Ingestion endpoint
Data Collection Rule (DCR) — Defines schema and destination
Custom Table — In Log Analytics (created via DCR or manually)

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.

Local dev: DefaultAzureCredential works as-is.

Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.

Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:

Sync: with <Client>(...) as client:

Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)

Snippets may abbreviate this setup, but production code should always follow both rules.

from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
import os
 
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
 
with LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=credential
) as client:
    # Use `client.upload(...)` for all subsequent operations (see examples below)
    ...

Upload Custom Logs

from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential
import os
 
rule_id = os.environ["AZURE_DCR_RULE_ID"]
stream_name = os.environ["AZURE_DCR_STREAM_NAME"]
 
logs = [
    {"TimeGenerated": "2024-01-15T10:00:00Z", "Computer": "server1", "Message": "Application started"},
    {"TimeGenerated": "2024-01-15T10:01:00Z", "Computer": "server1", "Message": "Processing request"},
    {"TimeGenerated": "2024-01-15T10:02:00Z", "Computer": "server2", "Message": "Connection established"}
]
 
with LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=DefaultAzureCredential()
) as client:
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

Upload from JSON File

import json
 
with open("logs.json", "r") as f:
    logs = json.load(f)
 
client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

Custom Error Handling

Handle partial failures with a callback:

failed_logs = []
 
def on_error(error):
    print(f"Upload failed: {error.error}")
    failed_logs.extend(error.failed_logs)
 
client.upload(
    rule_id=rule_id,
    stream_name=stream_name,
    logs=logs,
    on_error=on_error
)
 
# Retry failed logs
if failed_logs:
    print(f"Retrying {len(failed_logs)} failed logs...")
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=failed_logs)

Ignore Errors

def ignore_errors(error):
    pass  # Silently ignore upload failures
 
client.upload(
    rule_id=rule_id,
    stream_name=stream_name,
    logs=logs,
    on_error=ignore_errors
)

Async Client

import asyncio
from azure.monitor.ingestion.aio import LogsIngestionClient
from azure.identity.aio import DefaultAzureCredential
 
async def upload_logs():
    async with LogsIngestionClient(
        endpoint=endpoint,
        credential=DefaultAzureCredential()
    ) as client:
        await client.upload(
            rule_id=rule_id,
            stream_name=stream_name,
            logs=logs
        )
 
asyncio.run(upload_logs())

Sovereign Clouds

from azure.identity import AzureAuthorityHosts, DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient
 
# Azure Government
credential = DefaultAzureCredential(authority=AzureAuthorityHosts.AZURE_GOVERNMENT)
with LogsIngestionClient(
    endpoint="https://example.ingest.monitor.azure.us",
    credential=credential,
    credential_scopes=["https://monitor.azure.us/.default"]
) as client:
    # client.upload(...)
    ...

Batching Behavior

The SDK automatically:

Splits logs into chunks of 1MB or less
Compresses each chunk with gzip
Uploads chunks in parallel

No manual batching needed for large log sets.

Client Types

Client	Purpose
`LogsIngestionClient`	Sync client for uploading logs
`LogsIngestionClient` (aio)	Async client for uploading logs

Key Concepts

Concept	Description
DCE	Data Collection Endpoint — ingestion URL
DCR	Data Collection Rule — defines schema, transformations, destination
Stream	Named data flow within a DCR
Custom Table	Target table in Log Analytics (ends with `_CL`)

DCR Stream Name Format

Stream names follow patterns:

Custom-<TableName>_CL — For custom tables
Microsoft-<TableName> — For built-in tables

Best Practices

Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.
Always use context managers for clients and async credentials. Wrap every client in with Client(...) as client: (sync) or async with Client(...) as client: (async) to ensure proper cleanup. For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
Use DefaultAzureCredential for code that runs locally. Use a specific token credential for code that runs in Azure.
Handle errors gracefully — use on_error callback for partial failures
Include TimeGenerated — Required field for all logs
Match DCR schema — Log fields must match DCR column definitions
Use async client for high-throughput scenarios
Batch uploads — SDK handles batching, but send reasonable chunks
Monitor ingestion — Check Log Analytics for ingestion status

Reference Files

File	Contents
references/capabilities.md	Additional non-hero capabilities, operation-group coverage, and production checklists.
references/non-hero-scenarios.md	Dedicated non-hero examples for secondary/advanced scenarios.

npx skills add microsoft/agent-skills --skill azure-monitor-ingestion-py

Documentation

SKILL.md

This repo is now called microsoft/skills. Both names install the same content, but the install count here only covers this one.

npx skills add microsoft/agent-skills --skill azure-monitor-ingestion-py

Documentation

SKILL.mdView source

Azure Monitor Ingestion SDK for Python

Send custom logs to Azure Monitor Log Analytics workspace using the Logs Ingestion API.

Installation

pip install azure-monitor-ingestion
pip install azure-identity

Environment Variables

# Data Collection Endpoint (DCE)
AZURE_DCE_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com  # Required for all auth methods
 
# Data Collection Rule (DCR) immutable ID
AZURE_DCR_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # Required for all auth methods
 
# Stream name from DCR
AZURE_DCR_STREAM_NAME=Custom-MyTable_CL  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Prerequisites

Before using this SDK, you need:

Log Analytics Workspace — Target for your logs
Data Collection Endpoint (DCE) — Ingestion endpoint
Data Collection Rule (DCR) — Defines schema and destination
Custom Table — In Log Analytics (created via DCR or manually)

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.

Local dev: DefaultAzureCredential works as-is.

Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.

Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:

Sync: with <Client>(...) as client:

Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)

Snippets may abbreviate this setup, but production code should always follow both rules.

from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
import os
 
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
 
with LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=credential
) as client:
    # Use `client.upload(...)` for all subsequent operations (see examples below)
    ...

Upload Custom Logs

from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential
import os
 
rule_id = os.environ["AZURE_DCR_RULE_ID"]
stream_name = os.environ["AZURE_DCR_STREAM_NAME"]
 
logs = [
    {"TimeGenerated": "2024-01-15T10:00:00Z", "Computer": "server1", "Message": "Application started"},
    {"TimeGenerated": "2024-01-15T10:01:00Z", "Computer": "server1", "Message": "Processing request"},
    {"TimeGenerated": "2024-01-15T10:02:00Z", "Computer": "server2", "Message": "Connection established"}
]
 
with LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=DefaultAzureCredential()
) as client:
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

Upload from JSON File

import json
 
with open("logs.json", "r") as f:
    logs = json.load(f)
 
client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

Custom Error Handling

Handle partial failures with a callback:

failed_logs = []
 
def on_error(error):
    print(f"Upload failed: {error.error}")
    failed_logs.extend(error.failed_logs)
 
client.upload(
    rule_id=rule_id,
    stream_name=stream_name,
    logs=logs,
    on_error=on_error
)
 
# Retry failed logs
if failed_logs:
    print(f"Retrying {len(failed_logs)} failed logs...")
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=failed_logs)

Ignore Errors

def ignore_errors(error):
    pass  # Silently ignore upload failures
 
client.upload(
    rule_id=rule_id,
    stream_name=stream_name,
    logs=logs,
    on_error=ignore_errors
)

Async Client

import asyncio
from azure.monitor.ingestion.aio import LogsIngestionClient
from azure.identity.aio import DefaultAzureCredential
 
async def upload_logs():
    async with LogsIngestionClient(
        endpoint=endpoint,
        credential=DefaultAzureCredential()
    ) as client:
        await client.upload(
            rule_id=rule_id,
            stream_name=stream_name,
            logs=logs
        )
 
asyncio.run(upload_logs())

Sovereign Clouds

from azure.identity import AzureAuthorityHosts, DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient
 
# Azure Government
credential = DefaultAzureCredential(authority=AzureAuthorityHosts.AZURE_GOVERNMENT)
with LogsIngestionClient(
    endpoint="https://example.ingest.monitor.azure.us",
    credential=credential,
    credential_scopes=["https://monitor.azure.us/.default"]
) as client:
    # client.upload(...)
    ...

Batching Behavior

The SDK automatically:

Splits logs into chunks of 1MB or less
Compresses each chunk with gzip
Uploads chunks in parallel

No manual batching needed for large log sets.

Client Types

Client	Purpose
`LogsIngestionClient`	Sync client for uploading logs
`LogsIngestionClient` (aio)	Async client for uploading logs

Key Concepts

Concept	Description
DCE	Data Collection Endpoint — ingestion URL
DCR	Data Collection Rule — defines schema, transformations, destination
Stream	Named data flow within a DCR
Custom Table	Target table in Log Analytics (ends with `_CL`)

DCR Stream Name Format

Stream names follow patterns:

Custom-<TableName>_CL — For custom tables
Microsoft-<TableName> — For built-in tables

Best Practices

Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.
Always use context managers for clients and async credentials. Wrap every client in with Client(...) as client: (sync) or async with Client(...) as client: (async) to ensure proper cleanup. For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
Use DefaultAzureCredential for code that runs locally. Use a specific token credential for code that runs in Azure.
Handle errors gracefully — use on_error callback for partial failures
Include TimeGenerated — Required field for all logs
Match DCR schema — Log fields must match DCR column definitions
Use async client for high-throughput scenarios
Batch uploads — SDK handles batching, but send reasonable chunks
Monitor ingestion — Check Log Analytics for ingestion status

Reference Files

File	Contents
references/capabilities.md	Additional non-hero capabilities, operation-group coverage, and production checklists.
references/non-hero-scenarios.md	Dedicated non-hero examples for secondary/advanced scenarios.