|
| 1 | +# Copyright (c) Microsoft. All rights reserved. |
| 2 | + |
| 3 | +""" |
| 4 | +Utility functions for Microsoft Agent 365 runtime operations. |
| 5 | +
|
| 6 | +This module provides utility functions for token handling, agent identity resolution, |
| 7 | +and other common runtime operations. |
| 8 | +""" |
| 9 | + |
| 10 | +from __future__ import annotations |
| 11 | + |
| 12 | +import uuid |
| 13 | +from typing import Any, Optional |
| 14 | + |
| 15 | +import jwt |
| 16 | + |
| 17 | + |
| 18 | +class Utility: |
| 19 | + """ |
| 20 | + Utility class providing common runtime operations for Agent 365. |
| 21 | +
|
| 22 | + This class contains static methods for token processing, agent identity resolution, |
| 23 | + and other utility functions used across the Agent 365 runtime. |
| 24 | + """ |
| 25 | + |
| 26 | + @staticmethod |
| 27 | + def get_app_id_from_token(token: Optional[str]) -> str: |
| 28 | + """ |
| 29 | + Decodes the current token and retrieves the App ID (appid or azp claim). |
| 30 | +
|
| 31 | + Args: |
| 32 | + token: JWT token to decode. Can be None or empty. |
| 33 | +
|
| 34 | + Returns: |
| 35 | + str: The App ID from the token's claims, or empty GUID if token is invalid. |
| 36 | + Returns "00000000-0000-0000-0000-000000000000" if no valid App ID is found. |
| 37 | + """ |
| 38 | + if not token or not token.strip(): |
| 39 | + return str(uuid.UUID(int=0)) |
| 40 | + |
| 41 | + try: |
| 42 | + # Decode the JWT token without verification (we only need the claims) |
| 43 | + # Note: verify=False is used because we only need to extract claims, |
| 44 | + # not verify the token's authenticity |
| 45 | + decoded_payload = jwt.decode(token, options={"verify_signature": False}) |
| 46 | + |
| 47 | + # Look for appid or azp claims (appid takes precedence) |
| 48 | + app_id = decoded_payload.get("appid") or decoded_payload.get("azp") |
| 49 | + return app_id if app_id else "" |
| 50 | + |
| 51 | + except (jwt.DecodeError, jwt.InvalidTokenError): |
| 52 | + # Token is malformed or invalid |
| 53 | + return "" |
| 54 | + |
| 55 | + @staticmethod |
| 56 | + def resolve_agent_identity(context: Any, auth_token: Optional[str]) -> str: |
| 57 | + """ |
| 58 | + Resolves the agent identity from the turn context or auth token. |
| 59 | +
|
| 60 | + Args: |
| 61 | + context: Turn context of the conversation turn. Expected to have an Activity |
| 62 | + with methods like is_agentic_request() and get_agentic_instance_id(). |
| 63 | + auth_token: Authentication token if available. |
| 64 | +
|
| 65 | + Returns: |
| 66 | + str: The agent identity (App ID). Returns the agentic instance ID if the |
| 67 | + request is agentic, otherwise returns the App ID from the auth token. |
| 68 | + """ |
| 69 | + try: |
| 70 | + # App ID is required to pass to MCP server URL |
| 71 | + # Try to get agentic instance ID if this is an agentic request |
| 72 | + if context and context.activity and context.activity.is_agentic_request(): |
| 73 | + agentic_id = context.activity.get_agentic_instance_id() |
| 74 | + return agentic_id if agentic_id else "" |
| 75 | + |
| 76 | + except (AttributeError, TypeError, Exception): |
| 77 | + # Context/activity doesn't have the expected methods or properties |
| 78 | + # or any other error occurred while accessing context/activity |
| 79 | + pass |
| 80 | + |
| 81 | + # Fallback to extracting App ID from the auth token |
| 82 | + return Utility.get_app_id_from_token(auth_token) |
0 commit comments