nat.plugins.a365.telemetry.a365_exporter#

Attributes#

`Agent365Exporter`
`logger`
`_span_dump_lock`
`_DEFAULT_AGENT_NAME`
`_REQUIRED_A365_ATTRIBUTES`

Classes#

`_ReadableSpanAdapter`	Adapter that makes OtelSpan compatible with A365's ReadableSpan interface.
`A365OtelExporter`	Agent 365 exporter for AI workflow observability.

Functions#

`_warn_missing_blueprint_id_once`(→ None)
`_json_safe`(value)	Convert SDK/OpenTelemetry values into a stable JSON-safe shape.
`_span_context_to_dict`(context)
`_resource_to_dict`(resource)
`_instrumentation_scope_to_dict`(scope)
`_event_to_dict`(event)
`_link_to_dict`(link)
`_readable_span_to_dict`(span)
`_dump_readable_spans_if_configured`(→ None)
`_set_string_attr`(→ None)
`_messages_attr`(→ str \| None)
`_add_invoke_message_fallbacks`(→ None)
`_missing_required_attrs`(→ list[dict[str, Any]])
`_convert_otel_span_to_readable`(→ _ReadableSpanAdapter)	Convert an OtelSpan to a ReadableSpan-compatible adapter for A365 exporter.

Module Contents#

Agent365Exporter#

logger#

_span_dump_lock#

_DEFAULT_AGENT_NAME = 'NeMo Agent Toolkit Workflow'#

_warn_missing_blueprint_id_once() → None#

_REQUIRED_A365_ATTRIBUTES#

_json_safe(value)#: Convert SDK/OpenTelemetry values into a stable JSON-safe shape.

_span_context_to_dict(context)#

_resource_to_dict(resource)#

_instrumentation_scope_to_dict(scope)#

_event_to_dict(event)#

_link_to_dict(link)#

_readable_span_to_dict(span)#

_dump_readable_spans_if_configured( *, readable_spans, tenant_id: str, agent_id: str, use_s2s_endpoint: bool, ) → None#

_set_string_attr( attributes: dict[str, Any], key: str, value: Any, *, overwrite: bool = False, ) → None#

_messages_attr(role: str, content: Any) → str | None#

_add_invoke_message_fallbacks(attributes: dict[str, Any]) → None#

_missing_required_attrs(readable_spans) → list[dict[str, Any]]#

class _ReadableSpanAdapter( otel_span: nat.plugins.opentelemetry.otel_span.OtelSpan, tenant_id: str | None, agent_id: str | None, )#

Adapter that makes OtelSpan compatible with A365’s ReadableSpan interface.

A365’s Agent365Exporter expects ReadableSpan objects with specific attributes. This adapter wraps OtelSpan and provides the expected interface.

Initialize the adapter.

Args:: otel_span: The OtelSpan to adapt tenant_id: Fallback tenant ID (used when no per-turn identity is set) agent_id: Fallback agent ID (used when no per-turn identity is set)

context#

attributes#

events = []#

links = []#

name#

kind#

start_time#

end_time#

status#

instrumentation_scope#

resource#

_convert_otel_span_to_readable( otel_span: nat.plugins.opentelemetry.otel_span.OtelSpan, tenant_id: str | None, agent_id: str | None, ) → _ReadableSpanAdapter#

Convert an OtelSpan to a ReadableSpan-compatible adapter for A365 exporter.

A365’s Agent365Exporter expects ReadableSpan objects with specific attributes. This function creates a compatible adapter object.

Args:: otel_span: The OtelSpan to convert tenant_id: Fallback tenant ID (used when no per-turn identity is set) agent_id: Fallback agent ID (used when no per-turn identity is set)
Returns:: _ReadableSpanAdapter object that mimics ReadableSpan interface

class A365OtelExporter( agent_id: str | None, tenant_id: str | None, token_resolver: collections.abc.Callable[[str, str], str | None] | None, cluster_category: str = 'prod', use_s2s_endpoint: bool = False, suppress_invoke_agent_input: bool = False, context_state: nat.builder.context.ContextState | None = None, batch_size: int = 100, flush_interval: float = 5.0, max_queue_size: int = 1000, drop_on_overflow: bool = False, shutdown_timeout: float = 10.0, resource_attributes: dict[str, str] | None = None, token_cache=None, auth_ref=None, builder=None, token_extractor=None, )#

Bases: nat.plugins.opentelemetry.otel_span_exporter.OtelSpanExporter

Agent 365 exporter for AI workflow observability.

Integrates A365’s Agent365Exporter with NAT’s telemetry system to send OpenTelemetry spans to Microsoft Agent 365 backend endpoints.

Args:: agent_id: The Agent 365 agent ID tenant_id: The Azure tenant ID token_resolver: Callable that resolves auth token (agent_id, tenant_id) -> token cluster_category: Cluster category/environment (e.g., ‘prod’, ‘dev’) use_s2s_endpoint: Use service-to-service endpoint instead of standard endpoint suppress_invoke_agent_input: Suppress input messages for InvokeAgent spans context_state: Execution context for isolation batch_size: Batch size for exporting flush_interval: Flush interval for exporting max_queue_size: Maximum queue size for exporting drop_on_overflow: Drop on overflow for exporting shutdown_timeout: Shutdown timeout for exporting resource_attributes: Additional resource attributes for spans

Initialize the A365 exporter.

_token_extractor#

_agent_id#

_tenant_id#

_token_resolver#

_cluster_category = 'prod'#

_use_s2s_endpoint = False#

_suppress_invoke_agent_input = False#

_token_cache = None#

_auth_ref = None#

_builder = None#

_auth_providers: dict[tuple[str | None, str | None], Any]#

_auth_locks: dict[tuple[str | None, str | None], asyncio.Lock]#

_auth_locks_guard#

_a365_exporter#

async _ensure_token_for(agent_id: str, tenant_id: str) → None#

Populate or refresh the cached bearer for (agent_id, tenant_id).

Called from export_otel_spans for the identity stamped on the spans being exported. Skips the call when the cached token is still valid with a 5-minute buffer.

async export_otel_spans( spans: list[nat.plugins.opentelemetry.otel_span.OtelSpan], ) → None#: Export a list of OtelSpans using the A365 exporter.