autogen_core.application.telemetry

class autogen_core.application.telemetry.EnvelopeMetadata(*, traceparent: str | None = None, tracestate: str | None = None)

Bases: object

Metadata for an envelope.

traceparent: str | None = None

tracestate: str | None = None

class autogen_core.application.telemetry.MessageRuntimeTracingConfig(runtime_name: str)

Bases: TracingConfig[Literal[‘create’, ‘send’, ‘publish’, ‘receive’, ‘intercept’, ‘process’, ‘ack’], AgentId | TopicId | str | None, ExtraMessageRuntimeAttributes]

A class that defines the configuration for message runtime instrumentation.

This class implements the TracingConfig protocol and provides the name of the module being instrumented and the attributes for the instrumentation configuration.

build_attributes(operation: Literal['create', 'send', 'publish', 'receive', 'intercept', 'process', 'ack'], destination: AgentId | TopicId | str | None, extraAttributes: ExtraMessageRuntimeAttributes | None) → Dict[str, str | bool | int | float | Sequence[str] | Sequence[bool] | Sequence[int] | Sequence[float]]

Builds the attributes for the instrumentation configuration.

Returns:

Dict[str, str] – The attributes for the instrumentation configuration.

get_span_kind(operation: Literal['create', 'send', 'publish', 'receive', 'intercept', 'process', 'ack']) → SpanKind

Determines the span kind based on the given messaging operation. Semantic Conventions - https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#span-kind

Parameters:

operation (MessagingOperation) – The messaging operation.

Returns:

SpanKind – The span kind based on the messaging operation.

get_span_name(operation: Literal['create', 'send', 'publish', 'receive', 'intercept', 'process', 'ack'], destination: AgentId | TopicId | str | None) → str

Returns the span name based on the given operation and destination. Semantic Conventions - https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#span-name

Parameters:

• operation (MessagingOperation) – The messaging operation.

• destination (Optional[MessagingDestination]) – The messaging destination.

Returns:

str – The span name.

property name: str

Returns: The name of the module that is being instrumented.

class autogen_core.application.telemetry.TraceHelper(tracer_provider: TracerProvider | None, instrumentation_builder_config: TracingConfig[Operation, Destination, ExtraAttributes])

Bases: Generic[Operation, Destination, ExtraAttributes]

TraceHelper is a utility class to assist with tracing operations using OpenTelemetry.

This class provides a context manager trace_block to create and manage spans for tracing operations, following semantic conventions and supporting nested spans through metadata contexts.

trace_block(operation: Operation, destination: Destination, parent: EnvelopeMetadata | None | Mapping[str, str], *, extraAttributes: ExtraAttributes | None = None, kind: SpanKind | None = None, attributes: Mapping[str, str | bool | int | float | Sequence[str] | Sequence[bool] | Sequence[int] | Sequence[float]] | None = None, links: Sequence[Link] | None = None, start_time: int | None = None, record_exception: bool = True, set_status_on_exception: bool = True, end_on_exit: bool = True) → Iterator[Span]

Thin wrapper on top of start_as_current_span. 1. It helps us follow semantic conventions 2. It helps us get contexts from metadata so we can get nested spans

Parameters:

• operation (MessagingOperation) – The messaging operation being performed.

• destination (MessagingDestination) – The messaging destination being used.

• Optional[TelemetryMetadataContainer] (parent) – The parent telemetry metadta context

• kind (SpanKind, optional) – The kind of span. If not provided, it maps to PRODUCER or CONSUMER depending on the operation.

• extraAttributes (ExtraAttributes, optional) – Additional defined attributes for the span. Defaults to None.

• attributes (Optional[types.Attributes], optional) – Additional non-defined attributes for the span. Defaults to None.

• links (Optional[Sequence[Link]], optional) – Links to other spans. Defaults to None.

• start_time (Optional[int], optional) – The start time of the span. Defaults to None.

• record_exception (bool, optional) – Whether to record exceptions. Defaults to True.

• set_status_on_exception (bool, optional) – Whether to set the status on exception. Defaults to True.

• end_on_exit (bool, optional) – Whether to end the span on exit. Defaults to True.

Yields:

Iterator[Span] – The span object.

autogen_core.application.telemetry.get_telemetry_envelope_metadata() → EnvelopeMetadata

Retrieves the telemetry envelope metadata.

Returns:

EnvelopeMetadata – The envelope metadata containing the traceparent and tracestate.

autogen_core.application.telemetry.get_telemetry_grpc_metadata(existingMetadata: Mapping[str, str] | None = None) → Dict[str, str]

Retrieves the telemetry gRPC metadata.

Parameters:

existingMetadata (Optional[Mapping[str, str]]) – The existing metadata to include in the gRPC metadata.

Returns:

Mapping[str, str] – The gRPC metadata containing the traceparent and tracestate.