Support V1/V2 per-audience token acquisition for MCP servers

V2 MCP servers require individual OAuth tokens scoped to their own
audience GUID rather than the shared ATG token used by V1 servers.

- Add audience, scope, publisher, headers fields to MCPServerConfig
- Add resolve_token_scope_for_server() to determine OAuth scope:
  V2 servers (GUID audience) get <audience>/.default, V1 servers
  fall back to the shared ATG scope
- Add _attach_per_audience_tokens() to McpToolServerConfigurationService:
  acquires one token per unique audience (cached), attaches Authorization
  header to each server config after discovery
- Extend list_tool_servers() to accept optional authorization context;
  calls _attach_per_audience_tokens() when provided
- Preserve audience/scope/publisher fields in manifest and gateway parsers
- Update McpToolRegistrationService to pass auth context to
  list_tool_servers() and use per-server headers instead of a single
  shared token
- Update tests to reflect the new header flow

All V1 agents continue working unchanged (audience defaults to None,
falls back to ATG scope).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: biswapm

libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py

@@ -94,11 +94,15 @@ async def add_tool_servers_to_agent(
 
             options = ToolOptions(orchestrator_name=self._orchestrator_name)
 
-            # Get MCP server configurations
+            # Get MCP server configurations — pass auth context so each server receives
+            # its own per-audience Authorization token (V1 = shared ATG, V2 = per-GUID).
             server_configs = await self._mcp_server_configuration_service.list_tool_servers(
                 agentic_app_id=agentic_app_id,
                 auth_token=auth_token,
                 options=options,
+                authorization=auth,
+                auth_handler_name=auth_handler_name,
+                turn_context=turn_context,
             )
 
             self._logger.info(f"Loaded {len(server_configs)} MCP server configurations")
@@ -112,16 +116,15 @@ async def add_tool_servers_to_agent(
                 server_name = config.mcp_server_name or config.mcp_server_unique_name
 
                 try:
-                    # Prepare auth headers
-                    headers = {}
-                    if auth_token:
-                        headers[Constants.Headers.AUTHORIZATION] = (
-                            f"{Constants.Headers.BEARER_PREFIX} {auth_token}"
+                    # Merge base (non-auth) headers with per-server headers from list_tool_servers.
+                    # server.headers already contains the correct per-audience Authorization token.
+                    base_headers = {
+                        Constants.Headers.USER_AGENT: Utility.get_user_agent_header(
+                            self._orchestrator_name
                         )
-
-                    headers[Constants.Headers.USER_AGENT] = Utility.get_user_agent_header(
-                        self._orchestrator_name
-                    )
+                    }
+                    server_headers = dict(config.headers) if config.headers else {}
+                    headers = {**base_headers, **server_headers}  # server auth takes precedence
 
                     # Create httpx client with auth headers configured
                     http_client = httpx.AsyncClient(

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/models/mcp_server_config.py

@@ -6,7 +6,7 @@
 """
 
 from dataclasses import dataclass
-from typing import Optional
+from typing import Dict, Optional
 
 
 @dataclass
@@ -25,6 +25,18 @@ class MCPServerConfig:
     #: instead of constructing the URL from the base URL and unique name.
     url: Optional[str] = None
 
+    #: Per-server HTTP headers (includes the Authorization header set by attach_per_audience_tokens).
+    headers: Optional[Dict[str, str]] = None
+
+    #: Per-server AppId (V2) or shared ATG AppId (V1). None means treat as V1.
+    audience: Optional[str] = None
+
+    #: OAuth scope, e.g. "Tools.ListInvoke.All" (V2) or "McpServers.Mail.All" (V1).
+    scope: Optional[str] = None
+
+    #: Publisher identifier for the MCP server.
+    publisher: Optional[str] = None
+
     def __post_init__(self):
         """Validate the configuration after initialization."""
         if not self.mcp_server_name:

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/services/mcp_tool_server_configuration_service.py

@@ -29,15 +29,16 @@
 
 # Third-party imports
 import aiohttp
-from microsoft_agents.hosting.core import TurnContext
+from microsoft_agents.hosting.core import Authorization, TurnContext
 
 # Local imports
 from ..models import ChatHistoryMessage, ChatMessageRequest, MCPServerConfig, ToolOptions
 from ..utils import Constants
 from ..utils.utility import (
-    get_tooling_gateway_for_digital_worker,
     build_mcp_server_url,
     get_chat_history_endpoint,
+    get_tooling_gateway_for_digital_worker,
+    resolve_token_scope_for_server,
 )
 
 # Runtime Imports
@@ -89,22 +90,38 @@ def __init__(self, logger: Optional[logging.Logger] = None):
     # --------------------------------------------------------------------------
 
     async def list_tool_servers(
-        self, agentic_app_id: str, auth_token: str, options: Optional[ToolOptions] = None
+        self,
+        agentic_app_id: str,
+        auth_token: str,
+        options: Optional[ToolOptions] = None,
+        authorization: Optional[Authorization] = None,
+        auth_handler_name: Optional[str] = None,
+        turn_context: Optional[TurnContext] = None,
     ) -> List[MCPServerConfig]:
         """
         Gets the list of MCP Servers that are configured for the agent.
 
+        When ``authorization``, ``auth_handler_name``, and ``turn_context`` are all provided,
+        per-audience OAuth tokens are acquired for each server after discovery:
+        - V1 servers (no ``audience`` field) share the shared ATG token (one exchange).
+        - V2 servers each receive a token scoped to their own audience GUID.
+
         Args:
             agentic_app_id: Agentic App ID for the agent.
-            auth_token: Authentication token to access the MCP servers.
+            auth_token: Authentication token used for gateway discovery.
             options: Optional ToolOptions instance containing optional parameters.
+            authorization: Optional Authorization context for per-audience token exchange.
+            auth_handler_name: Optional auth handler name used with ``authorization``.
+            turn_context: Optional TurnContext used with ``authorization``.
 
         Returns:
-            List[MCPServerConfig]: Returns the list of MCP Servers that are configured.
+            List[MCPServerConfig]: Returns the list of MCP Servers that are configured,
+            each with an ``Authorization`` header attached when auth context is provided.
 
         Raises:
             ValueError: If required parameters are invalid or empty.
-            Exception: If there's an error communicating with the tooling gateway.
+            Exception: If there's an error communicating with the tooling gateway or
+                       a per-audience token exchange fails.
         """
         # Validate input parameters
         self._validate_input_parameters(agentic_app_id, auth_token)
@@ -117,9 +134,17 @@ async def list_tool_servers(
 
         # Determine configuration source based on environment
         if self._is_development_scenario():
-            return self._load_servers_from_manifest()
+            servers = self._load_servers_from_manifest()
         else:
-            return await self._load_servers_from_gateway(agentic_app_id, auth_token, options)
+            servers = await self._load_servers_from_gateway(agentic_app_id, auth_token, options)
+
+        # Acquire per-audience tokens and attach Authorization headers when auth context provided
+        if authorization is not None and auth_handler_name is not None and turn_context is not None:
+            servers = await self._attach_per_audience_tokens(
+                servers, authorization, auth_handler_name, turn_context
+            )
+
+        return servers
 
     # --------------------------------------------------------------------------
     # ENVIRONMENT DETECTION
@@ -138,6 +163,72 @@ def _is_development_scenario(self) -> bool:
         self._logger.debug(f"Environment: {environment}, Development scenario: {is_dev}")
         return is_dev
 
+    async def _attach_per_audience_tokens(
+        self,
+        servers: List[MCPServerConfig],
+        authorization: Authorization,
+        auth_handler_name: str,
+        turn_context: TurnContext,
+    ) -> List[MCPServerConfig]:
+        """
+        Acquire one OAuth token per unique audience and attach an ``Authorization: Bearer``
+        header to each server's headers.
+
+        V1 servers (no ``audience`` field, or audience matching the shared ATG AppId) all
+        share the same ATG-scoped token (one exchange). V2 servers each receive a token
+        scoped to their own audience GUID.
+
+        Args:
+            servers: List of MCP server configs returned from discovery.
+            authorization: Authorization context for token exchange.
+            auth_handler_name: Auth handler name to pass to the token exchange.
+            turn_context: TurnContext to pass to the token exchange.
+
+        Returns:
+            List[MCPServerConfig]: New list of server configs with ``Authorization`` headers set.
+
+        Raises:
+            Exception: If a token exchange fails for any server.
+        """
+        token_cache: Dict[str, str] = {}  # scope → bearer token
+        result: List[MCPServerConfig] = []
+
+        for server in servers:
+            scope = resolve_token_scope_for_server(server)
+
+            if scope not in token_cache:
+                self._logger.debug(
+                    f"Acquiring token for MCP server '{server.mcp_server_name}' (scope: {scope})"
+                )
+                token_result = await authorization.exchange_token(
+                    turn_context, [scope], auth_handler_name
+                )
+                if token_result is None or not token_result.token:
+                    raise Exception(
+                        f"Failed to obtain token for MCP server '{server.mcp_server_name}'"
+                        f" (scope: {scope})"
+                    )
+                token_cache[scope] = token_result.token
+
+            merged_headers: Dict[str, str] = dict(server.headers) if server.headers else {}
+            merged_headers[Constants.Headers.AUTHORIZATION] = (
+                f"{Constants.Headers.BEARER_PREFIX} {token_cache[scope]}"
+            )
+
+            result.append(
+                MCPServerConfig(
+                    mcp_server_name=server.mcp_server_name,
+                    mcp_server_unique_name=server.mcp_server_unique_name,
+                    url=server.url,
+                    headers=merged_headers,
+                    audience=server.audience,
+                    scope=server.scope,
+                    publisher=server.publisher,
+                )
+            )
+
+        return result
+
     # --------------------------------------------------------------------------
     # DEVELOPMENT: MANIFEST-BASED CONFIGURATION
     # --------------------------------------------------------------------------
@@ -481,6 +572,9 @@ def _parse_manifest_server_config(
                 mcp_server_name=mcp_server_name,
                 mcp_server_unique_name=mcp_server_unique_name,
                 url=final_url,
+                audience=server_element.get("audience"),
+                scope=server_element.get("scope"),
+                publisher=server_element.get("publisher"),
             )
 
         except Exception:
@@ -518,6 +612,9 @@ def _parse_gateway_server_config(
                 mcp_server_name=mcp_server_name,
                 mcp_server_unique_name=mcp_server_unique_name,
                 url=final_url,
+                audience=server_element.get("audience"),
+                scope=server_element.get("scope"),
+                publisher=server_element.get("publisher"),
             )
 
         except Exception:

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/utils/utility.py

@@ -5,7 +5,13 @@
 Provides utility functions for the Tooling components.
 """
 
+from __future__ import annotations
+
 import os
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..models.mcp_server_config import MCPServerConfig
 
 
 # Constants for base URLs
@@ -17,6 +23,9 @@
 PPAPI_TOKEN_SCOPE = "https://api.powerplatform.com"
 PROD_MCP_PLATFORM_AUTHENTICATION_SCOPE = "ea9ffc3e-8a23-4a7d-836d-234d7c7565c1/.default"
 
+# Shared ATG AppId — V1 servers (no audience field) use this scope
+ATG_APP_ID = "ea9ffc3e-8a23-4a7d-836d-234d7c7565c1"
+
 
 def get_tooling_gateway_for_digital_worker(agentic_app_id: str) -> str:
     """
@@ -104,3 +113,26 @@ def get_chat_history_endpoint() -> str:
         str: The chat history endpoint URL.
     """
     return f"{_get_mcp_platform_base_url()}{CHAT_HISTORY_ENDPOINT_PATH}"
+
+
+def resolve_token_scope_for_server(server: MCPServerConfig) -> str:
+    """
+    Resolve the OAuth scope to request for a given MCP server.
+
+    V2 servers carry their own audience GUID in the ``audience`` field and receive
+    a token scoped to that GUID. V1 servers (no audience, audience equals the shared
+    ATG AppId, or audience starting with ``api://``) fall back to the shared ATG scope.
+
+    Args:
+        server: The MCP server configuration to resolve the scope for.
+
+    Returns:
+        str: The OAuth scope string (e.g. ``"<guid>/.default"``).
+    """
+    if (
+        server.audience is not None
+        and server.audience != ATG_APP_ID
+        and not server.audience.startswith("api://")
+    ):
+        return f"{server.audience}/.default"
+    return f"{ATG_APP_ID}/.default"

tests/tooling/extensions/agentframework/services/test_mcp_tool_registration_service.py

@@ -62,6 +62,7 @@ def mock_mcp_server_config(self):
         config.mcp_server_name = "test-mcp-server"
         config.mcp_server_unique_name = "test-mcp-server-unique"
         config.url = "https://test-mcp-server.example.com/api"
+        config.headers = None  # per-audience headers attached by list_tool_servers
         return config
 
     @pytest.fixture
@@ -79,8 +80,17 @@ async def test_httpx_client_has_authorization_header(
         mock_chat_client,
         mock_mcp_server_config,
     ):
-        """Test that httpx.AsyncClient is created with Authorization header."""
+        """Test that httpx.AsyncClient is created with Authorization header.
+
+        In the V1/V2 model, list_tool_servers (via _attach_per_audience_tokens) attaches
+        the per-audience Authorization header to config.headers before returning. The
+        mock server config must reflect this to verify the header reaches the httpx client.
+        """
         auth_token = "test-bearer-token-xyz"
+        # Simulate the Authorization header that _attach_per_audience_tokens would attach
+        mock_mcp_server_config.headers = {
+            Constants.Headers.AUTHORIZATION: (f"{Constants.Headers.BEARER_PREFIX} {auth_token}")
+        }
 
         with (
             patch.object(
@@ -481,6 +491,7 @@ async def test_full_client_lifecycle_single_server(
         mock_server_config.mcp_server_name = "test-server"
         mock_server_config.mcp_server_unique_name = "test-server-unique"
         mock_server_config.url = "https://test.example.com/api"
+        mock_server_config.headers = None  # per-audience headers attached by list_tool_servers
 
         mock_http_client_instance = MagicMock()
 
@@ -553,16 +564,19 @@ async def test_full_client_lifecycle_multiple_servers(
         mock_server_config1.mcp_server_name = "server-1"
         mock_server_config1.mcp_server_unique_name = "server-1-unique"
         mock_server_config1.url = "https://server1.example.com/api"
+        mock_server_config1.headers = None  # per-audience headers attached by list_tool_servers
 
         mock_server_config2 = Mock()
         mock_server_config2.mcp_server_name = "server-2"
         mock_server_config2.mcp_server_unique_name = "server-2-unique"
         mock_server_config2.url = "https://server2.example.com/api"
+        mock_server_config2.headers = None
 
         mock_server_config3 = Mock()
         mock_server_config3.mcp_server_name = "server-3"
         mock_server_config3.mcp_server_unique_name = "server-3-unique"
         mock_server_config3.url = "https://server3.example.com/api"
+        mock_server_config3.headers = None
 
         # Create unique mock clients for each server
         mock_clients = [MagicMock() for _ in range(3)]
@@ -661,6 +675,7 @@ async def test_cleanup_called_twice_after_creating_clients(
         mock_server_config.mcp_server_name = "test-server"
         mock_server_config.mcp_server_unique_name = "test-server-unique"
         mock_server_config.url = "https://test.example.com/api"
+        mock_server_config.headers = None  # per-audience headers attached by list_tool_servers
 
         mock_http_client_instance = MagicMock()