Agent 365 导出器在导出遥测数据时需要一个令牌解析器来进行身份验证。 本指南介绍如何为使用 Microsoft 365 智能体 SDK 生成的智能体进行设置，涵盖启用了 Agent 365 的智能体以及跨 .NET、Python 和 Node.js 的自定义引擎智能体。
关于 Distro 安装、一般配置和非智能体 SDK 应用场景，请参阅 Microsoft OpenTelemetry Distro。
概述
根据智能体类型及其获取令牌的方式，有四种身份验证应用场景。 令牌获取可使用代表流 (OBO) 或服务对服务 (S2S)。 请选择与您的设置相匹配的应用场景：
| 情况 | 描述 |
|---|---|
| 使用 OBO 启用的 Agent 365 | Distro 的内置 AgenticTokenCache 会自动处理令牌获取。 不需要自定义解析器。 这是推荐用于启用了 Agent 365 的智能体的方法。 |
| 使用 S2S 启用的 Agent 365 | 智能体通过使用智能体标识链（getAgenticApplicationToken + Microsoft 身份验证库 (MSAL)）获取令牌。 需要自定义 TokenResolver。 在 OBO 不可用或需要仅应用令牌的情况下，使用此方法。 |
| 使用 OBO 的自定义引擎 | 智能体通过 Azure 机器人 OAuth 获取用户令牌，该令牌限定于可观测性 API。 需要自定义 TokenResolver 和 Azure 机器人 OAuth 连接。 |
| 使用 S2S 的自定义引擎 | 智能体使用客户端凭据获取仅限应用的令牌。 需要自定义 TokenResolver。 应用注册必须是标准（非智能体）应用。 |
通过 OBO 启用 Agent 365
启用 Agent 365 的智能体会从 Agent 365 平台接收带有智能体身份 (agenticAppId, agenticUserId) 的请求。 使用 OBO 时，Distro 的内置 AgenticTokenCache 会自动处理令牌获取：无需自定义令牌解析器。
必备条件
- Entra 应用注册：一个包含客户端 ID、客户端密钥和租户 ID 的服务主体（应用注册）
-
委派的 API 权限：添加
Agent365.Observability.OtelWrite（已委派），授予管理员同意。 有关详细步骤，请参阅授予权限。
设置
每个回合时，智能体会调用具有回合上下文的 RegisterObservability 函数。 内置缓存使用来自 AgenticUserAuthorization 处理程序的用户委托令牌执行 OBO 交换，获取作用域为 Agent365.Observability.OtelWrite 的令牌。
有关完整的设置指南，包括包、配置和代码示例，请参阅 Agent Framework 应用中的智能体令牌缓存。
使用 S2S 启用的 Agent 365
启用 Agent 365 的智能体也可以使用 S2S（服务到服务）身份验证，而不是 OBO。 智能体通过两步智能体身份链使用其自身的服务主体身份获取令牌：
-
getAgenticApplicationToken(tenantId, agentId)：客户端凭据 + 联合托管身份 (FMI) 路径 - MSAL
acquireTokenForClient，其中应用令牌为clientAssertion且作用域为api://9b975845-388f-4429-889e-eab1ef63949c/.default
备注
联邦托管标识 (FMI) 是一种体系结构，其中托管标识通过联邦标识凭据参与工作负载标识联合中，从而实现基于标识之间信任关系的令牌交换和无密码身份验证。
必须提供自定义 TokenResolver 并设置 UseS2SEndpoint = true。
必备条件
Entra 应用注册：一个包含客户端 ID、客户端密钥和租户 ID 的服务主体（应用注册）
应用程序 API 权限：添加
Agent365.Observability.OtelWrite（应用程序），授予管理员同意Agent365.Observability.OtelWrite应用角色：智能体的服务主体必须在 Agent365 可观测性资源上被分配了OtelWrite角色。 使用 Agent 365 CLI：a365 setup permissions bot --config-dir "<path-to-config-dir>"备注
角色传播可能需要几分钟。 在此期间，预计导出终结点会出现初始 401 或 403 错误。
步骤 1：环境配置
以下代码示例显示如何在启用自定义 S2S 令牌流之前，设置所需的连接、租户、客户端凭据和可观测性导出器环境设置。
不需要 AgenticUserAuthorization 处理程序。 S2S 使用手动智能体身份链 (get_agentic_application_token + MSAL acquire_token_for_client) 来获取作用域限定于可观测性资源的令牌。
CONNECTIONSMAP__0__SERVICEURL=*
CONNECTIONSMAP__0__CONNECTION=SERVICE_CONNECTION
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID=<your-client-id>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET=<your-client-secret>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID=<your-tenant-id>
ENABLE_A365_OBSERVABILITY=true
ENABLE_A365_OBSERVABILITY_EXPORTER=true
步骤 2：使用自定义令牌解析器配置 Distro
以下示例显示如何启用 Agent 365 导出，并注册自定义 TokenResolver，以便导出器能够针对每个智能体和租户检索 S2S 令牌。
from microsoft.opentelemetry import use_microsoft_opentelemetry
_token_cache: dict[str, str] = {}
def token_resolver(agent_id: str, tenant_id: str) -> str | None:
return _token_cache.get(f"{agent_id}:{tenant_id}")
use_microsoft_opentelemetry(
enable_a365=True,
a365_token_resolver=token_resolver,
a365_use_s2s_endpoint=True,
a365_enable_observability_exporter=True,
)
步骤 3：获取并缓存 S2S 令牌
针对每条传入消息，通过智能体标识链获取 S2S 令牌，并将其缓存以供解析器使用。
import asyncio
from msal import ConfidentialClientApplication
from microsoft.opentelemetry.a365.core import BaggageBuilder, InvokeAgentScope, InvokeAgentScopeDetails, Request
OBSERVABILITY_S2S_SCOPE = "api://9b975845-388f-4429-889e-eab1ef63949c/.default"
async def get_agentic_s2s_token(connection, tenant_id: str, agent_id: str) -> str:
# Step 1: Get agentic application token (client_credentials + fmi_path)
app_token = await connection.get_agentic_application_token(tenant_id, agent_id)
if not app_token:
raise ValueError(f"Failed to get agentic app token for agent {agent_id}")
# Step 2: Exchange for observability-scoped token
cca = ConfidentialClientApplication(
client_id=agent_id,
authority=f"https://login.microsoftonline.com/{tenant_id}",
client_credential={"client_assertion": app_token},
)
result = await asyncio.to_thread(
lambda: cca.acquire_token_for_client(scopes=[OBSERVABILITY_S2S_SCOPE])
)
if not result or "access_token" not in result:
raise ValueError(f"Token acquisition failed: {result}")
return result["access_token"]
# In your message handler : use SDK helpers to get agent/tenant from the activity:
@AGENT_APP.activity("message")
async def on_message(context: TurnContext, _state: TurnState):
# get_agentic_instance_id reads from recipient (SDK convention)
agent_id = context.activity.get_agentic_instance_id()
tenant_id = context.activity.get_agentic_tenant_id()
# Acquire S2S token and cache BEFORE creating spans
connection = CONNECTION_MANAGER.get_connection("SERVICE_CONNECTION")
token = await get_agentic_s2s_token(connection, tenant_id, agent_id)
_token_cache[f"{agent_id}:{tenant_id}"] = token
# Wrap spans in BaggageBuilder so the exporter can resolve the token
request = Request(content=user_message, session_id=None)
with BaggageBuilder().tenant_id(tenant_id).agent_id(agent_id).build():
invoke_scope = InvokeAgentScope.start(request, InvokeAgentScopeDetails(), agent_details)
with invoke_scope:
invoke_scope.record_input_messages([user_message])
invoke_scope.record_output_messages([response])
重要提示
S2S 必须采用手动两步流 (get_agentic_application_token + MSAL acquire_token_for_client)。
AgenticUserAuthorization.get_token() 返回一个限定为 5a807f24-.../.default (Bot Framework) 的令牌，而不是可观测性资源 api://9b975845-.../.default：S2S 终结点以 401 InvalidAudience 拒绝它。
- 使用
context.activity.get_agentic_instance_id()和get_agentic_tenant_id()从活动读取智能体和租户（根据 SDK 约定从recipient读取）。 - 在创建跨度之前，先获取并缓存 S2S 令牌。 导出器的
BatchSpanProcessor可能会在处理程序完成之前刷新：如果令牌尚未缓存，导出会失败。 - 将所有 A365 作用域整合在
BaggageBuilder中，以便导出器知道应为哪个智能体和租户解析令牌。 若未包含这些信息，跨度将被静默丢弃，并显示“未找到具有租户/智能体身份的跨度”。
使用 OBO 的自定义引擎
自定义引擎智能体使用带有 Azure Bot OAuth 连接的标准应用注册，而非智能体身份链。 通过使用 OBO，智能体通过 Azure 机器人 OAuth 获得一个用户令牌，该令牌已由 Bot Framework 令牌服务限定为 A365 可观测性 API。 单次 getToken 或 GetTurnTokenAsync 调用会返回正确的限定令牌，因此无需 exchangeToken。
必备条件
使用委派的 API 权限的 Entra 应用注册。 添加 Agent365.Observability.OtelWrite（已委派）并授予管理员同意
重要提示
令牌缓存中的 agentId 必须与应用注册的客户端 ID 相匹配，而不是与活动的 agenticAppId 相匹配，自定义引擎智能体中没有该项。 导出 URL 包含 agentId，如果不匹配会导致 HTTP 403。
步骤 1：环境和应用配置
以下示例演示如何配置您的应用和运行环境，包括服务连接值、租户和客户端设置，以及所需的授权映射。
# .env
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID=<your-client-id>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET=<your-client-secret>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID=<your-tenant-id>
CONNECTIONSMAP__0__CONNECTION=SERVICE_CONNECTION
CONNECTIONSMAP__0__SERVICEURL=*
# Auth handler config : TYPE is required, name is uppercased by load_configuration_from_env
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__OBOCONNECTIONPROFILE__TYPE=UserAuthorization
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__OBOCONNECTIONPROFILE__SETTINGS__AZUREBOTOAUTHCONNECTIONNAME=oboConnectionProfile
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__OBOCONNECTIONPROFILE__SETTINGS__SCOPES=api://9b975845-388f-4429-889e-eab1ef63949c/Agent365.Observability.OtelWrite
ENABLE_A365_OBSERVABILITY=true
ENABLE_A365_OBSERVABILITY_EXPORTER=true
重要提示
load_configuration_from_env 会将所有环境变量键转换为大写。 处理程序名称变为 OBOCONNECTIONPROFILE，您必须在 auth_handlers 和 get_token() 调用中使用该确切大小写引用它。 缺少 TYPE 会导致运行时出现 Auth handler ... not recognized or not configured。
步骤 2：为 OBO 配置 Distro
以下示例显示如何启用 Agent 365 导出，保持导出器在 OBO 终结点上，并注册自定义 TokenResolver 以在导出期间返回委派的令牌。
from microsoft.opentelemetry import use_microsoft_opentelemetry
_token_cache: dict[str, str] = {}
def token_resolver(agent_id: str, tenant_id: str) -> str | None:
return _token_cache.get(f"{agent_id}:{tenant_id}")
environ["ENABLE_A365_OBSERVABILITY_EXPORTER"] = "true"
use_microsoft_opentelemetry(
enable_a365=True,
a365_token_resolver=token_resolver,
a365_use_s2s_endpoint=False, # OBO uses /observability endpoint
a365_enable_observability_exporter=True,
)
备注
OBO 模式在 aiohttpApplication 上需要 jwt_authorization_middleware（用于验证来自 Bot Framework 的入站 JWT（JSON Web 令牌））。 S2S/模拟器路径不应该包含此中间件。
from microsoft_agents.hosting.aiohttp import jwt_authorization_middleware
app = Application(middlewares=[jwt_authorization_middleware])
步骤 3：获取 OBO 令牌
以下示例显示如何从已配置的 Azure 机器人 OAuth 连接请求委托的 OBO 令牌，并按应用客户端和租户将其缓存以供导出器使用。
from microsoft_agents.hosting.core import (
AgentApplication, Authorization, MemoryStorage, TurnContext, TurnState,
)
from microsoft_agents.activity import load_configuration_from_env
from microsoft_agents.authentication.msal import MsalConnectionManager
from microsoft_agents.hosting.aiohttp import CloudAdapter
# Auth handlers are loaded from .env via load_configuration_from_env (see Environment config above)
agents_sdk_config = load_configuration_from_env(environ)
STORAGE = MemoryStorage()
CONNECTION_MANAGER = MsalConnectionManager(**agents_sdk_config)
ADAPTER = CloudAdapter(connection_manager=CONNECTION_MANAGER)
AUTHORIZATION = Authorization(STORAGE, CONNECTION_MANAGER, **agents_sdk_config)
AGENT_APP = AgentApplication[TurnState](
storage=STORAGE, adapter=ADAPTER, authorization=AUTHORIZATION, **agents_sdk_config,
)
CLIENT_ID = environ.get("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID", "")
TENANT_ID = environ.get("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID", "")
# Message handler : get_token returns a token already scoped to the observability API.
# The Azure Bot Token Service performs the OBO exchange internally based on the
# OAuth connection's configured scope. No manual MSAL exchange_token call is needed.
@AGENT_APP.activity("message", auth_handlers=["OBOCONNECTIONPROFILE"])
async def on_message(context: TurnContext, _state: TurnState):
token_response = await AGENT_APP.auth.get_token(context, "OBOCONNECTIONPROFILE")
# token_response.token has aud=<a365-observability-app-id>,
# scp=Agent365.Observability.OtelWrite
_token_cache[f"{CLIENT_ID}:{TENANT_ID}"] = token_response.token
重要提示
Azure 门户先决条件：名为 oboConnectionProfile 的 Azure 机器人 OAuth 连接必须将其范围设置为 api://9b975845-388f-4429-889e-eab1ef63949c/Agent365.Observability.OtelWrite。 如果没有此设置，令牌将限定为机器人自己的访问群体 (api://botid-...)，并且导出会失败并具有 HTTP 401 InvalidAudience。
备注
AGENT_APP.auth.get_token() 会直接返回正确限定的令牌，无需调用 exchange_token()。 当 OAuth 连接范围针对 A365 可观测性资源时，Bot Framework 令牌服务会处理 OBO 交换。
使用 S2S 的自定义引擎
自定义引擎智能体可通过 S2S（客户端凭据）使用服务连接凭据获取仅限应用的令牌。 该方法使用标准 MSAL 客户端身份验证，无需智能体标识链。
必备条件
-
Azure AD 应用注册：必须是自定义引擎（标准）应用。 启用 Agent 365 的应用注册不能在可观测性资源 (
AADSTS82001)中使用纯client_credentials。 -
应用程序权限：添加
Agent365.Observability.OtelWrite（应用程序，非委托），并授予管理员同意。
重要提示
用于缓存的 agentId必须是 ServiceConnection 的 ClientId。 导出 URL 是 /observabilityService/tenants/{tenantId}/otlp/agents/{agentId}/traces：不匹配会导致 HTTP 403。
步骤 1：环境和应用配置
以下示例演示如何配置您的应用和运行环境，包括服务连接值、租户和客户端设置，以及所需的授权映射。
# .env
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID=<your-client-id>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET=<your-client-secret>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID=<your-tenant-id>
CONNECTIONSMAP__0__CONNECTION=SERVICE_CONNECTION
CONNECTIONSMAP__0__SERVICEURL=*
ENABLE_A365_OBSERVABILITY=true
ENABLE_A365_OBSERVABILITY_EXPORTER=true
步骤 2：为 S2S 配置 Distro
以下示例显示如何启用 Agent 365 导出，将导出器设置为 S2S 终结点，以及在导出时注册一个自定义 TokenResolver 以查找令牌。
from microsoft.opentelemetry import use_microsoft_opentelemetry
_token_cache: dict[str, str] = {}
def token_resolver(agent_id: str, tenant_id: str) -> str | None:
return _token_cache.get(f"{agent_id}:{tenant_id}")
use_microsoft_opentelemetry(
enable_a365=True,
a365_token_resolver=token_resolver,
a365_use_s2s_endpoint=True, # S2S uses /observabilityService endpoint
a365_enable_observability_exporter=True,
)
步骤 3：获取 S2S 令牌
以下示例显示如何使用服务连接凭据为可观测性资源请求仅应用访问令牌，然后按智能体和租户进行缓存以供导出器使用。
# Force agentId to ServiceConnection ClientId (custom engine agents have no agenticAppId)
agent_id = os.environ.get("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID")
tenant_id = os.environ.get("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID")
connection = CONNECTION_MANAGER.get_connection("SERVICE_CONNECTION")
token = await connection.get_access_token(
resource_url="https://login.microsoftonline.com",
scopes=["api://9b975845-388f-4429-889e-eab1ef63949c/.default"],
)
_token_cache[f"{agent_id}:{tenant_id}"] = token
步骤 4：为跨度导出设置负载
Agent365 导出器要求在跨度上下文中设置负载（租户 ID 和智能体 ID）。 没有它，导出器会静默丢弃跨度，并显示消息 No spans with tenant/agent identity found.。
from microsoft.opentelemetry.a365.core import BaggageBuilder, InvokeAgentScope
# Baggage must wrap the span as a context manager
with BaggageBuilder().tenant_id(tenant_id).agent_id(agent_id).build():
invoke_scope = InvokeAgentScope.start(request, InvokeAgentScopeDetails(), agent_details)
with invoke_scope:
invoke_scope.record_input_messages([user_message])
invoke_scope.record_output_messages([response])