diff --git a/docs/prd-x-ms-agentid-header.md b/docs/prd-x-ms-agentid-header.md new file mode 100644 index 00000000..c0f47068 --- /dev/null +++ b/docs/prd-x-ms-agentid-header.md @@ -0,0 +1,262 @@ +# PRD: x-ms-agentId Header for MCP Platform Calls + +## Overview + +Add an `x-ms-agentid` header to all outbound HTTP requests from the tooling package to the MCP platform. This header identifies the calling agent using the best available identifier. + +## Problem Statement + +The MCP platform needs to identify which agent is making tooling requests for: +- Logging and diagnostics +- Usage analytics + +Currently, no consistent agent identifier is sent with MCP platform requests. + +## Requirements + +### Functional Requirements + +1. All HTTP requests to the MCP platform SHALL include the `x-ms-agentid` header +2. The header value SHALL be determined using the following priority: + 1. **Agent Blueprint ID** from TurnContext (highest priority) + 2. **Token Claims** - `xms_par_app_azp` (agent blueprint ID) > `appid` > `azp` + 3. **Application Name** from environment or pyproject.toml (lowest priority fallback) +3. If no identifier is available, the header SHOULD be omitted (not sent with empty value) + +### Non-Functional Requirements + +1. No additional network calls to retrieve identifiers +2. Minimal performance impact on existing flows +3. Backward compatible - existing integrations continue to work + +## Technical Design + +### Affected Components + +| Package | File | Change | +|---------|------|--------| +| `microsoft-agents-a365-runtime` | `utility.py` | Add `get_agent_id_from_token()` (checks `xms_par_app_azp` → `appid` → `azp`) | +| `microsoft-agents-a365-runtime` | `utility.py` | Add `get_application_name()` helper (reads package name) | +| `microsoft-agents-a365-tooling` | `constants.py` | Add `HEADER_AGENT_ID` and `HEADER_CHANNEL_ID` constants | +| `microsoft-agents-a365-tooling` | `mcp_tool_server_configuration_service.py` | Update `_prepare_gateway_headers()` to include `x-ms-agentid` | + +### Identifier Retrieval Strategy + +#### 1. Agent Blueprint ID (Highest Priority) + +**Source**: `turn_context.activity.from_.agentic_app_blueprint_id` + +**Availability**: Only available in agentic request scenarios where a `TurnContext` is present and the request originates from another agent. + +**Format**: GUID (e.g., `12345678-1234-1234-1234-123456789abc`) + +--- + +#### 2 & 3. Agent ID from Token (Second/Third Priority) + +**Sources** (checked in order): +1. `xms_par_app_azp` claim - Agent Blueprint ID (parent application's Azure app ID) +2. `appid` or `azp` claim - Entra Application ID + +**Availability**: Available when an `auth_token` is provided to the tooling methods. + +**Implementation**: + +```python +# libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py +@staticmethod +def get_agent_id_from_token(token: Optional[str]) -> str: + """ + Decodes the token and retrieves the best available agent identifier. + Checks claims in priority order: xms_par_app_azp > appid > azp. + + Returns empty string for empty/missing tokens (unlike get_app_id_from_token + which returns a default GUID). + """ + if not token or not token.strip(): + return "" + + try: + decoded = jwt.decode(token, options={"verify_signature": False}) + # Priority: xms_par_app_azp (agent blueprint ID) > appid > azp + return decoded.get("xms_par_app_azp") or decoded.get("appid") or decoded.get("azp") or "" + except (jwt.DecodeError, jwt.InvalidTokenError): + return "" +``` + +**Format**: GUID (e.g., `12345678-1234-1234-1234-123456789abc`) + +--- + +#### 4. Application Name (Lowest Priority Fallback) + +**Source**: Application's pyproject.toml `name` field or environment variable + +**Strategy**: +1. Check `AGENT365_APPLICATION_NAME` environment variable +2. Fall back to reading and caching the application's `pyproject.toml` name field +3. If neither available, omit the header + +**Implementation**: + +```python +# libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py +_cached_application_name: Optional[str] = None + +@staticmethod +def get_application_name() -> Optional[str]: + """Gets the application name from environment or pyproject.toml.""" + # First try environment variable + env_name = os.environ.get("AGENT365_APPLICATION_NAME") + if env_name: + return env_name + + # Fall back to cached pyproject.toml name + if Utility._cached_application_name is None: + Utility._cached_application_name = Utility._read_application_name() + + return Utility._cached_application_name or None +``` + +--- + +### Implementation + +#### Updated Header Preparation + +```python +# libraries/microsoft-agents-a365-tooling/.../mcp_tool_server_configuration_service.py +def _prepare_gateway_headers( + self, + auth_token: str, + turn_context: Optional[TurnContext], + options: ToolOptions +) -> Dict[str, str]: + """Prepares headers for tooling gateway requests.""" + headers = { + Constants.Headers.AUTHORIZATION: f"{Constants.Headers.BEARER_PREFIX} {auth_token}", + Constants.Headers.USER_AGENT: RuntimeUtility.get_user_agent_header( + options.orchestrator_name + ), + } + + # Add x-ms-agentid header with priority fallback + agent_id = self._resolve_agent_id_for_header(auth_token, turn_context) + if agent_id: + headers[Constants.Headers.AGENT_ID] = agent_id + + return headers + +def _resolve_agent_id_for_header( + self, + auth_token: str, + turn_context: Optional[TurnContext] +) -> Optional[str]: + """Resolves the best available agent identifier for the x-ms-agentid header.""" + # Priority 1: Agent Blueprint ID from TurnContext + try: + if turn_context and turn_context.activity and turn_context.activity.from_: + blueprint_id = getattr(turn_context.activity.from_, 'agentic_app_blueprint_id', None) + if blueprint_id: + return blueprint_id + except (AttributeError, TypeError): + pass + + # Priority 2 & 3: Agent ID from token (xms_par_app_azp > appid > azp) + agent_id = RuntimeUtility.get_agent_id_from_token(auth_token) + if agent_id: + return agent_id + + # Priority 4: Application name + return RuntimeUtility.get_application_name() +``` + +### Call Sites Summary + +| Call Site | auth_token | turn_context | Gets `x-ms-agentid`? | +|-----------|-----------|-------------|----------------------| +| `_load_servers_from_gateway()` | ✅ | ❌ (None currently) | ✅ Yes (from token/app name) | +| `send_chat_history()` | ❌ | ✅ | ❌ No (authToken required) | + +**Note**: The `x-ms-agentid` header is only added when `auth_token` is present. `send_chat_history()` does not pass an auth token, so it won't include this header. + +--- + +## Open Questions + +### Q1: Application Name Strategy ✅ RESOLVED + +**Decision**: Use `AGENT365_APPLICATION_NAME` environment variable as primary, with pyproject.toml fallback. Cache the pyproject.toml read to avoid repeated file I/O. + +### Q2: Header Name Casing ✅ RESOLVED + +**Decision**: Use `x-ms-agentid` (all lowercase, case insensitive). + +HTTP headers are case-insensitive per RFC 7230, so the server will accept any casing. Using lowercase is the conventional choice. + +### Q3: TurnContext Availability ✅ RESOLVED + +**Decision**: For this initial implementation, we will pass `None` for turn_context in `_load_servers_from_gateway()` since the current `list_tool_servers` signature doesn't accept it. The agent ID will be resolved from the token or application name. + +A future enhancement can add a new overloaded signature similar to the Node.js SDK that accepts `TurnContext`. + +--- + +## Testing Strategy + +### Unit Tests + +1. Test `get_agent_id_from_token()` with each priority level: + - Token with `xms_par_app_azp` → returns blueprint ID from token + - No `xms_par_app_azp`, token with `appid` → returns Entra app ID + - No `appid`, token with `azp` → returns azp claim + - No token claims → returns empty string + - Empty/invalid token → returns empty string +2. Test `get_application_name()`: + - Returns env var when set + - Returns pyproject.toml name when env not set + - Returns None when nothing available + - Caches the result +3. Test `_resolve_agent_id_for_header()`: + - TurnContext with `agentic_app_blueprint_id` → returns blueprint ID + - No blueprint ID, token with claims → returns token claim + - No claims → returns application name + - Nothing available → returns None +4. Test `_prepare_gateway_headers()`: + - Includes `x-ms-agentid` when identifier available + - Omits header when no identifier available + +### Integration Tests + +1. Verify header is sent in `list_tool_servers()` requests +2. Verify header is NOT sent in `send_chat_history()` requests (no authToken) + +--- + +## Breaking Changes + +**None** - This implementation is fully backward compatible. + +--- + +## Rollout Plan + +1. **Phase 1**: Add utility methods and `x-ms-agentid` header (this PR) +2. **Phase 2**: Add overloaded `list_tool_servers()` signature with TurnContext (future) +3. **Phase 3**: Update documentation and samples + +--- + +## Dependencies + +- Runtime package for token utility (already exists) +- PyJWT library (already a dependency) +- No new external dependencies required + +--- + +## Success Metrics + +1. 100% of MCP platform requests include `x-ms-agentid` header (when identifier available) +2. No increase in request latency +3. No breaking changes for existing consumers 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 index 6d02505a..b4ace0c0 100644 --- a/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py +++ b/libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/utility.py @@ -10,9 +10,13 @@ from __future__ import annotations +import os import platform +import re +import threading import uuid from importlib.metadata import PackageNotFoundError, version +from pathlib import Path from typing import Any, Optional import jwt @@ -26,13 +30,25 @@ class Utility: and other utility functions used across the Agent 365 runtime. """ - _cached_version = None + _cached_version: Optional[str] = None + _cached_application_name: Optional[str] = None + _application_name_initialized: bool = False + _cache_lock: threading.Lock = threading.Lock() @staticmethod def get_app_id_from_token(token: Optional[str]) -> str: """ Decodes the current token and retrieves the App ID (appid or azp claim). + **WARNING: NO SIGNATURE VERIFICATION** - This method uses jwt.decode() which does NOT + verify the token signature. The token claims can be spoofed by malicious actors. + This method is ONLY suitable for logging, analytics, and diagnostics purposes. + Do NOT use the returned value for authorization, access control, or security decisions. + + Note: Returns a default GUID ('00000000-0000-0000-0000-000000000000') for empty tokens + for backward compatibility with callers that expect a valid-looking GUID. + For agent identification where empty string is preferred, use get_agent_id_from_token(). + Args: token: JWT token to decode. Can be None or empty. @@ -57,6 +73,44 @@ def get_app_id_from_token(token: Optional[str]) -> str: # Token is malformed or invalid return "" + @staticmethod + def get_agent_id_from_token(token: Optional[str]) -> str: + """ + Decodes the token and retrieves the best available agent identifier. + Checks claims in priority order: xms_par_app_azp (agent blueprint ID) > appid > azp. + + **WARNING: NO SIGNATURE VERIFICATION** - This method uses jwt.decode() which does NOT + verify the token signature. The token claims can be spoofed by malicious actors. + This method is ONLY suitable for logging, analytics, and diagnostics purposes. + Do NOT use the returned value for authorization, access control, or security decisions. + + Note: Returns empty string for empty/missing tokens (unlike get_app_id_from_token() which + returns a default GUID). This allows callers to omit headers when no identifier is available. + + Args: + token: JWT token to decode. Can be None or empty. + + Returns: + str: Agent ID (GUID) or empty string if not found or token is empty. + """ + if not token or not token.strip(): + return "" + + try: + decoded_payload = jwt.decode(token, options={"verify_signature": False}) + + # Priority: xms_par_app_azp (agent blueprint ID) > appid > azp + return ( + decoded_payload.get("xms_par_app_azp") + or decoded_payload.get("appid") + or decoded_payload.get("azp") + or "" + ) + + except (jwt.DecodeError, jwt.InvalidTokenError): + # Silent error handling - return empty string on decode failure + return "" + @staticmethod def resolve_agent_identity(context: Any, auth_token: Optional[str]) -> str: """ @@ -109,3 +163,87 @@ def get_user_agent_header(orchestrator: Optional[str] = None) -> str: os_type = platform.system() python_version = platform.python_version() return f"Agent365SDK/{Utility._cached_version} ({os_type}; Python {python_version}{orchestrator_part})" + + @staticmethod + def get_application_name() -> Optional[str]: + """ + Gets the application name from environment variable or pyproject.toml. + The pyproject.toml result is cached at first access to avoid repeated file I/O. + + Returns: + Optional[str]: Application name or None if not available. + """ + # First try environment variable (highest priority) + env_name = os.environ.get("AGENT365_APPLICATION_NAME") + if env_name: + return env_name + + # Fall back to cached pyproject.toml name with thread-safe caching + if not Utility._application_name_initialized: + with Utility._cache_lock: + # Double-checked locking pattern + if not Utility._application_name_initialized: + Utility._cached_application_name = ( + Utility._read_application_name_from_pyproject() + ) + Utility._application_name_initialized = True + + return Utility._cached_application_name + + @staticmethod + def _read_application_name_from_pyproject() -> Optional[str]: + """ + Reads the application name from pyproject.toml at the current working directory. + + Note: Uses Path.cwd() which assumes the application is started from its root directory. + This is a fallback mechanism - AGENT365_APPLICATION_NAME env var is the preferred source. + + Returns: + Optional[str]: Application name from pyproject.toml or None if not found. + """ + # Regex pattern to match: name = "value" or name = 'value' + # This handles exact field name matching and ignores comments + name_pattern = re.compile(r'^\s*name\s*=\s*["\']([^"\']*)["\']') + + try: + pyproject_path = Path.cwd() / "pyproject.toml" + if not pyproject_path.exists(): + return None + + content = pyproject_path.read_text(encoding="utf-8") + + # Simple TOML parsing for [project] name = "..." + # We avoid importing tomli/tomllib for this simple case + in_project_section = False + for line in content.splitlines(): + stripped = line.strip() + if stripped == "[project]": + in_project_section = True + continue + elif stripped.startswith("[") and stripped.endswith("]"): + in_project_section = False + continue + + if in_project_section: + # Use regex to properly parse name = "value" with exact field matching + match = name_pattern.match(stripped) + if match: + value = match.group(1) + if value: + return value + + return None + + except (OSError, ValueError): + # File read errors or parsing errors + return None + + @staticmethod + def reset_application_name_cache() -> None: + """ + Resets the cached application name. Used for testing purposes. + + This method is intended for internal testing only. + """ + Utility._cached_application_name = None + Utility._application_name_initialized = False diff --git a/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/services/mcp_tool_server_configuration_service.py b/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/services/mcp_tool_server_configuration_service.py index 4e93f830..c56ab4bb 100644 --- a/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/services/mcp_tool_server_configuration_service.py +++ b/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/services/mcp_tool_server_configuration_service.py @@ -352,24 +352,74 @@ async def _load_servers_from_gateway( return mcp_servers - def _prepare_gateway_headers(self, auth_token: str, options: ToolOptions) -> Dict[str, str]: + def _prepare_gateway_headers( + self, auth_token: str, options: ToolOptions, turn_context: Optional[TurnContext] = None + ) -> Dict[str, str]: """ Prepares headers for tooling gateway requests. Args: auth_token: Authentication token. options: ToolOptions instance containing optional parameters. + turn_context: Optional TurnContext for extracting agent blueprint ID for request headers. Returns: Dictionary of HTTP headers. """ - return { + headers: Dict[str, str] = { Constants.Headers.AUTHORIZATION: f"{Constants.Headers.BEARER_PREFIX} {auth_token}", Constants.Headers.USER_AGENT: RuntimeUtility.get_user_agent_header( options.orchestrator_name ), } + # Add x-ms-agentid header with priority fallback + agent_id = self._resolve_agent_id_for_header(auth_token, turn_context) + if agent_id: + headers[Constants.Headers.AGENT_ID] = agent_id + + return headers + + def _resolve_agent_id_for_header( + self, auth_token: str, turn_context: Optional[TurnContext] = None + ) -> Optional[str]: + """ + Resolves the best available agent identifier for the x-ms-agentid header. + Priority: TurnContext.agenticAppBlueprintId > token claims (xms_par_app_azp > appid > azp) + > application name + + Note: This differs from RuntimeUtility.resolve_agent_identity() which resolves the agenticAppId + for URL construction. This method resolves the identifier specifically for the x-ms-agentid header. + + Args: + auth_token: The authentication token to extract claims from. + turn_context: Optional TurnContext to extract agent blueprint ID from. + + Returns: + Agent ID string or None if not available. + """ + # Priority 1: Agent Blueprint ID from TurnContext + # The 'from_' property may include agentic_app_blueprint_id when the request originates + # from an agentic app + try: + if turn_context and turn_context.activity and turn_context.activity.from_: + blueprint_id = getattr( + turn_context.activity.from_, "agentic_app_blueprint_id", None + ) + if blueprint_id: + return blueprint_id + except (AttributeError, TypeError): + pass + + # Priority 2 & 3: Agent ID from token (xms_par_app_azp > appid > azp) + # Single decode, checks claims in priority order + agent_id = RuntimeUtility.get_agent_id_from_token(auth_token) + if agent_id: + return agent_id + + # Priority 4: Application name from AGENT365_APPLICATION_NAME env or pyproject.toml + return RuntimeUtility.get_application_name() + async def _parse_gateway_response( self, response: aiohttp.ClientResponse ) -> List[MCPServerConfig]: diff --git a/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/utils/constants.py b/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/utils/constants.py index c788447d..d1ff406c 100644 --- a/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/utils/constants.py +++ b/libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/utils/constants.py @@ -24,3 +24,12 @@ class Headers: #: The header name for User-Agent information. USER_AGENT = "User-Agent" + + #: Header name for sending the agent identifier to MCP platform for logging/analytics. + AGENT_ID = "x-ms-agentid" + + #: Header name for the channel ID. + CHANNEL_ID = "x-ms-channel-id" + + #: Header name for the subchannel ID. + SUBCHANNEL_ID = "x-ms-subchannel-id" diff --git a/tests/runtime/test_utility.py b/tests/runtime/test_utility.py index 54ad232d..dbfbe11f 100644 --- a/tests/runtime/test_utility.py +++ b/tests/runtime/test_utility.py @@ -3,10 +3,12 @@ """Unit tests for Utility class.""" +import os import platform import re import uuid -from unittest.mock import Mock +from pathlib import Path +from unittest.mock import Mock, patch import jwt import pytest @@ -49,6 +51,14 @@ def _create(activity=None): return _create +@pytest.fixture(autouse=True) +def reset_application_name_cache(): + """Reset the application name cache before each test.""" + Utility.reset_application_name_cache() + yield + Utility.reset_application_name_cache() + + # Tests for get_app_id_from_token @pytest.mark.parametrize( "token,expected", @@ -151,3 +161,196 @@ def test_get_user_agent_header_with_orchestrator(): # Regex for Agent365SDK/version (OS; Python version; TestOrchestrator) pattern = rf"^Agent365SDK/.+ \({os_type}; Python {py_version}; {orchestrator}\)$" assert re.match(pattern, result) + + +# Tests for get_agent_id_from_token +class TestGetAgentIdFromToken: + """Tests for the get_agent_id_from_token method.""" + + def test_empty_token_returns_empty_string(self): + """Test that empty token returns empty string (not default GUID).""" + assert Utility.get_agent_id_from_token("") == "" + + def test_none_token_returns_empty_string(self): + """Test that None token returns empty string.""" + assert Utility.get_agent_id_from_token(None) == "" + + def test_whitespace_token_returns_empty_string(self): + """Test that whitespace-only token returns empty string.""" + assert Utility.get_agent_id_from_token(" ") == "" + + def test_xms_par_app_azp_takes_highest_priority(self, create_test_jwt): + """Test xms_par_app_azp claim takes priority over appid and azp.""" + token = create_test_jwt({ + "xms_par_app_azp": "blueprint-id-123", + "appid": "app-id-456", + "azp": "azp-id-789", + }) + result = Utility.get_agent_id_from_token(token) + assert result == "blueprint-id-123" + + def test_appid_takes_priority_when_no_xms_par_app_azp(self, create_test_jwt): + """Test appid claim is used when xms_par_app_azp is not present.""" + token = create_test_jwt({ + "appid": "app-id-456", + "azp": "azp-id-789", + }) + result = Utility.get_agent_id_from_token(token) + assert result == "app-id-456" + + def test_azp_used_when_no_other_claims(self, create_test_jwt): + """Test azp claim is used when xms_par_app_azp and appid are not present.""" + token = create_test_jwt({"azp": "azp-id-789"}) + result = Utility.get_agent_id_from_token(token) + assert result == "azp-id-789" + + def test_returns_empty_when_no_relevant_claims(self, create_test_jwt): + """Test returns empty string when no relevant claims are present.""" + token = create_test_jwt({"sub": "some-subject", "iss": "some-issuer"}) + result = Utility.get_agent_id_from_token(token) + assert result == "" + + def test_invalid_token_returns_empty_string(self): + """Test invalid token returns empty string.""" + result = Utility.get_agent_id_from_token("invalid.token") + assert result == "" + + def test_falls_back_to_appid_when_xms_par_app_azp_is_empty(self, create_test_jwt): + """Test falls back to appid when xms_par_app_azp is empty string.""" + token = create_test_jwt({ + "xms_par_app_azp": "", + "appid": "app-id-456", + }) + result = Utility.get_agent_id_from_token(token) + assert result == "app-id-456" + + def test_falls_back_to_azp_when_appid_is_also_empty(self, create_test_jwt): + """Test falls back to azp when both xms_par_app_azp and appid are empty.""" + token = create_test_jwt({ + "xms_par_app_azp": "", + "appid": "", + "azp": "azp-id-789", + }) + result = Utility.get_agent_id_from_token(token) + assert result == "azp-id-789" + + +# Tests for get_application_name +class TestGetApplicationName: + """Tests for the get_application_name method.""" + + def test_returns_env_var_when_set(self): + """Test returns AGENT365_APPLICATION_NAME env var when set.""" + with patch.dict(os.environ, {"AGENT365_APPLICATION_NAME": "my-test-app"}): + result = Utility.get_application_name() + assert result == "my-test-app" + + def test_env_var_takes_priority_over_pyproject(self, tmp_path): + """Test env var takes priority over pyproject.toml.""" + # Create a pyproject.toml with a different name + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text('[project]\nname = "pyproject-app-name"') + + with patch.dict(os.environ, {"AGENT365_APPLICATION_NAME": "env-app-name"}): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result == "env-app-name" + + def test_reads_from_pyproject_when_env_not_set(self, tmp_path): + """Test reads from pyproject.toml when env var is not set.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text('[project]\nname = "pyproject-app-name"') + + # Remove env var if set + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result == "pyproject-app-name" + + def test_caches_pyproject_result(self, tmp_path): + """Test that pyproject.toml is only read once.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text('[project]\nname = "cached-name"') + + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + + # First call + result1 = Utility.get_application_name() + # Modify file (but cache should prevent re-read) + pyproject.write_text('[project]\nname = "new-name"') + # Second call should return cached value + result2 = Utility.get_application_name() + + assert result1 == "cached-name" + assert result2 == "cached-name" + + def test_returns_none_when_nothing_available(self, tmp_path): + """Test returns None when no env var and no pyproject.toml.""" + # Use a directory without pyproject.toml + empty_dir = tmp_path / "empty" + empty_dir.mkdir() + + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=empty_dir): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result is None + + def test_handles_pyproject_without_name(self, tmp_path): + """Test handles pyproject.toml without name field.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text('[project]\nversion = "1.0.0"') + + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result is None + + def test_handles_pyproject_with_different_sections(self, tmp_path): + """Test correctly parses name from [project] section only.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text( + '[tool.ruff]\nname = "ruff-name"\n\n[project]\nname = "project-name"\n' + ) + + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result == "project-name" + + def test_ignores_fields_starting_with_name(self, tmp_path): + """Test only matches exact 'name' field, not 'name_something'.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text( + '[project]\nname_something = "wrong"\nnamespace = "also-wrong"\nname = "correct"\n' + ) + + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result == "correct" + + def test_handles_inline_comments(self, tmp_path): + """Test ignores inline comments after the value.""" + pyproject = tmp_path / "pyproject.toml" + pyproject.write_text('[project]\nname = "my-app" # this is a comment\n') + + env = {k: v for k, v in os.environ.items() if k != "AGENT365_APPLICATION_NAME"} + with patch.dict(os.environ, env, clear=True): + with patch.object(Path, "cwd", return_value=tmp_path): + Utility.reset_application_name_cache() + result = Utility.get_application_name() + assert result == "my-app" diff --git a/tests/tooling/test_mcp_server_configuration.py b/tests/tooling/test_mcp_server_configuration.py index 41ff26a4..619f51ec 100644 --- a/tests/tooling/test_mcp_server_configuration.py +++ b/tests/tooling/test_mcp_server_configuration.py @@ -241,3 +241,228 @@ async def test_list_tool_servers_production_with_custom_url(self, mock_gateway_u assert servers[0].mcp_server_name == "ProdServer" assert servers[0].mcp_server_unique_name == "prod_server" assert servers[0].url == "https://prod.custom.url/mcp" + + +class TestPrepareGatewayHeaders: + """Tests for _prepare_gateway_headers and _resolve_agent_id_for_header.""" + + @pytest.fixture + def service(self): + """Create a service instance for testing.""" + return McpToolServerConfigurationService() + + @pytest.fixture + def create_test_jwt(self): + """Fixture to create test JWT tokens.""" + import jwt + + def _create(claims: dict) -> str: + return jwt.encode(claims, key="", algorithm="none") + + return _create + + @pytest.fixture + def default_options(self): + """Default ToolOptions for tests.""" + from microsoft_agents_a365.tooling.models import ToolOptions + + return ToolOptions(orchestrator_name="TestOrchestrator") + + def test_includes_authorization_header(self, service, default_options): + """Test that Authorization header is always included.""" + headers = service._prepare_gateway_headers("test-token", default_options) + assert headers["Authorization"] == "Bearer test-token" + + def test_includes_user_agent_header(self, service, default_options): + """Test that User-Agent header is always included.""" + headers = service._prepare_gateway_headers("test-token", default_options) + assert "User-Agent" in headers + assert "Agent365SDK" in headers["User-Agent"] + assert "TestOrchestrator" in headers["User-Agent"] + + def test_includes_x_ms_agentid_from_token_claims( + self, service, create_test_jwt, default_options + ): + """Test x-ms-agentid header is populated from token claims.""" + token = create_test_jwt({"appid": "token-app-id-123"}) + headers = service._prepare_gateway_headers(token, default_options) + assert headers.get("x-ms-agentid") == "token-app-id-123" + + def test_includes_x_ms_agentid_from_xms_par_app_azp( + self, service, create_test_jwt, default_options + ): + """Test x-ms-agentid prefers xms_par_app_azp over appid.""" + token = create_test_jwt({ + "xms_par_app_azp": "blueprint-id-from-token", + "appid": "app-id-456", + }) + headers = service._prepare_gateway_headers(token, default_options) + assert headers.get("x-ms-agentid") == "blueprint-id-from-token" + + def test_includes_x_ms_agentid_from_turn_context_blueprint_id( + self, service, create_test_jwt, default_options + ): + """Test x-ms-agentid prefers TurnContext agenticAppBlueprintId over token.""" + token = create_test_jwt({"appid": "token-app-id"}) + + # Create mock TurnContext with agenticAppBlueprintId + mock_from = MagicMock() + mock_from.agentic_app_blueprint_id = "context-blueprint-id" + mock_activity = MagicMock() + mock_activity.from_ = mock_from + mock_context = MagicMock() + mock_context.activity = mock_activity + + headers = service._prepare_gateway_headers(token, default_options, mock_context) + assert headers.get("x-ms-agentid") == "context-blueprint-id" + + def test_falls_back_to_application_name(self, service, create_test_jwt, default_options): + """Test x-ms-agentid falls back to application name when no token claims.""" + # Token with no relevant claims + token = create_test_jwt({"sub": "some-subject"}) + + with patch( + "microsoft_agents_a365.runtime.utility.Utility.get_application_name", + return_value="my-application", + ): + headers = service._prepare_gateway_headers(token, default_options) + assert headers.get("x-ms-agentid") == "my-application" + + def test_omits_x_ms_agentid_when_nothing_available( + self, service, create_test_jwt, default_options + ): + """Test x-ms-agentid header is omitted when no identifier is available.""" + # Token with no relevant claims + token = create_test_jwt({"sub": "some-subject"}) + + with patch( + "microsoft_agents_a365.runtime.utility.Utility.get_application_name", + return_value=None, + ): + headers = service._prepare_gateway_headers(token, default_options) + assert "x-ms-agentid" not in headers + + +class TestResolveAgentIdForHeader: + """Tests for _resolve_agent_id_for_header method.""" + + @pytest.fixture + def service(self): + """Create a service instance for testing.""" + return McpToolServerConfigurationService() + + @pytest.fixture + def create_test_jwt(self): + """Fixture to create test JWT tokens.""" + import jwt as pyjwt + + def _create(claims: dict) -> str: + return pyjwt.encode(claims, key="", algorithm="none") + + return _create + + def test_priority_1_turn_context_blueprint_id(self, service, create_test_jwt): + """Test TurnContext agenticAppBlueprintId has highest priority.""" + token = create_test_jwt({ + "xms_par_app_azp": "token-blueprint", + "appid": "token-appid", + }) + + mock_from = MagicMock() + mock_from.agentic_app_blueprint_id = "context-blueprint-id" + mock_activity = MagicMock() + mock_activity.from_ = mock_from + mock_context = MagicMock() + mock_context.activity = mock_activity + + result = service._resolve_agent_id_for_header(token, mock_context) + assert result == "context-blueprint-id" + + def test_priority_2_token_xms_par_app_azp(self, service, create_test_jwt): + """Test token xms_par_app_azp claim is second priority.""" + token = create_test_jwt({ + "xms_par_app_azp": "token-blueprint", + "appid": "token-appid", + }) + + result = service._resolve_agent_id_for_header(token, None) + assert result == "token-blueprint" + + def test_priority_3_token_appid(self, service, create_test_jwt): + """Test token appid claim is third priority.""" + token = create_test_jwt({"appid": "token-appid"}) + + result = service._resolve_agent_id_for_header(token, None) + assert result == "token-appid" + + def test_priority_4_application_name(self, service, create_test_jwt): + """Test application name is lowest priority.""" + token = create_test_jwt({"sub": "no-relevant-claims"}) + + with patch( + "microsoft_agents_a365.runtime.utility.Utility.get_application_name", + return_value="fallback-app-name", + ): + result = service._resolve_agent_id_for_header(token, None) + assert result == "fallback-app-name" + + def test_returns_none_when_nothing_available(self, service, create_test_jwt): + """Test returns None when no identifier is available.""" + token = create_test_jwt({"sub": "no-relevant-claims"}) + + with patch( + "microsoft_agents_a365.runtime.utility.Utility.get_application_name", + return_value=None, + ): + result = service._resolve_agent_id_for_header(token, None) + assert result is None + + def test_handles_turn_context_without_activity(self, service, create_test_jwt): + """Test handles TurnContext with None activity gracefully.""" + token = create_test_jwt({"appid": "token-appid"}) + + mock_context = MagicMock() + mock_context.activity = None + + result = service._resolve_agent_id_for_header(token, mock_context) + assert result == "token-appid" + + def test_handles_turn_context_without_from(self, service, create_test_jwt): + """Test handles TurnContext activity with None from_ gracefully.""" + token = create_test_jwt({"appid": "token-appid"}) + + mock_activity = MagicMock() + mock_activity.from_ = None + mock_context = MagicMock() + mock_context.activity = mock_activity + + result = service._resolve_agent_id_for_header(token, mock_context) + assert result == "token-appid" + + def test_handles_turn_context_without_blueprint_id_attribute(self, service, create_test_jwt): + """Test handles from_ object without agentic_app_blueprint_id attribute.""" + token = create_test_jwt({"appid": "token-appid"}) + + # Mock from_ that doesn't have agentic_app_blueprint_id + mock_from = MagicMock(spec=[]) # Empty spec means no attributes + mock_activity = MagicMock() + mock_activity.from_ = mock_from + mock_context = MagicMock() + mock_context.activity = mock_activity + + result = service._resolve_agent_id_for_header(token, mock_context) + assert result == "token-appid" + + def test_skips_empty_blueprint_id(self, service, create_test_jwt): + """Test skips empty string blueprint ID from TurnContext.""" + token = create_test_jwt({"appid": "token-appid"}) + + mock_from = MagicMock() + mock_from.agentic_app_blueprint_id = "" # Empty string + mock_activity = MagicMock() + mock_activity.from_ = mock_from + mock_context = MagicMock() + mock_context.activity = mock_activity + + result = service._resolve_agent_id_for_header(token, mock_context) + assert result == "token-appid"