Agent 365 匯出器在匯出遙測資料時需要權杖解析器來進行驗證。 本指南涵蓋透過Microsoft 365 Agents SDK 建置的 Agent 設定,包括支援已啟用 Agent 365 的 Agent,以及跨 .NET、Python 和 Node.js 的自訂引擎 Agent。
如需發行版安裝、一般組態,以及非 Agent SDK 情境,請參閱 Microsoft OpenTelemetry 發行版。
概觀
有四種驗證情境,取決於您的 Agent 類型及其取得權杖的方式。 權杖取得可以使用代理者流程 (OBO) 或服務對服務 (S2S)。 選擇符合您設定的情境:
| 案例 | 描述 |
|---|---|
| 使用 OBO,且已啟用 Agent 365 | 發行版內建的 AgenticTokenCache 會自動處理權杖取得。 無需自訂解析器。 這是建議用於已啟用 Agent 365 之 Agent 的方式。 |
| 使用 S2S,且已啟用 Agent 365 | Agent 透過使用 Agent 身分識別鏈 (getAgenticApplicationToken + Microsoft 驗證庫 (MSAL)) 來取得權杖。 需要自訂 TokenResolver。 當 OBO 不可用或您需要僅限應用程式的權杖時,請使用此方法。 |
| 使用 OBO 的自訂引擎 | Agent 透過 Azure 機器人 OAuth 取得一個使用者權杖,其範圍限定於可檢視性 API。 需要自訂 TokenResolver 以及 Azure 機器人 OAuth 連線。 |
| 使用 S2S 的自訂引擎 | Agent 使用用戶端認證來取得應用程式專用權杖。 需要自訂 TokenResolver。 應用程式註冊必須是標準 (非 Agent) 應用程式。 |
使用 OBO,且已啟用 Agent 365
已啟用 Agent 365 的 Agent 會從 Agent 365 平台接收具有 Agent 身分識別 (agenticAppId、agenticUserId) 的要求。 透過 OBO 時,發行版內建的 AgenticTokenCache 會自動處理權杖取得,無需自訂權杖解析器。
先決條件
- Entra 應用程式註冊:具備用戶端識別碼、客戶端密碼和租用戶識別碼的服務主體 (應用程式註冊)
-
委派 API 權限:新增
Agent365.Observability.OtelWrite(委派),授與管理員同意。 詳細步驟請參閱授與權限。
設定
在每回合中,您的 Agent 會呼叫具有回合上下文的 RegisterObservability 函式。 內建快取會使用使用者從 AgenticUserAuthorization 處理常式取得的委派權杖來執行 OBO 交換,並取得範圍限定於 Agent365.Observability.OtelWrite 的權杖。
如需完整設定說明,包括套件、組態及程式碼範例,請參閱 Agent Framework 應用程式中的 Agent 權杖快取。
使用 S2S,且已啟用 Agent 365
已啟用 Agent 365 的Agent 也能使用 S2S (服務對服務),取代 OBO。 Agent 會透過其專屬服務主體身分識別,透過兩步驟 Agent 身分識別鏈來取得權杖:
-
getAgenticApplicationToken(tenantId, agentId):用戶端認證 + 同盟受控識別 (FMI) 路徑 - 具備
clientAssertion等應用程式權杖且範圍為api://9b975845-388f-4429-889e-eab1ef63949c/.default的 MSALacquireTokenForClient
注意
同盟受控識別 (FMI)是一種架構,受控識別透過同盟身分識別憑證參與工作負載身分識別同盟,實現基於身分識別間信任關係的權杖交換與無密碼驗證。
您必須提供自訂 TokenResolver 並設定 UseS2SEndpoint = true。
先決條件
Entra 應用程式註冊:具備用戶端識別碼、客戶端密碼和租用戶識別碼的服務主體 (應用程式註冊)
應用程式 API 權限:新增
Agent365.Observability.OtelWrite(應用程式) 授與管理員同意Agent365.Observability.OtelWrite應用程式角色:Agent 的服務主體必須在 Agent365 可檢視性資源中獲指派OtelWrite角色。 使用 Agent 365 CLI:a365 setup permissions bot --config-dir "<path-to-config-dir>"注意
角色傳播可能需要幾分鐘的時間。 在此期間,預期匯出端點會出現初始的 401 或 403 錯誤。
步驟 1:環境組態
以下程式碼範例說明如何在啟用自訂 S2S 權杖流程前,設定所需的連線、租用戶、客戶認證及可檢視性匯出器環境設定。
不需要 AgenticUserAuthorization 處理常式。 S2S 使用手動 Agent身分識別鏈 (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:用自訂權杖解析器設定發行版
以下範例說明如何啟用 Agent 365 匯出並註冊自訂 TokenResolver ,讓匯出器能為每個 Agent 與租用戶檢索 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 權杖
在每則傳入訊息中,透過 Agent 身分識別鏈取得 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 端點會以 401InvalidAudience 拒絕該權杖。
- 使用
context.activity.get_agentic_instance_id()和get_agentic_tenant_id()從活動中讀取 Agent 和租用戶 (依 SDK慣例從recipient讀取)。 - 在建立跨度前,先取得並快取 S2S 權杖。 匯出器的
BatchSpanProcessor可能會在處理常式結束前排清;如果尚未快取權杖,匯出將會失敗。 - 將所有 A365 範圍裝合在
BaggageBuilder中,以便匯出器知道要為哪個 Agent 和租用戶解析權杖。 若未帶 baggage,,跨度會遭到靜默丟棄,並顯示「未尋找到具有租用戶/Agent 身分識別的跨度」。
使用 OBO 的自訂引擎
自訂引擎 Agent 會使用具有 Azure Bot OAuth 連線,而非 Agent 身分識別鏈的標準應用程式註冊。 透過使用 OBO,Agent 會透過 Azure 機器人 OAuth 取得一個使用者權杖,該權杖的範圍已由 Bot Framework 權杖服務預先設定好,適用於 A365 可檢視性 API。 單一 getToken 或 GetTurnTokenAsync 呼叫會傳回已正確設定範圍的權杖,因此無需 exchangeToken。
先決條件
使用委派 API 權限的 Entra 應用程式註冊。 新增 Agent365.Observability.OtelWrite (委派),並授與管理員同意
重要
權杖快取中的 agentId 必須符合應用程式註冊的用戶端識別碼,而不是活動的 agenticAppId (自訂引擎 Agent 沒有此項)。 匯出 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 的發行版
以下範例說明如何啟用 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 的自訂引擎
自訂引擎 Agent 可以使用 S2S (用戶端認證),使用服務連線認證取得應用程式專用的權杖。 此方法使用標準 MSAL 用戶端認證,無需 Agent 身分識別鏈。
先決條件
-
Azure AD 應用程式註冊:必須為自訂引擎 (標準) 應用程式。 已啟用 Agent 365 的應用程式註冊無法使用純
client_credentials作為可檢視性資源 (AADSTS82001)。 -
應用程式權限:新增
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 的發行版
以下範例說明如何啟用 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 權杖
以下範例展示如何使用服務連線認證,要求應用程式專用的可檢視性資源存取權杖,並將其依 Agent 與租用戶快取,供匯出器使用。
# 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:設定 Baggage 以進行跨度匯出
Agent365 匯出器要求在跨度上下文中設定 Baggage (租用戶識別碼和 Agent 識別碼)。 若未設定,匯出器會靜默丟棄跨度並帶有訊息 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])