utils Module

Utilities for the Agent365 exporter.

Vendored from microsoft-agents-a365-observability-core exporters/utils.py.

Classes

A365Handlers

Processors created for Agent365 export, mirroring OtlpHandlers.

Functions

build_export_url

Construct the full export URL from endpoint and agent ID.

build_export_url(endpoint: str, agent_id: str, tenant_id: str, use_s2s_endpoint: bool = False) -> str

Parameters

Name Description
endpoint
Required
agent_id
Required
tenant_id
Required
use_s2s_endpoint
Default value: False

chunk_by_size

Split items into sub-batches whose cumulative estimated size stays under max_chunk_bytes.

Multi-item chunks are guaranteed to stay within the limit. A single item whose estimated size exceeds max_chunk_bytes forms its own one-item chunk (never silently dropped) even though that chunk exceeds the limit.

Invariants:

  • Input order is preserved across chunks.
  • Empty input produces empty output.
  • No item is ever dropped.
  • No chunk is ever empty.
chunk_by_size(items: Sequence[T], get_size: Callable[[T], int], max_chunk_bytes: int) -> list[list[T]]

Parameters

Name Description
items
Required
get_size
Required
max_chunk_bytes
Required

Exceptions

Type Description
ValueError

If max_chunk_bytes is not positive, or if get_size returns a negative value for any item.

create_a365_components

Create Agent365 span processors ready to be added to a TracerProvider.

All other configuration is read from environment variables:

  • ENABLE_A365_OBSERVABILITY_EXPORTER – must be true for the HTTP exporter

  • A365_CLUSTER_CATEGORY – defaults to "prod"

  • A365_USE_S2S_ENDPOINT – defaults to False

  • A365_SUPPRESS_INVOKE_AGENT_INPUT – defaults to False

create_a365_components(token_resolver: Callable[[str, str], str | None] | None = None, contextual_token_resolver: Callable[[TokenResolverContext], str | None] | None = None) -> A365Handlers

Parameters

Name Description
token_resolver

Optional callable (agent_id, tenant_id) -> str | None. When provided, it is used instead of the default DefaultAzureCredential resolver. This allows callers to supply FIC-based or other custom tokens.

Default value: None
contextual_token_resolver

Optional callable (TokenResolverContext) -> str | None. Provides rich context including the agentic user ID. Takes precedence over token_resolver when set.

Default value: None

estimate_span_bytes

Heuristic estimator for the serialized size of an OTLP span in HTTP JSON.

Uses generous constants tuned to over-estimate by ~25-50%, providing headroom for JSON serializer variance (whitespace, enum representation, integer-as-string).

estimate_span_bytes(span: dict[str, Any]) -> int

Parameters

Name Description
span
Required

estimate_value_bytes

Estimate the serialized byte size of a single attribute value in OTLP/HTTP JSON.

estimate_value_bytes(value: Any) -> int

Parameters

Name Description
value
Required

filter_and_partition_by_identity

Filter export-eligible spans and partition them by (tenantId, agentId).

Only spans whose gen_ai.operation.name is in GEN_AI_OPERATION_NAMES are included; non-genAI spans (e.g. HTTP, DB) and spans with other operation names are filtered out. Spans without both tenant and agent identity are also skipped.

filter_and_partition_by_identity(spans: Sequence[ReadableSpan]) -> dict[tuple[str, str], list[opentelemetry.sdk.trace.ReadableSpan]]

Parameters

Name Description
spans
Required

get_validated_domain_override

Get and validate the domain override from environment variable.

Returns the validated domain override, or None if not set or invalid.

get_validated_domain_override() -> str | None

hex_span_id

Convert a 64-bit span ID to a 16-character hex string.

hex_span_id(value: int) -> str

Parameters

Name Description
value
Required

hex_trace_id

Convert a 128-bit trace ID to a 32-character hex string.

hex_trace_id(value: int) -> str

Parameters

Name Description
value
Required

is_a365_enabled

Determine whether Agent365 export should be enabled.

is_a365_enabled(enable_a365: bool = False) -> bool

Parameters

Name Description
enable_a365
Default value: False

is_agent365_exporter_enabled

Check if Agent365 exporter is enabled via environment variable.

is_agent365_exporter_enabled() -> bool

kind_name

Return span kind name (enum name or numeric).

kind_name(kind: SpanKind) -> str

Parameters

Name Description
kind
Required

parse_retry_after

Parse the Retry-After header value.

Only numeric (seconds) values are supported. HTTP-date values are ignored.

parse_retry_after(headers: Mapping[str, str]) -> float | None

Parameters

Name Description
headers
Required

status_name

Return status code name.

status_name(code: StatusCode) -> str

Parameters

Name Description
code
Required

truncate_span

Truncate span attributes if the serialized span exceeds MAX_SPAN_SIZE_BYTES.

Removes the largest attributes first until the span fits within the limit.

truncate_span(span_dict: dict[str, Any]) -> dict[str, Any]

Parameters

Name Description
span_dict
Required