microsoft.opentelemetry.a365.core package

class microsoft.opentelemetry.a365.core.OpenTelemetryScope(operation_name, activity_name, agent_details=None, span_details=None)[source]

Bases: object

Base class for OpenTelemetry tracing scopes in the SDK.

record_error(exception)[source]

Record an error in the span.

Parameters:: exception (Exception) – The exception that occurred
Return type:: None

record_response(response)[source]

Record an response in the span.

Parameters:: response (str) – The response content to record
Return type:: None

record_cancellation()[source]

Record task cancellation.

Return type:: None

set_tag_maybe(name, value)[source]

Set a tag on the span if the value is not None.

Parameters:

name (str) – The name of the tag
value (Any) – The value to set (will be skipped if None)

Return type:

None

record_attributes(attributes)[source]

Record multiple attribute key/value pairs for telemetry tracking.

This method allows setting multiple custom attributes on the span at once.

Parameters:: attributes (dict[str, Any] | list[tuple[str, Any]]) – Dictionary or list of tuples containing attribute key-value pairs. Keys that are None or empty will be skipped.
Return type:: None

set_end_time(end_time)[source]

Set a custom end time for the scope.

When set, dispose() will pass this value to span.end() instead of using the current wall-clock time. This is useful when the actual end time of the operation is known before the scope is disposed.

Parameters:: end_time (datetime) – The end time as a datetime object.
Return type:: None

get_context()[source]

Get the OpenTelemetry context for this scope’s span.

This method returns a Context object containing this scope’s span, which can be used to propagate trace context to child operations or downstream services.

Return type:: Context | None
Returns:: A Context containing this scope’s span, or None if telemetry is disabled or no span exists.

inject_context_to_headers()[source]

Inject this span’s trace context into W3C HTTP headers.

Returns a dictionary of headers containing traceparent and optionally tracestate that can be forwarded to downstream services or stored for later context propagation.

Return type:: dict[str, str]
Returns:: A dictionary containing W3C trace context headers. Returns an empty dictionary if telemetry is disabled or no span exists.

dispose()[source]

Dispose the scope and finalize telemetry data collection.

Return type:: None

class microsoft.opentelemetry.a365.core.ApplyGuardrailScope(details, agent_details, request=None, user_details=None, span_details=None)[source]

Bases: OpenTelemetryScope

Provides OpenTelemetry tracing scope for security guardrail evaluations.

Guardian spans SHOULD be children of the operation span they are protecting (e.g., inference or execute_tool spans). Multiple guardian spans MAY exist under a single operation span if multiple guardians are chained.

Example usage:

from microsoft.opentelemetry.a365.core import (
    ApplyGuardrailScope, GuardrailDetails, AgentDetails,
    GuardrailDecisionType, GuardrailTargetType, GuardrailRiskSeverity,
    GuardrailFinding,
)

details = GuardrailDetails(
    target_type=GuardrailTargetType.LLM_INPUT,
    decision_type=GuardrailDecisionType.ALLOW,
    guardian_name="Azure Content Safety",
)

with ApplyGuardrailScope.start(details, agent_details) as scope:
    # ... run guardrail evaluation ...
    scope.record_finding(GuardrailFinding(
        risk_category="hate_speech",
        risk_severity=GuardrailRiskSeverity.HIGH,
        risk_score=0.95,
    ))
    scope.record_decision(GuardrailDecisionType.DENY, "Blocked by policy")

static start(details, agent_details, request=None, user_details=None, span_details=None)[source]

Create and start a new scope for guardrail evaluation tracing.

Parameters:

details (GuardrailDetails) – Guardrail evaluation details (required).
agent_details (AgentDetails) – Agent identity details (required).
request (Request | None) – Optional request context (conversation ID, channel, content).
user_details (UserDetails | None) – Optional human user details.
span_details (SpanDetails | None) – Optional span configuration (parent context, timing, kind).

Return type:

ApplyGuardrailScope

Returns:

A new ApplyGuardrailScope instance.

record_decision(decision_type, reason=None)[source]

Update the guardrail decision on the span.

Use this to update the decision mid-flight if the initial decision changes after evaluation completes.

Parameters:

decision_type (str) – The decision type (use GuardrailDecisionType constants).
reason (str | None) – Optional human-readable reason for the decision.

Return type:

None

record_content_output(output_value)[source]

Record the sanitized/modified output content (opt-in).

This is an opt-in field for recording output content after guardrail processing. Only set this when content capture is explicitly enabled.

Parameters:: output_value (str) – The output content string.
Return type:: None

record_content_input(input_value)[source]

Record the input content being evaluated (opt-in).

This is an opt-in field for recording input content sent to the guardrail. Only set this when content capture is explicitly enabled.

Accepts plain strings or structured InputMessages containers. Structured messages are normalized and serialized to a JSON string before being set as an attribute.

Parameters:: input_value (Union[str, list[str], InputMessages]) – The input content as a string or InputMessagesParam.
Return type:: None

record_finding(finding)[source]

Record a security finding as a span event.

Each call adds a separate microsoft.security.finding event to the span. Multiple findings can be recorded for a single guardrail evaluation.

Parameters:: finding (GuardrailFinding) – The security finding to record.
Return type:: None

class microsoft.opentelemetry.a365.core.ExecuteToolScope(request, details, agent_details, user_details=None, span_details=None)[source]

Bases: OpenTelemetryScope

Provides OpenTelemetry tracing scope for AI tool execution operations.

static start(request, details, agent_details, user_details=None, span_details=None)[source]

Creates and starts a new scope for tool execution tracing.

Parameters:

request (Request) – Request details for the tool execution
details (ToolCallDetails) – The details of the tool call
agent_details (AgentDetails) – The details of the agent making the call
user_details (UserDetails | None) – Optional human user details
span_details (SpanDetails | None) – Optional span configuration (parent context, timing, kind)

Return type:

ExecuteToolScope

Returns:

A new ExecuteToolScope instance

record_response(result)[source]

Record the tool call result for telemetry tracking.

Per OTEL spec, the result is expected to be an object. If a string is provided, it is recorded as-is (JSON string fallback). If a dict is provided, it is serialized to JSON.

Parameters:: result (dict[str, object] | str) – Tool call result as a structured dict or JSON string
Return type:: None

class microsoft.opentelemetry.a365.core.InvokeAgentScope(request, scope_details, agent_details, caller_details=None, span_details=None)[source]

Bases: OpenTelemetryScope

Provides OpenTelemetry tracing scope for AI agent invocation operations.

static start(request, scope_details, agent_details, caller_details=None, span_details=None)[source]

Create and start a new scope for agent invocation tracing.

Parameters:

request (Request) – Request details for the invocation
scope_details (InvokeAgentScopeDetails) – Scope-level configuration (endpoint)
agent_details (AgentDetails) – The details of the agent being invoked
caller_details (CallerDetails | None) – Optional composite caller details (human user and/or calling agent for A2A scenarios)
span_details (SpanDetails | None) – Optional span configuration (parent context, timing, kind)

Return type:

InvokeAgentScope

Returns:

A new InvokeAgentScope instance

record_response(response)[source]

Record response information for telemetry tracking.

Parameters:: response (str) – The response string to record
Return type:: None

record_input_messages(messages)[source]

Record the input messages for telemetry tracking.

Accepts plain strings (auto-wrapped as OTEL ChatMessage with role user) or a structured InputMessages container.

Parameters:: messages (Union[str, list[str], InputMessages]) – List of input message strings or an InputMessages container
Return type:: None

record_output_messages(messages)[source]

Record the output messages for telemetry tracking.

Accepts plain strings (auto-wrapped as OTEL OutputMessage with role assistant) or a structured OutputMessages container.

Parameters:: messages (Union[str, list[str], OutputMessages]) – List of output message strings or an OutputMessages container
Return type:: None

class microsoft.opentelemetry.a365.core.InferenceScope(request, details, agent_details, user_details=None, span_details=None)[source]

Bases: OpenTelemetryScope

Provides OpenTelemetry tracing scope for generative AI inference operations.

static start(request, details, agent_details, user_details=None, span_details=None)[source]

Creates and starts a new scope for inference tracing.

Parameters:

request (Request) – Request details for the inference
details (InferenceCallDetails) – The details of the inference call
agent_details (AgentDetails) – The details of the agent making the call
user_details (UserDetails | None) – Optional human user details
span_details (SpanDetails | None) – Optional span configuration (parent context, timing)

Return type:

InferenceScope

Returns:

A new InferenceScope instance

record_input_messages(messages)[source]

Records the input messages for telemetry tracking.

Accepts plain strings (auto-wrapped as OTEL ChatMessage with role user) or a structured InputMessages container.

Parameters:: messages (Union[str, list[str], InputMessages]) – List of input message strings or an InputMessages container
Return type:: None

record_output_messages(messages)[source]

Records the output messages for telemetry tracking.

Accepts plain strings (auto-wrapped as OTEL OutputMessage with role assistant) or a structured OutputMessages container.

Parameters:: messages (Union[str, list[str], OutputMessages]) – List of output message strings or an OutputMessages container
Return type:: None

record_input_tokens(input_tokens)[source]

Records the number of input tokens for telemetry tracking.

Parameters:: input_tokens (int) – Number of input tokens
Return type:: None

record_output_tokens(output_tokens)[source]

Records the number of output tokens for telemetry tracking.

Parameters:: output_tokens (int) – Number of output tokens
Return type:: None

record_finish_reasons(finish_reasons)[source]

Records the finish reasons for telemetry tracking.

Parameters:: finish_reasons (List[str]) – List of finish reasons
Return type:: None

record_thought_process(thought_process)[source]

Records the thought process.

Parameters:: thought_process (str) – The thought process to record
Return type:: None

class microsoft.opentelemetry.a365.core.OutputScope(request, response, agent_details, user_details=None, span_details=None)[source]

Bases: OpenTelemetryScope

Provides OpenTelemetry tracing scope for output messages.

Output messages are set once (via the constructor or record_output_messages) rather than accumulated. For streaming scenarios, the agent developer should collect all output (e.g. via a list or string builder) and pass the final result to OutputScope.

static start(request, response, agent_details, user_details=None, span_details=None)[source]

Creates and starts a new scope for output tracing.

Parameters:

request (Request) – Request details for the output
response (Response) – The response details from the agent
agent_details (AgentDetails) – The details of the agent
user_details (UserDetails | None) – Optional human user details
span_details (SpanDetails | None) – Optional span configuration (parent context, timing)

Return type:

OutputScope

Returns:

A new OutputScope instance

record_output_messages(messages)[source]

Records the output messages for telemetry tracking.

Overwrites any previously set output messages. Accepts a single string, a list of strings (auto-wrapped as OTEL OutputMessage), a structured OutputMessages container, or a dict[str, object] for tool call results (per OTEL spec).

Parameters:: messages (Union[str, list[str], OutputMessages, dict[str, object]]) – String(s), OutputMessages, or dict for tool call results
Return type:: None

class microsoft.opentelemetry.a365.core.GuardrailDecisionType[source]

Bases: object

Well-known guardrail decision type values.

ALLOW = 'allow'

AUDIT = 'audit'

DENY = 'deny'

MODIFY = 'modify'

WARN = 'warn'

class microsoft.opentelemetry.a365.core.GuardrailDetails(target_type, decision_type, guardian_name=None, guardian_id=None, guardian_provider_name=None, guardian_version=None, target_id=None, decision_reason=None, decision_code=None, policy_id=None, policy_name=None, policy_version=None, content_input_hash=None, content_modified=None, external_event_id=None)[source]

Bases: object

Immutable input contract describing a guardrail evaluation.

Variables:

target_type – What content is being guarded (e.g., “llm_input”, “tool_call”).
decision_type – The guardian’s decision (e.g., “allow”, “deny”).
guardian_name – Human-readable guardian name.
guardian_id – Unique guardian identifier.
guardian_provider_name – Provider name (e.g., “azure.ai.content_safety”).
guardian_version – Guardian version string.
target_id – ID of the targeted content.
decision_reason – Human-readable decision reason.
decision_code – Machine-readable decision code.
policy_id – Triggered policy ID.
policy_name – Triggered policy name.
policy_version – Policy version.
content_input_hash – Hash of input content for forensic correlation.
content_modified – Whether the content was altered by the guardrail.
external_event_id – External event ID for SIEM correlation.

target_type: str

decision_type: str

guardian_name: str | None = None

guardian_id: str | None = None

guardian_provider_name: str | None = None

guardian_version: str | None = None

target_id: str | None = None

decision_reason: str | None = None

decision_code: str | None = None

policy_id: str | None = None

policy_name: str | None = None

policy_version: str | None = None

content_input_hash: str | None = None

content_modified: bool | None = None

external_event_id: str | None = None

class microsoft.opentelemetry.a365.core.GuardrailFinding(risk_category, risk_severity, policy_decision_type=None, policy_id=None, policy_name=None, policy_version=None, risk_score=None, risk_metadata=None)[source]

Bases: object

A single security finding from a guardrail evaluation.

Variables:

risk_category – Category of risk detected (e.g., “hate_speech”, “pii”, “jailbreak”).
risk_severity – Severity level (use GuardrailRiskSeverity constants).
policy_decision_type – Per-finding decision override.
policy_id – Policy that triggered this finding.
policy_name – Policy name.
policy_version – Policy version.
risk_score – Confidence score from 0.0 to 1.0.
risk_metadata – Non-PII structural metadata about the finding.

risk_category: str

risk_severity: str

policy_decision_type: str | None = None

policy_id: str | None = None

policy_name: str | None = None

policy_version: str | None = None

risk_score: float | None = None

risk_metadata: list[str] | None = None

class microsoft.opentelemetry.a365.core.GuardrailRiskSeverity[source]

Bases: object

Well-known risk severity level values.

NONE = 'none'

LOW = 'low'

MEDIUM = 'medium'

HIGH = 'high'

CRITICAL = 'critical'

class microsoft.opentelemetry.a365.core.GuardrailTargetType[source]

Bases: object

Well-known guardrail target type values.

Users may also supply custom string values not listed here.

LLM_INPUT = 'llm_input'

LLM_OUTPUT = 'llm_output'

TOOL_CALL = 'tool_call'

TOOL_DEFINITION = 'tool_definition'

MEMORY_STORE = 'memory_store'

MEMORY_RETRIEVE = 'memory_retrieve'

KNOWLEDGE_QUERY = 'knowledge_query'

KNOWLEDGE_RESULT = 'knowledge_result'

MESSAGE = 'message'

class microsoft.opentelemetry.a365.core.BaggageBuilder[source]

Bases: object

Per request baggage builder.

This class provides a fluent API for setting baggage values that will be propagated in the OpenTelemetry context.

Example

builder = (BaggageBuilder()
           .tenant_id("tenant-123")
           .agent_id("agent-456"))

with builder.build():
    # Baggage is set in this context
    pass
# Baggage is restored after exiting the context

operation_source(value)[source]

Set the operation source baggage value.

This captures the name of the service using the SDK.

Parameters:: value (str | None) – The service name (e.g., “my-agent-service”, “weather-bot”)
Return type:: BaggageBuilder
Returns:: Self for method chaining

tenant_id(value)[source]

Set the tenant ID baggage value.

Parameters:: value (str | None) – The tenant ID
Return type:: BaggageBuilder
Returns:: Self for method chaining

agent_id(value)[source]

Set the agent ID baggage value.

Parameters:: value (str | None) – The agent ID
Return type:: BaggageBuilder
Returns:: Self for method chaining

agentic_user_id(value)[source]

Set the agentic user ID baggage value.

Parameters:: value (str | None) – The agentic user ID
Return type:: BaggageBuilder
Returns:: Self for method chaining

agentic_user_email(value)[source]

Set the agentic user email baggage value.

Parameters:: value (str | None) – The agentic user email
Return type:: BaggageBuilder
Returns:: Self for method chaining

agent_blueprint_id(value)[source]

Set the agent blueprint ID baggage value.

Parameters:: value (str | None) – The agent blueprint ID
Return type:: BaggageBuilder
Returns:: Self for method chaining

user_id(value)[source]

Set the user ID baggage value.

Parameters:: value (str | None) – The user ID
Return type:: BaggageBuilder
Returns:: Self for method chaining

agent_name(value)[source]

Set the agent name baggage value.

Return type:: BaggageBuilder

agent_description(value)[source]

Set the agent description baggage value.

Return type:: BaggageBuilder

agent_version(value)[source]

Set the agent version baggage value.

Return type:: BaggageBuilder

user_name(value)[source]

Set the user name baggage value.

Return type:: BaggageBuilder

user_email(value)[source]

Set the user email baggage value.

Return type:: BaggageBuilder

user_client_ip(value)[source]

Set the user client IP baggage value.

Return type:: BaggageBuilder

invoke_agent_server(address, port=None)[source]

Set the invoke agent server address and port baggage values.

Parameters:

address (str | None) – The server address (hostname) of the target agent service.
port (int | None) – Optional server port. Only recorded when different from 443.

Return type:

BaggageBuilder

Returns:

Self for method chaining

conversation_id(value)[source]

Set the conversation ID baggage value.

Return type:: BaggageBuilder

conversation_item_link(value)[source]

Set the conversation item link baggage value.

Return type:: BaggageBuilder

session_id(value)[source]

Set the session ID baggage value.

Return type:: BaggageBuilder

session_description(value)[source]

Set the session description baggage value.

Return type:: BaggageBuilder

channel_name(value)[source]

Sets the channel name baggage value (e.g., ‘Teams’, ‘msteams’).

Return type:: BaggageBuilder

channel_links(value)[source]

Sets the channel link baggage value.

Return type:: BaggageBuilder

set_pairs(pairs)[source]

Accept dict or iterable of (k,v).

Return type:: BaggageBuilder

build()[source]

Apply the collected baggage to the current context.

Return type:: BaggageScope
Returns:: A context manager that restores the previous baggage on exit

class microsoft.opentelemetry.a365.core.InvokeAgentScopeDetails(endpoint=None)[source]

Bases: object

Scope-level configuration for agent invocation tracing.

endpoint: ServiceEndpoint | None = None

class microsoft.opentelemetry.a365.core.AgentDetails(agent_id, agent_name=None, agent_description=None, agentic_user_id=None, agentic_user_email=None, agent_blueprint_id=None, agent_platform_id=None, tenant_id=None, icon_uri=None, provider_name=None, agent_version=None)[source]

Bases: object

Details about an AI agent in the system.

agent_id: str: The unique identifier for the AI agent.

agent_name: str | None = None: The human-readable name of the AI agent.

agent_description: str | None = None: A description of the AI agent’s purpose or capabilities.

agentic_user_id: str | None = None: Agentic User ID for the agent.

agentic_user_email: str | None = None: Email address for the agentic user.

agent_blueprint_id: str | None = None: Blueprint/Application ID for the agent.

agent_platform_id: str | None = None: Platform ID for the agent.

tenant_id: str | None = None: Tenant ID for the agent.

icon_uri: str | None = None: Optional icon URI for the agent.

provider_name: str | None = None: The provider name (e.g., openai, anthropic).

agent_version: str | None = None: Optional version of the agent (e.g., “1.0.0”, “2025-05-01”).

class microsoft.opentelemetry.a365.core.CallerDetails(user_details=None, caller_agent_details=None)[source]

Bases: object

Composite caller details for agent-to-agent (A2A) scenarios.

Groups the human caller identity and the calling agent identity together.

user_details: UserDetails | None = None: Details about the human user in the call chain.

caller_agent_details: AgentDetails | None = None: Details about the calling agent in A2A scenarios.

class microsoft.opentelemetry.a365.core.UserDetails(user_id=None, user_email=None, user_name=None, user_client_ip=None)[source]

Bases: object

Details about the human user that invoked an agent.

user_id: str | None = None: The unique identifier for the user.

user_email: str | None = None: The email address of the user.

user_name: str | None = None: The human-readable name of the user.

user_client_ip: str | None = None: The client IP address of the user.

class microsoft.opentelemetry.a365.core.ToolCallDetails(tool_name, arguments=None, tool_call_id=None, description=None, tool_type=None, endpoint=None)[source]

Bases: object

Details of a tool call made by an agent in the system.

tool_name: str

arguments: dict[str, object] | str | None = None

tool_call_id: str | None = None

description: str | None = None

tool_type: str | None = None

endpoint: ServiceEndpoint | None = None

class microsoft.opentelemetry.a365.core.Channel(name=None, link=None)[source]

Bases: object

Channel information for agent execution context.

name: str | None = None

link: str | None = None

class microsoft.opentelemetry.a365.core.Request(content=None, session_id=None, channel=None, conversation_id=None)[source]

Bases: object

Request details for agent execution.

content: str | list[str] | InputMessages | None = None

session_id: str | None = None

channel: Channel | None = None

conversation_id: str | None = None

class microsoft.opentelemetry.a365.core.Response(messages)[source]

Bases: object

Response details from agent execution.

Accepts plain strings (backward compat), structured OutputMessages, or a dict for tool call results (per OTEL spec).

messages: str | list[str] | OutputMessages | dict[str, object]

class microsoft.opentelemetry.a365.core.SpanDetails(span_kind=None, parent_context=None, start_time=None, end_time=None, span_links=None)[source]

Bases: object

Groups span configuration for scope construction.

span_kind: SpanKind | None = None: Optional span kind override.

parent_context: Context | None = None: Optional OpenTelemetry Context used to link this span to an upstream operation.

start_time: datetime | None = None: Optional explicit start time as a datetime object.

end_time: datetime | None = None: Optional explicit end time as a datetime object.

span_links: list[Link] | None = None: Optional span links to associate with this span for causal relationships.

class microsoft.opentelemetry.a365.core.InferenceCallDetails(operationName, model, providerName, inputTokens=None, outputTokens=None, finishReasons=None, thoughtProcess=None, endpoint=None)[source]