From 02a08fdbd04cc10d507b38fe0fa8bbf468f07839 Mon Sep 17 00:00:00 2001 From: Johan Broberg Date: Sat, 15 Nov 2025 13:15:47 -0800 Subject: [PATCH 1/2] Add auth_handler_name parameter to Authorization.exchange_token call --- .../services/mcp_tool_registration_service.py | 4 +++- .../services/mcp_tool_registration_service.py | 6 +++++- .../extensions/openai/mcp_tool_registration_service.py | 8 ++++++-- .../services/mcp_tool_registration_service.py | 6 +++++- 4 files changed, 19 insertions(+), 5 deletions(-) diff --git a/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py index 4e0727f0..e87da87e 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py @@ -47,6 +47,7 @@ async def add_tool_servers_to_agent( initial_tools: List[Any], agentic_app_id: str, auth: Authorization, + auth_handler_name: str, turn_context: TurnContext, auth_token: Optional[str] = None, ) -> Optional[ChatAgent]: @@ -59,6 +60,7 @@ async def add_tool_servers_to_agent( initial_tools: List of initial tools to add to the agent agentic_app_id: Agentic app identifier for the agent auth: Authorization context for token exchange + auth_handler_name: Name of the authorization handler. turn_context: Turn context for the operation auth_token: Optional bearer token for authentication @@ -69,7 +71,7 @@ async def add_tool_servers_to_agent( # Exchange token if not provided if not auth_token: scopes = get_mcp_platform_authentication_scope() - authToken = await auth.exchange_token(turn_context, scopes, "AGENTIC") + authToken = await auth.exchange_token(turn_context, scopes, auth_handler_name) auth_token = authToken.token self._logger.info(f"Listing MCP tool servers for agent {agentic_app_id}") diff --git a/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py index f534e9fd..2c40077e 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py @@ -71,6 +71,7 @@ async def add_tool_servers_to_agent( project_client: "AIProjectClient", agentic_app_id: str, auth: Authorization, + auth_handler_name: str, context: TurnContext, auth_token: Optional[str] = None, ) -> None: @@ -80,6 +81,9 @@ async def add_tool_servers_to_agent( Args: project_client: The Azure Foundry AIProjectClient instance. agentic_app_id: Agentic App ID for the agent. + auth: Authorization handler for token exchange. + auth_handler_name: Name of the authorization handler. + context: Turn context for the current operation. auth_token: Authentication token to access the MCP servers. Raises: @@ -91,7 +95,7 @@ async def add_tool_servers_to_agent( if not auth_token: scopes = get_mcp_platform_authentication_scope() - authToken = await auth.exchange_token(context, scopes, "AGENTIC") + authToken = await auth.exchange_token(context, scopes, auth_handler_name) auth_token = authToken.token try: diff --git a/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py index 4d8c6866..ba319c7d 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py @@ -52,6 +52,7 @@ async def add_tool_servers_to_agent( agent: Agent, agentic_app_id: str, auth: Authorization, + auth_handler_name: str, context: TurnContext, auth_token: Optional[str] = None, ): @@ -65,7 +66,10 @@ async def add_tool_servers_to_agent( Args: agent: The existing agent to add servers to agentic_app_id: Agentic App ID for the agent - auth_token: Authentication token to access the MCP servers + auth: Authorization handler for token exchange. + auth_handler_name: Name of the authorization handler. + context: Turn context for the current operation. + auth_token: Authentication token to access the MCP servers. Returns: New Agent instance with all MCP servers, or original agent if no new servers @@ -73,7 +77,7 @@ async def add_tool_servers_to_agent( if not auth_token: scopes = get_mcp_platform_authentication_scope() - authToken = await auth.exchange_token(context, scopes, "AGENTIC") + authToken = await auth.exchange_token(context, scopes, auth_handler_name) auth_token = authToken.token # Get MCP server configurations from the configuration service diff --git a/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py index 578e2bce..0cf3a2a9 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py @@ -79,6 +79,7 @@ async def add_tool_servers_to_agent( kernel: sk.Kernel, agentic_app_id: str, auth: Authorization, + auth_handler_name: str, context: TurnContext, auth_token: Optional[str] = None, ) -> None: @@ -88,6 +89,9 @@ async def add_tool_servers_to_agent( Args: kernel: The Semantic Kernel instance to which the tools will be added. agentic_app_id: Agentic App ID for the agent. + auth: Authorization handler for token exchange. + auth_handler_name: Name of the authorization handler. + context: Turn context for the current operation. auth_token: Authentication token to access the MCP servers. Raises: @@ -97,7 +101,7 @@ async def add_tool_servers_to_agent( if not auth_token: scopes = get_mcp_platform_authentication_scope() - authToken = await auth.exchange_token(context, scopes, "AGENTIC") + authToken = await auth.exchange_token(context, scopes, auth_handler_name) auth_token = authToken.token self._validate_inputs(kernel, agentic_app_id, auth_token) From 5644a684b515782158cb4e0e5ca66b21b17da763 Mon Sep 17 00:00:00 2001 From: Johan Broberg Date: Sat, 15 Nov 2025 15:29:13 -0800 Subject: [PATCH 2/2] Add support for resolving agent identity. --- .../microsoft_agents_a365/runtime/__init__.py | 2 + .../microsoft_agents_a365/runtime/utility.py | 82 +++++++++ .../pyproject.toml | 1 + .../services/mcp_tool_registration_service.py | 5 +- .../services/mcp_tool_registration_service.py | 4 +- .../openai/mcp_tool_registration_service.py | 4 +- .../services/mcp_tool_registration_service.py | 4 +- tests/runtime/test_utility.py | 159 ++++++++++++++++++ 8 files changed, 253 insertions(+), 8 deletions(-) create mode 100644 libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py create mode 100644 tests/runtime/test_utility.py diff --git a/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/__init__.py b/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/__init__.py index c6a7c99c..24a54ef8 100644 --- a/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/__init__.py +++ b/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/__init__.py @@ -2,11 +2,13 @@ from .environment_utils import get_observability_authentication_scope from .power_platform_api_discovery import ClusterCategory, PowerPlatformApiDiscovery +from .utility import Utility __all__ = [ "get_observability_authentication_scope", "PowerPlatformApiDiscovery", "ClusterCategory", + "Utility", ] __path__ = __import__("pkgutil").extend_path(__path__, __name__) diff --git a/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py b/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py new file mode 100644 index 00000000..3e93f631 --- /dev/null +++ b/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py @@ -0,0 +1,82 @@ +# Copyright (c) Microsoft. All rights reserved. + +""" +Utility functions for Microsoft Agent 365 runtime operations. + +This module provides utility functions for token handling, agent identity resolution, +and other common runtime operations. +""" + +from __future__ import annotations + +import uuid +from typing import Any, Optional + +import jwt + + +class Utility: + """ + Utility class providing common runtime operations for Agent 365. + + This class contains static methods for token processing, agent identity resolution, + and other utility functions used across the Agent 365 runtime. + """ + + @staticmethod + def get_app_id_from_token(token: Optional[str]) -> str: + """ + Decodes the current token and retrieves the App ID (appid or azp claim). + + Args: + token: JWT token to decode. Can be None or empty. + + Returns: + str: The App ID from the token's claims, or empty GUID if token is invalid. + Returns "00000000-0000-0000-0000-000000000000" if no valid App ID is found. + """ + if not token or not token.strip(): + return str(uuid.UUID(int=0)) + + try: + # Decode the JWT token without verification (we only need the claims) + # Note: verify=False is used because we only need to extract claims, + # not verify the token's authenticity + decoded_payload = jwt.decode(token, options={"verify_signature": False}) + + # Look for appid or azp claims (appid takes precedence) + app_id = decoded_payload.get("appid") or decoded_payload.get("azp") + return app_id if app_id else "" + + except (jwt.DecodeError, jwt.InvalidTokenError): + # Token is malformed or invalid + return "" + + @staticmethod + def resolve_agent_identity(context: Any, auth_token: Optional[str]) -> str: + """ + Resolves the agent identity from the turn context or auth token. + + Args: + context: Turn context of the conversation turn. Expected to have an Activity + with methods like is_agentic_request() and get_agentic_instance_id(). + auth_token: Authentication token if available. + + Returns: + str: The agent identity (App ID). Returns the agentic instance ID if the + request is agentic, otherwise returns the App ID from the auth token. + """ + try: + # App ID is required to pass to MCP server URL + # Try to get agentic instance ID if this is an agentic request + if context and context.activity and context.activity.is_agentic_request(): + agentic_id = context.activity.get_agentic_instance_id() + return agentic_id if agentic_id else "" + + except (AttributeError, TypeError, Exception): + # Context/activity doesn't have the expected methods or properties + # or any other error occurred while accessing context/activity + pass + + # Fallback to extracting App ID from the auth token + return Utility.get_app_id_from_token(auth_token) diff --git a/libraries/microsoft-agents-a365-runtime/pyproject.toml b/libraries/microsoft-agents-a365-runtime/pyproject.toml index 0c16df7a..c994444c 100644 --- a/libraries/microsoft-agents-a365-runtime/pyproject.toml +++ b/libraries/microsoft-agents-a365-runtime/pyproject.toml @@ -25,6 +25,7 @@ classifiers = [ license = {text = "MIT"} keywords = ["observability", "telemetry", "tracing", "opentelemetry", "monitoring", "ai", "agents"] dependencies = [ + "PyJWT >= 2.8.0", ] [project.urls] diff --git a/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py index e87da87e..2b2ea04f 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py @@ -9,6 +9,7 @@ from microsoft_agents.hosting.core import Authorization, TurnContext +from microsoft_agents_a365.runtime.utility import Utility from microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service import ( McpToolServerConfigurationService, ) @@ -45,7 +46,6 @@ async def add_tool_servers_to_agent( chat_client: Union[OpenAIChatClient, AzureOpenAIChatClient], agent_instructions: str, initial_tools: List[Any], - agentic_app_id: str, auth: Authorization, auth_handler_name: str, turn_context: TurnContext, @@ -58,7 +58,6 @@ async def add_tool_servers_to_agent( chat_client: The chat client instance (Union[OpenAIChatClient, AzureOpenAIChatClient]) agent_instructions: Instructions for the agent behavior initial_tools: List of initial tools to add to the agent - agentic_app_id: Agentic app identifier for the agent auth: Authorization context for token exchange auth_handler_name: Name of the authorization handler. turn_context: Turn context for the operation @@ -74,6 +73,8 @@ async def add_tool_servers_to_agent( authToken = await auth.exchange_token(turn_context, scopes, auth_handler_name) auth_token = authToken.token + agentic_app_id = Utility.resolve_agent_identity(turn_context, auth_token) + self._logger.info(f"Listing MCP tool servers for agent {agentic_app_id}") # Get MCP server configurations diff --git a/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py index 2c40077e..22816ce1 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py @@ -17,6 +17,7 @@ from azure.identity import DefaultAzureCredential from azure.ai.agents.models import McpTool, ToolResources from microsoft_agents.hosting.core import Authorization, TurnContext +from microsoft_agents_a365.runtime.utility import Utility from microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service import ( McpToolServerConfigurationService, ) @@ -69,7 +70,6 @@ def __init__( async def add_tool_servers_to_agent( self, project_client: "AIProjectClient", - agentic_app_id: str, auth: Authorization, auth_handler_name: str, context: TurnContext, @@ -80,7 +80,6 @@ async def add_tool_servers_to_agent( Args: project_client: The Azure Foundry AIProjectClient instance. - agentic_app_id: Agentic App ID for the agent. auth: Authorization handler for token exchange. auth_handler_name: Name of the authorization handler. context: Turn context for the current operation. @@ -99,6 +98,7 @@ async def add_tool_servers_to_agent( auth_token = authToken.token try: + agentic_app_id = Utility.resolve_agent_identity(context, auth_token) # Get the tool definitions and resources using the async implementation tool_definitions, tool_resources = await self._get_mcp_tool_definitions_and_resources( agentic_app_id, auth_token or "" diff --git a/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py index ba319c7d..eca9bede 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py @@ -12,6 +12,7 @@ MCPServerStreamableHttp, MCPServerStreamableHttpParams, ) +from microsoft_agents_a365.runtime.utility import Utility from microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service import ( McpToolServerConfigurationService, ) @@ -50,7 +51,6 @@ def __init__(self, logger: Optional[logging.Logger] = None): async def add_tool_servers_to_agent( self, agent: Agent, - agentic_app_id: str, auth: Authorization, auth_handler_name: str, context: TurnContext, @@ -65,7 +65,6 @@ async def add_tool_servers_to_agent( Args: agent: The existing agent to add servers to - agentic_app_id: Agentic App ID for the agent auth: Authorization handler for token exchange. auth_handler_name: Name of the authorization handler. context: Turn context for the current operation. @@ -84,6 +83,7 @@ async def add_tool_servers_to_agent( # mcp_server_configs = [] # TODO: radevika: Update once the common project is merged. + agentic_app_id = Utility.resolve_agent_identity(context, auth_token) self._logger.info(f"Listing MCP tool servers for agent {agentic_app_id}") mcp_server_configs = await self.config_service.list_tool_servers( agentic_app_id=agentic_app_id, diff --git a/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py b/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py index 0cf3a2a9..836b27a1 100644 --- a/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py +++ b/libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py @@ -16,6 +16,7 @@ from semantic_kernel import kernel as sk from semantic_kernel.connectors.mcp import MCPStreamableHttpPlugin from microsoft_agents.hosting.core import Authorization, TurnContext +from microsoft_agents_a365.runtime.utility import Utility from microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service import ( McpToolServerConfigurationService, ) @@ -77,7 +78,6 @@ def __init__( async def add_tool_servers_to_agent( self, kernel: sk.Kernel, - agentic_app_id: str, auth: Authorization, auth_handler_name: str, context: TurnContext, @@ -88,7 +88,6 @@ async def add_tool_servers_to_agent( Args: kernel: The Semantic Kernel instance to which the tools will be added. - agentic_app_id: Agentic App ID for the agent. auth: Authorization handler for token exchange. auth_handler_name: Name of the authorization handler. context: Turn context for the current operation. @@ -104,6 +103,7 @@ async def add_tool_servers_to_agent( authToken = await auth.exchange_token(context, scopes, auth_handler_name) auth_token = authToken.token + agentic_app_id = Utility.resolve_agent_identity(context, auth_token) self._validate_inputs(kernel, agentic_app_id, auth_token) # Get and process servers diff --git a/tests/runtime/test_utility.py b/tests/runtime/test_utility.py new file mode 100644 index 00000000..d6cd2d4f --- /dev/null +++ b/tests/runtime/test_utility.py @@ -0,0 +1,159 @@ +# Copyright (c) Microsoft. All rights reserved. + +import unittest +import uuid +import jwt + +from microsoft_agents_a365.runtime.utility import Utility + + +class TestUtility(unittest.TestCase): + """Test cases for the Utility class.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_app_id = "12345678-1234-1234-1234-123456789abc" + self.test_azp_id = "87654321-4321-4321-4321-cba987654321" + + def create_test_jwt(self, claims: dict) -> str: + """Create a test JWT token with the given claims.""" + # Use PyJWT to create a proper JWT token (unsigned for testing) + return jwt.encode(claims, key="", algorithm="none") + + def test_get_app_id_from_token_with_none_token(self): + """Test get_app_id_from_token with None token.""" + result = Utility.get_app_id_from_token(None) + self.assertEqual(result, str(uuid.UUID(int=0))) + + def test_get_app_id_from_token_with_empty_token(self): + """Test get_app_id_from_token with empty token.""" + result = Utility.get_app_id_from_token("") + self.assertEqual(result, str(uuid.UUID(int=0))) + + result = Utility.get_app_id_from_token(" ") + self.assertEqual(result, str(uuid.UUID(int=0))) + + def test_get_app_id_from_token_with_appid_claim(self): + """Test get_app_id_from_token with appid claim.""" + token = self.create_test_jwt({"appid": self.test_app_id, "other": "value"}) + result = Utility.get_app_id_from_token(token) + self.assertEqual(result, self.test_app_id) + + def test_get_app_id_from_token_with_azp_claim(self): + """Test get_app_id_from_token with azp claim.""" + token = self.create_test_jwt({"azp": self.test_azp_id, "other": "value"}) + result = Utility.get_app_id_from_token(token) + self.assertEqual(result, self.test_azp_id) + + def test_get_app_id_from_token_with_both_claims(self): + """Test get_app_id_from_token with both appid and azp claims (appid takes precedence).""" + token = self.create_test_jwt({"appid": self.test_app_id, "azp": self.test_azp_id}) + result = Utility.get_app_id_from_token(token) + self.assertEqual(result, self.test_app_id) + + def test_get_app_id_from_token_without_app_claims(self): + """Test get_app_id_from_token with token containing no app claims.""" + token = self.create_test_jwt({"sub": "user123", "iss": "issuer"}) + result = Utility.get_app_id_from_token(token) + self.assertEqual(result, "") + + def test_get_app_id_from_token_with_invalid_token(self): + """Test get_app_id_from_token with invalid token formats.""" + # Invalid token format + result = Utility.get_app_id_from_token("invalid.token") + self.assertEqual(result, "") + + # Token with only two parts + result = Utility.get_app_id_from_token("header.payload") + self.assertEqual(result, "") + + # Token with invalid base64 + result = Utility.get_app_id_from_token("invalid.!!!invalid!!!.signature") + self.assertEqual(result, "") + + +class MockActivity: + """Mock activity class for testing.""" + + def __init__(self, is_agentic: bool = False, agentic_id: str = ""): + self._is_agentic = is_agentic + self._agentic_id = agentic_id + + def is_agentic_request(self) -> bool: + return self._is_agentic + + def get_agentic_instance_id(self) -> str: + return self._agentic_id + + +class MockContext: + """Mock context class for testing.""" + + def __init__(self, activity=None): + self.activity = activity + + +class TestUtilityResolveAgentIdentity(unittest.TestCase): + """Test cases for the resolve_agent_identity method.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_app_id = "token-app-id-123" + self.agentic_id = "agentic-id-456" + + # Create a test token with PyJWT + claims = {"appid": self.test_app_id} + self.test_token = jwt.encode(claims, key="", algorithm="none") + + def test_resolve_agent_identity_with_agentic_request(self): + """Test resolve_agent_identity with agentic request.""" + activity = MockActivity(is_agentic=True, agentic_id=self.agentic_id) + context = MockContext(activity) + + result = Utility.resolve_agent_identity(context, self.test_token) + self.assertEqual(result, self.agentic_id) + + def test_resolve_agent_identity_with_non_agentic_request(self): + """Test resolve_agent_identity with non-agentic request.""" + activity = MockActivity(is_agentic=False) + context = MockContext(activity) + + result = Utility.resolve_agent_identity(context, self.test_token) + self.assertEqual(result, self.test_app_id) + + def test_resolve_agent_identity_with_context_without_activity(self): + """Test resolve_agent_identity with context that has no activity.""" + context = MockContext() + + result = Utility.resolve_agent_identity(context, self.test_token) + self.assertEqual(result, self.test_app_id) + + def test_resolve_agent_identity_with_none_context(self): + """Test resolve_agent_identity with None context.""" + result = Utility.resolve_agent_identity(None, self.test_token) + self.assertEqual(result, self.test_app_id) + + def test_resolve_agent_identity_with_agentic_but_empty_id(self): + """Test resolve_agent_identity with agentic request but empty agentic ID.""" + activity = MockActivity(is_agentic=True, agentic_id="") + context = MockContext(activity) + + result = Utility.resolve_agent_identity(context, self.test_token) + self.assertEqual(result, "") + + def test_resolve_agent_identity_fallback_on_exception(self): + """Test resolve_agent_identity falls back to token when context access fails.""" + + # Create a context that will raise an exception when accessed + class FaultyContext: + @property + def activity(self): + raise RuntimeError("Context access failed") + + context = FaultyContext() + result = Utility.resolve_agent_identity(context, self.test_token) + self.assertEqual(result, self.test_app_id) + + +if __name__ == "__main__": + unittest.main()