Agent365 Auth token with external 3rd party agents

[DeepMalh44]

Is your feature request related to a problem?

We are working with one of our customers who is building a third-party (3P) agent hosted on Google Cloud Platform (GCP) outside the Microsoft 365 Agents hosting framework and we want to integrate Agent 365 Observability to send telemetry into the A365 Admin Center. We need guidance on how to obtain the authentication token required by the observability exporter when we don't have access to a TurnContext or the M365 Agents SDK hosting layer
.
The SDK's token_resolver requires a Bearer token for instance from sample code the scope iis https://api.powerplatform.com/.default, but all documented paths obtain this via OBO token exchange through TurnContext.exchange_token() which requires the M365 Agents hosting framework.

These customer agent doesn't run inside that framework, so we have no TurnContext and no incoming Microsoft channel tokens to exchange.

Questions:

Can a 3P external agent use Entra ID client credentials flow (or another method) to independently obtain a valid token for the observability API?
What Entra ID app registration / permissions are needed for this?
Is the SpectraExporterOptions (OTLP sidecar) path an alternative for 3P agents to route telemetry into the Admin Center?

Any guidance or sample on how this can be achieved for externally hosted agents?
Following is the sample I am yet to test but was not sure if the same MSAL generated token method would work, before giving it a shot thought of checking here with the experts. here is that sample code (assumptions here are frontier enrolled, Agent blueprint id registered in Entra):

Priority

High

Describe the solution you'd like

Guidance on how this can be achieved with 3rd party agents.

Describe alternatives you've considered

No response

Available workarounds

No response

Additional context

No response

Code of Conduct

• I agree to follow the Microsoft Open Source Code of Conduct.

[DeepMalh44]

Added sample code to the original post with MSAL auth token generation, if experts can validate we can use it?

agent_with_observability.py

Complete example of a 3P agent (e.g., hosted on GCP) using:
- Agent 365 Observability SDK for telemetry
- MSAL client credentials flow for token acquisition (inferred for 3P agents)
- OpenAI for LLM inference
"""

import logging
import os
from urllib.parse import urlparse

from dotenv import load_dotenv
from msal import ConfidentialClientApplication
from openai import OpenAI

# --- Agent 365 Observability Imports ---
from microsoft_agents_a365.observability.core.config import configure
from microsoft_agents_a365.observability.core import (
    InvokeAgentScope,
    InvokeAgentDetails,
    InferenceScope,
    InferenceCallDetails,
    InferenceOperationType,
    ExecuteToolScope,
    ToolCallDetails,
    AgentDetails,
    TenantDetails,
    Request,
    ExecutionType,
)
from microsoft_agents_a365.observability.core.middleware.baggage_builder import (
    BaggageBuilder,
)

load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# =========================================================================
# 1. TOKEN RESOLVER (MSAL Client Credentials — inferred for 3P agents)
# =========================================================================

# These values come from a365.generated.config.json after "a365 setup all"
AGENT_BLUEPRINT_CLIENT_ID = os.getenv("AGENT_BLUEPRINT_CLIENT_ID")
AGENT_BLUEPRINT_CLIENT_SECRET = os.getenv("AGENT_BLUEPRINT_CLIENT_SECRET")
TENANT_ID = os.getenv("TENANT_ID")
AGENT_ID = os.getenv("AGENT_ID")  # Your agent's ID from the blueprint

AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}"
OBSERVABILITY_SCOPES = ["https://api.powerplatform.com/.default"]

# Create MSAL app once (singleton)
_msal_app = None
if AGENT_BLUEPRINT_CLIENT_ID and AGENT_BLUEPRINT_CLIENT_SECRET:
    _msal_app = ConfidentialClientApplication(
        client_id=AGENT_BLUEPRINT_CLIENT_ID,
        client_credential=AGENT_BLUEPRINT_CLIENT_SECRET,
        authority=AUTHORITY,
    )


def token_resolver(agent_id: str, tenant_id: str) -> str | None:
    """
    Token resolver for the Agent 365 Observability exporter.

    NOTE: This uses MSAL client credentials flow. This is INFERRED
    for 3P agents — the documented path uses OBO via TurnContext.
    If this doesn't work, the observability API may require delegated tokens.
    """
    if not _msal_app:
        logger.warning("MSAL app not configured — no token available")
        return None

    try:
        result = _msal_app.acquire_token_for_client(scopes=OBSERVABILITY_SCOPES)
        if "access_token" in result:
            logger.debug("Token acquired for agent %s, tenant %s", agent_id, tenant_id)
            return result["access_token"]

        logger.error(
            "Token acquisition failed: %s — %s",
            result.get("error"),
            result.get("error_description"),
        )
        return None
    except Exception as e:
        logger.error("Token resolver error: %s", e)
        return None


# =========================================================================
# 2. OBSERVABILITY SETUP
# =========================================================================

# Set to "true" to export to A365 Admin Center, "false" for console output
os.environ.setdefault("ENABLE_A365_OBSERVABILITY_EXPORTER", "false")

configure(
    service_name=os.getenv("SERVICE_NAME", "my-gcp-agent"),
    service_namespace=os.getenv("SERVICE_NAMESPACE", "my-org.agents"),
    token_resolver=token_resolver,
    cluster_category="prod",
)
logger.info("Observability configured")


# =========================================================================
# 3. AGENT LOGIC WITH OBSERVABILITY SCOPES
# =========================================================================

# Your LLM client
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
MODEL_NAME = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

# Agent metadata
agent_details = AgentDetails(
    agent_id=AGENT_ID or "my-gcp-agent-id",
    agent_name="My GCP Agent",
    agent_description="A 3P agent hosted on GCP",
    tenant_id=TENANT_ID,
)
tenant_details = TenantDetails(tenant_id=TENANT_ID)


def process_message(user_message: str) -> str:
    """
    Process a user message with full observability tracing.
    This shows the manual instrumentation pattern (like the CrewAI sample).
    """

    # --- Set baggage context (flows to all child spans) ---
    with (
        BaggageBuilder()
        .tenant_id(TENANT_ID or "")
        .agent_id(AGENT_ID or "")
        .build()
    ):

        # --- InvokeAgentScope: wraps the entire agent invocation ---
        endpoint = urlparse("https://my-gcp-agent.run.app/api/messages")
        invoke_details = InvokeAgentDetails(
            endpoint=endpoint,
            session_id="session-001",
            details=agent_details,
        )
        request = Request(
            content=user_message,
            execution_type=ExecutionType.HUMAN_TO_AGENT,
        )

        with InvokeAgentScope.start(
            invoke_agent_details=invoke_details,
            tenant_details=tenant_details,
            request=request,
        ) as invoke_scope:

            # --- InferenceScope: wraps the LLM call ---
            inference_details = InferenceCallDetails(
                operationName=InferenceOperationType.CHAT,
                model=MODEL_NAME,
                providerName="OpenAI",
            )

            with InferenceScope.start(
                details=inference_details,
                agent_details=agent_details,
                tenant_details=tenant_details,
                request=request,
            ) as inference_scope:

                # Actual LLM call
                completion = client.chat.completions.create(
                    model=MODEL_NAME,
                    messages=[{"role": "user", "content": user_message}],
                )

                response_text = completion.choices[0].message.content
                usage = completion.usage

                # Record token usage on the inference span
                if usage:
                    inference_scope.record_input_tokens(usage.prompt_tokens)
                    inference_scope.record_output_tokens(usage.completion_tokens)
                inference_scope.record_finish_reasons(["stop"])

            # Record the final response on the invoke span
            invoke_scope.record_response(response_text)

    return response_text


# =========================================================================
# 4. EXAMPLE WITH TOOL EXECUTION
# =========================================================================

def process_message_with_tool(user_message: str) -> str:
    """
    Shows how to trace tool execution alongside LLM calls.
    """

    with (
        BaggageBuilder()
        .tenant_id(TENANT_ID or "")
        .agent_id(AGENT_ID or "")
        .build()
    ):

        endpoint = urlparse("https://my-gcp-agent.run.app/api/messages")
        invoke_details = InvokeAgentDetails(
            endpoint=endpoint,
            session_id="session-002",
            details=agent_details,
        )
        request = Request(content=user_message)

        with InvokeAgentScope.start(
            invoke_agent_details=invoke_details,
            tenant_details=tenant_details,
            request=request,
        ) as invoke_scope:

            # --- ExecuteToolScope: wraps a tool call ---
            tool_details = ToolCallDetails(
                tool_name="get_weather",
                tool_arguments={"location": "Seattle"},
            )

            with ExecuteToolScope.start(
                tool_call_details=tool_details,
                agent_details=agent_details,
                tenant_details=tenant_details,
            ) as tool_scope:
                # Your tool logic here
                tool_result = "72°F, sunny"
                tool_scope.record_response(tool_result)

            # --- InferenceScope: LLM uses tool result ---
            inference_details = InferenceCallDetails(
                operationName=InferenceOperationType.CHAT,
                model=MODEL_NAME,
                providerName="OpenAI",
            )

            with InferenceScope.start(
                details=inference_details,
                agent_details=agent_details,
                tenant_details=tenant_details,
                request=request,
            ) as inference_scope:

                completion = client.chat.completions.create(
                    model=MODEL_NAME,
                    messages=[
                        {"role": "user", "content": user_message},
                        {"role": "assistant", "content": f"Weather data: {tool_result}"},
                    ],
                )
                response_text = completion.choices[0].message.content

                if completion.usage:
                    inference_scope.record_input_tokens(completion.usage.prompt_tokens)
                    inference_scope.record_output_tokens(completion.usage.completion_tokens)

            invoke_scope.record_response(response_text)

    return response_text


# =========================================================================
# 5. RUN IT
# =========================================================================

if __name__ == "__main__":
    # Test with console exporter first (ENABLE_A365_OBSERVABILITY_EXPORTER=false)
    print("\n--- Simple message ---")
    result = process_message("What is the capital of France?")
    print(f"Response: {result}")

    print("\n--- Message with tool ---")
    result = process_message_with_tool("What's the weather in Seattle?")
    print(f"Response: {result}")