diff --git a/README.md b/README.md index 3e486cdb..b0601355 100644 --- a/README.md +++ b/README.md @@ -348,6 +348,7 @@ remain enabled by default. |---|---|---| | [samples/a365/exporter.py](https://github.com/microsoft/opentelemetry-distro-python/blob/main/samples/a365/exporter.py) | A365 | LangChain with A365 auto-instrumentation | | [samples/a365/manual_telemetry.py](https://github.com/microsoft/opentelemetry-distro-python/blob/main/samples/a365/manual_telemetry.py) | A365 | Manual instrumentation using all scope classes | +| [samples/a365/s2s/s2s_exporter.py](https://github.com/microsoft/opentelemetry-distro-python/blob/main/samples/a365/s2s/s2s_exporter.py) | A365 | Store-ready S2S export with all manual observability scopes | | [samples/distro/tracing.py](https://github.com/microsoft/opentelemetry-distro-python/blob/main/samples/distro/tracing.py) | Azure Monitor | Basic tracing | | [samples/distro/metrics.py](https://github.com/microsoft/opentelemetry-distro-python/blob/main/samples/distro/metrics.py) | Azure Monitor | Metrics collection | | [samples/distro/logging_sample.py](https://github.com/microsoft/opentelemetry-distro-python/blob/main/samples/distro/logging_sample.py) | Azure Monitor | Log export | diff --git a/samples/a365/s2s/README.md b/samples/a365/s2s/README.md new file mode 100644 index 00000000..13a011b4 --- /dev/null +++ b/samples/a365/s2s/README.md @@ -0,0 +1,109 @@ +# A365 S2S Exporter Sample + +This sample exports [Agent 365](https://learn.microsoft.com/en-us/microsoft-agent-365/) +telemetry through the service-to-service (S2S) endpoint and demonstrates every +public manual observability scope. + +The service authenticates on its own behalf. The token resolver uses the +Blueprint application credentials to obtain an agent-instance token and then an +application token for the A365 observability scope. It deliberately omits the +agentic-user FIC step and does not emit `microsoft.agent.user.id` or +`microsoft.agent.user.email`. + +## Sample organization + +| File | Responsibility | +| --- | --- | +| `s2s_exporter.py` | Executable entry point that loads configuration, enables S2S export, and runs the scenario | +| `sample_config.py` | Environment validation and deterministic agent, caller, and request inputs | +| `token_resolver.py` | Blueprint-to-agent token exchange, observability token acquisition, and token caching | +| `sample_scenario.py` | Deterministic manual-scope telemetry used for Store validation | + +## Prerequisites + +- Python 3.10+ +- [uv](https://docs.astral.sh/uv/) +- A Blueprint app registration granted the + `Agent365.Observability.OtelWrite` **application** permission with admin + consent. See [`MIGRATION_A365.md`](../../../MIGRATION_A365.md) under + "Troubleshooting - Permissions and Setup". +- Real tenant, Blueprint app client ID, agent app instance client ID, and human caller + values from the deployment being validated. Angle-bracket placeholders are + rejected at startup. + +## Configure and run + +PowerShell: + +```powershell +$env:ENABLE_OBSERVABILITY = "true" +$env:ENABLE_A365_OBSERVABILITY_EXPORTER = "true" +$env:CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID = "" +$env:CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET = "" +$env:CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID = "" +$env:A365_AGENT_APP_INSTANCE_ID = "" +$env:A365_CALLER_USER_ID = "" +$env:A365_CALLER_USER_EMAIL = "" +$env:A365_CALLER_CLIENT_IP = "" + +uv run --with msal python samples\a365\s2s\s2s_exporter.py +``` + +Bash: + +```bash +export ENABLE_OBSERVABILITY=true +export ENABLE_A365_OBSERVABILITY_EXPORTER=true +export CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID="" +export CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET="" +export CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID="" +export A365_AGENT_APP_INSTANCE_ID="" +export A365_CALLER_USER_ID="" +export A365_CALLER_USER_EMAIL="" +export A365_CALLER_CLIENT_IP="" + +uv run --with msal python samples/a365/s2s/s2s_exporter.py +``` + +`a365_use_s2s_endpoint=True` routes exports to the S2S ingest endpoint. The +tenant and agent ID in each export request must match the configured tenant and +agent app instance client ID; the resolver rejects mismatches before token +acquisition. `gen_ai.agent.id` and the `{agentId}` export URL segment therefore +use `A365_AGENT_APP_INSTANCE_ID`, not the Blueprint client ID. +The Blueprint telemetry attribute uses the same +`CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID` value used to authenticate +the Blueprint application. + +## Scope and Store-validation coverage + +| Scope | Sample behavior | Store validation | +| --- | --- | --- | +| `InvokeAgentScope` | Root request, agent/Blueprint/caller identity, endpoint, input, and final output | Required | +| `InferenceScope` | Model/provider, messages, token usage, and finish reason | Required | +| `ExecuteToolScope` | Tool identity, arguments, and result | Required | +| `OutputScope` | Child span representing asynchronous/final output | Validate before publishing | +| `ApplyGuardrailScope` | Input-safety decision and finding event | Additional security telemetry | + +The sample populates the publishing attributes documented in +[Validate for Store publishing](https://learn.microsoft.com/en-us/microsoft-agent-365/developer/observability?tabs=python#validate-for-store-publishing), +including the tenant, agent, Blueprint, human caller, client address, channel, +conversation, operation, message, endpoint, model, and tool fields applicable +to each span. Human caller identity uses the standard `user.id` and +`user.email` attributes. S2S agentic-user attributes +`microsoft.agent.user.id` and `microsoft.agent.user.email` are intentionally +absent. + +## Verify export + +The sample enables DEBUG logging for the A365 exporter. A successful run reports +the HTTP status and correlation ID on stderr: + +```text +DEBUG ...agent365_exporter: HTTP 200 success on attempt 1. Correlation ID: . Response: ... +``` + +HTTP 401 usually indicates an invalid token audience or tenant. HTTP 403 +usually indicates missing `Agent365.Observability.OtelWrite` application +permission, missing admin consent, or an agent/Blueprint identity that is not +onboarded for the tenant. Use the logged correlation ID when investigating +either response. diff --git a/samples/a365/s2s/s2s_exporter.py b/samples/a365/s2s/s2s_exporter.py new file mode 100644 index 00000000..8eb7c5f1 --- /dev/null +++ b/samples/a365/s2s/s2s_exporter.py @@ -0,0 +1,63 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Sample: A365 Exporter with S2S (Service-to-Service) authentication + +Demonstrates how to export A365 telemetry using the S2S endpoint with a +service-principal token resolver. The resolver performs the app-to-instance +exchange and acquires an application token for the A365 observability scope; +the deterministic scenario exercises every public manual observability scope. + +Run this file directly from the repository root as documented in +samples/a365/s2s/README.md. +""" + +import logging + +from microsoft.opentelemetry import use_microsoft_opentelemetry + +from sample_config import SampleConfig +from sample_scenario import emit_sample_telemetry +from token_resolver import build_s2s_token_resolver + + +def _configure_export_logging() -> None: + """Surface exporter status and correlation IDs without duplicate handlers.""" + exporter_logger = logging.getLogger("microsoft.opentelemetry.a365.core.exporters.agent365_exporter") + exporter_logger.setLevel(logging.DEBUG) + exporter_logger.propagate = False + handler_name = "a365-s2s-sample-export-logging" + if any(getattr(handler, "name", None) == handler_name for handler in exporter_logger.handlers): + return + handler = logging.StreamHandler() + handler.name = handler_name + handler.setFormatter(logging.Formatter("%(levelname)s %(name)s: %(message)s")) + exporter_logger.addHandler(handler) + + +def main() -> None: + config = SampleConfig.load() + _configure_export_logging() + use_microsoft_opentelemetry( + enable_a365=True, + a365_use_s2s_endpoint=True, + a365_token_resolver=build_s2s_token_resolver(config), + ) + print("Telemetry configured for S2S export.\n") + + emit_sample_telemetry( + config.create_agent_details(), + config.create_user_details(), + config.create_request(), + ) + + print( + "\nDone. All spans have been recorded. They are flushed to the A365 " + "batch span processor and exported on shutdown when " + "ENABLE_A365_OBSERVABILITY_EXPORTER=true." + ) + + +if __name__ == "__main__": + main() diff --git a/samples/a365/s2s/sample_config.py b/samples/a365/s2s/sample_config.py new file mode 100644 index 00000000..41771080 --- /dev/null +++ b/samples/a365/s2s/sample_config.py @@ -0,0 +1,83 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Validated configuration and deterministic inputs for the A365 S2S sample.""" + +import os +from dataclasses import dataclass, field + +from microsoft.opentelemetry.a365.core import AgentDetails, Channel, Request, UserDetails + +A365_SERVICE_CLIENT_ID_ENV = "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID" +A365_SERVICE_CLIENT_SECRET_ENV = "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET" +A365_SERVICE_TENANT_ID_ENV = "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID" +A365_AGENT_APP_INSTANCE_ID_ENV = "A365_AGENT_APP_INSTANCE_ID" +A365_CALLER_USER_ID_ENV = "A365_CALLER_USER_ID" +A365_CALLER_USER_EMAIL_ENV = "A365_CALLER_USER_EMAIL" +A365_CALLER_CLIENT_IP_ENV = "A365_CALLER_CLIENT_IP" + + +def _require_env(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value or (value.startswith("<") and value.endswith(">")): + raise SystemExit( + f"Environment variable {name} is not set. Set the required shell variables " + "and run the sample as described in samples/a365/s2s/README.md." + ) + return value + + +@dataclass(frozen=True) +class SampleConfig: + """Validated environment settings for the A365 S2S sample.""" + + client_id: str + client_secret: str = field(repr=False) + tenant_id: str + agent_instance_id: str + caller_user_id: str + caller_user_email: str + caller_client_ip: str + + @classmethod + def load(cls) -> "SampleConfig": + """Load all required settings from the environment.""" + return cls( + client_id=_require_env(A365_SERVICE_CLIENT_ID_ENV), + client_secret=_require_env(A365_SERVICE_CLIENT_SECRET_ENV), + tenant_id=_require_env(A365_SERVICE_TENANT_ID_ENV), + agent_instance_id=_require_env(A365_AGENT_APP_INSTANCE_ID_ENV), + caller_user_id=_require_env(A365_CALLER_USER_ID_ENV), + caller_user_email=_require_env(A365_CALLER_USER_EMAIL_ENV), + caller_client_ip=_require_env(A365_CALLER_CLIENT_IP_ENV), + ) + + def create_agent_details(self) -> AgentDetails: + """Create the deterministic agent identity used by the sample.""" + return AgentDetails( + agent_id=self.agent_instance_id, + agent_name="Weather Agent", + agent_description="Answers weather-related questions", + agent_blueprint_id=self.client_id, + tenant_id=self.tenant_id, + provider_name="azure-openai", + agent_version="1.0.0", + ) + + def create_user_details(self) -> UserDetails: + """Create the deterministic human caller used by the sample.""" + return UserDetails( + user_id=self.caller_user_id, + user_email=self.caller_user_email, + user_name="Sample Caller", + user_client_ip=self.caller_client_ip, + ) + + def create_request(self) -> Request: + """Create the deterministic request used by the sample.""" + return Request( + content="What's the weather in Seattle?", + session_id="session-s2s-123", + channel=Channel(name="service", link="https://contoso.example/a365-s2s"), + conversation_id="conv-s2s-789", + ) diff --git a/samples/a365/s2s/sample_scenario.py b/samples/a365/s2s/sample_scenario.py new file mode 100644 index 00000000..a9b2e838 --- /dev/null +++ b/samples/a365/s2s/sample_scenario.py @@ -0,0 +1,201 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Deterministic telemetry scenario for the A365 S2S sample.""" + +from microsoft.opentelemetry.a365.core import ( + AgentDetails, + ApplyGuardrailScope, + BaggageBuilder, + CallerDetails, + ChatMessage, + ExecuteToolScope, + GuardrailDecisionType, + GuardrailDetails, + GuardrailFinding, + GuardrailRiskSeverity, + GuardrailTargetType, + InferenceCallDetails, + InferenceOperationType, + InferenceScope, + InputMessages, + InvokeAgentScope, + InvokeAgentScopeDetails, + MessageRole, + OutputScope, + OutputMessage, + OutputMessages, + Request, + Response, + ServiceEndpoint, + SpanDetails, + TextPart, + ToolCallDetails, + ToolType, + UserDetails, +) + + +def emit_sample_telemetry( + agent_details: AgentDetails, + user_details: UserDetails, + request: Request, +) -> str: + """Emit deterministic Store-validation telemetry without network calls.""" + user_question = request.content + if not isinstance(user_question, str): + raise ValueError("The S2S sample request content must be a string.") + final_answer = "It's currently 62°F and partly cloudy in Seattle." + invoke_context = None + baggage = ( + BaggageBuilder() + .tenant_id(agent_details.tenant_id) + .agent_id(agent_details.agent_id) + .agent_blueprint_id(agent_details.agent_blueprint_id) + .agent_name(agent_details.agent_name) + .agent_description(agent_details.agent_description) + .agent_version(agent_details.agent_version) + .user_id(user_details.user_id) + .user_email(user_details.user_email) + .user_name(user_details.user_name) + .user_client_ip(user_details.user_client_ip) + .channel_name(request.channel.name if request.channel else None) + .channel_links(request.channel.link if request.channel else None) + .session_id(request.session_id) + .conversation_id(request.conversation_id) + .invoke_agent_server("weather-agent.contoso.com", 8443) + ) + + with baggage.build(): + with InvokeAgentScope.start( + request=request, + scope_details=InvokeAgentScopeDetails( + endpoint=ServiceEndpoint(hostname="weather-agent.contoso.com", port=8443), + ), + agent_details=agent_details, + caller_details=CallerDetails(user_details=user_details), + ) as invoke_scope: + invoke_scope.record_input_messages( + InputMessages( + messages=[ + ChatMessage( + role=MessageRole.USER, + parts=[TextPart(content=user_question)], + ), + ] + ) + ) + + with ApplyGuardrailScope.start( + details=GuardrailDetails( + target_type=GuardrailTargetType.LLM_INPUT, + decision_type=GuardrailDecisionType.ALLOW, + guardian_name="Sample Content Safety", + guardian_id="sample-content-safety", + guardian_provider_name="contoso.security", + guardian_version="1.0", + target_id="prompt-123", + decision_reason="No unsafe content detected", + decision_code="allowed", + policy_id="policy-123", + policy_name="Default Prompt Safety", + policy_version="1.0", + content_modified=False, + ), + agent_details=agent_details, + request=request, + user_details=user_details, + ) as guardrail_scope: + guardrail_scope.record_content_input(user_question) + guardrail_scope.record_finding( + GuardrailFinding( + risk_category="unsafe_content", + risk_severity=GuardrailRiskSeverity.NONE, + risk_score=0.0, + policy_decision_type=GuardrailDecisionType.ALLOW, + policy_id="policy-123", + policy_name="Default Prompt Safety", + policy_version="1.0", + ) + ) + + with InferenceScope.start( + request=request, + details=InferenceCallDetails( + operationName=InferenceOperationType.CHAT, + model="gpt-4o", + providerName="azure-openai", + endpoint=ServiceEndpoint(hostname="example.openai.azure.com", port=443), + ), + agent_details=agent_details, + user_details=user_details, + ) as inference_scope: + inference_scope.record_input_messages( + InputMessages( + messages=[ + ChatMessage( + role=MessageRole.SYSTEM, + parts=[TextPart(content="You are a helpful weather assistant.")], + ), + ChatMessage( + role=MessageRole.USER, + parts=[TextPart(content=user_question)], + ), + ] + ) + ) + inference_scope.record_input_tokens(45) + inference_scope.record_output_tokens(12) + inference_scope.record_finish_reasons(["tool_call"]) + inference_scope.record_output_messages( + OutputMessages( + messages=[ + OutputMessage( + role=MessageRole.ASSISTANT, + parts=[TextPart(content="I'll look up the weather for Seattle.")], + finish_reason="tool_call", + ), + ] + ) + ) + + with ExecuteToolScope.start( + request=request, + details=ToolCallDetails( + tool_name="get_weather", + arguments={"city": "Seattle", "units": "fahrenheit"}, + tool_call_id="call-123", + description="Fetches current weather for a city", + tool_type=ToolType.FUNCTION.value, + endpoint=ServiceEndpoint(hostname="weather-api.contoso.com", port=443), + ), + agent_details=agent_details, + user_details=user_details, + ) as tool_scope: + tool_scope.record_response('{"temperature":62,"condition":"Partly cloudy"}') + + invoke_scope.record_output_messages( + OutputMessages( + messages=[ + OutputMessage( + role=MessageRole.ASSISTANT, + parts=[TextPart(content=final_answer)], + finish_reason="stop", + ), + ] + ) + ) + invoke_context = invoke_scope.get_context() + + assert invoke_context is not None + with OutputScope.start( + request=request, + response=Response(messages=final_answer), + agent_details=agent_details, + user_details=user_details, + span_details=SpanDetails(parent_context=invoke_context), + ) as output_scope: + if request.channel: + output_scope.set_tag_maybe("microsoft.channel.name", request.channel.name) + + return final_answer diff --git a/samples/a365/s2s/token_resolver.py b/samples/a365/s2s/token_resolver.py new file mode 100644 index 00000000..1afbeaf8 --- /dev/null +++ b/samples/a365/s2s/token_resolver.py @@ -0,0 +1,105 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""S2S token resolution for the A365 exporter sample.""" + +import threading +import time +from collections.abc import Callable +from typing import Optional + +from sample_config import SampleConfig + +# Azure AD requires the resource's ``/.default`` scope rather than a specific +# delegated scope like ``Agent365.Observability.OtelWrite`` (which is only valid +# in the FIC ``user_fic`` grant used by the AI-teammate flow). +A365_OBSERVABILITY_SCOPE = "api://9b975845-388f-4429-889e-eab1ef63949c/.default" + + +def build_s2s_token_resolver(config: SampleConfig) -> Callable[[str, str], Optional[str]]: + """Build the app-only S2S token resolver for the configured agent. + + Returns a ``(agent_id, tenant_id) -> token | None`` callable. + """ + try: + import msal + except ImportError as exc: + raise SystemExit( + "msal is required for the S2S sample. Run `uv run --with msal python " + "samples/a365/s2s/s2s_exporter.py` from the repository root as described " + "in samples/a365/s2s/README.md." + ) from exc + + cache: dict[str, tuple[str, float]] = {} + lock = threading.Lock() + + def resolve(agent_id: str, request_tenant_id: str) -> Optional[str]: # pylint: disable=too-many-return-statements + cache_key = f"{request_tenant_id}:{agent_id}" + + if request_tenant_id != config.tenant_id: + print( + f"S2S token acquisition failed: request tenant {request_tenant_id!r} " + f"does not match configured tenant {config.tenant_id!r}." + ) + return None + if agent_id != config.agent_instance_id: + print( + f"S2S token acquisition failed: request agent {agent_id!r} " + f"does not match configured agent instance {config.agent_instance_id!r}." + ) + return None + authority = f"https://login.microsoftonline.com/{request_tenant_id}" + + with lock: + cached = cache.get(cache_key) + if cached is not None: + token, expires_at = cached + if time.time() < expires_at - 60: + return token + + try: + # Step 1: Agent application token via fmi_path. + app = msal.ConfidentialClientApplication( + client_id=config.client_id, + client_credential=config.client_secret, + authority=authority, + ) + result = app.acquire_token_for_client( + scopes=["api://AzureAdTokenExchange/.default"], + fmi_path=config.agent_instance_id, + ) + if "access_token" not in result: + print(f"S2S step 1 (app token) failed: {result.get('error_description', result)}") + return None + agent_token = result["access_token"] + + # Step 2: Instance app authenticated with the agent token as a + # client assertion. (No agentic-user / user_fic step in S2S.) + # Pass the assertion as a no-arg callable (MSAL's recommended form) + # so it can be re-read on demand instead of as a static string. + instance_app = msal.ConfidentialClientApplication( + client_id=config.agent_instance_id, + client_credential={"client_assertion": lambda: agent_token}, + authority=authority, + ) + + # Step 3: Application token for the A365 observability scope. + result = instance_app.acquire_token_for_client( + scopes=[A365_OBSERVABILITY_SCOPE], + ) + if "access_token" not in result: + print("S2S step 3 (observability token) failed: " f"{result.get('error_description', result)}") + return None + + access_token = result["access_token"] + expires_in = result.get("expires_in", 3600) + + with lock: + cache[cache_key] = (access_token, time.time() + expires_in) + return access_token + + except Exception as exc: # pylint: disable=broad-exception-caught + print(f"S2S token acquisition failed: {exc}") + return None + + return resolve