Distribute error resources across packages with error codes and help URLs

libraries/microsoft-agents-activity/microsoft_agents/activity/_channel_id_field_mixin.py

@@ -42,9 +42,9 @@ def channel_id(self, value: Any):
         elif isinstance(value, str):
             self._channel_id = ChannelId(value)
         else:
+            from microsoft_agents.hosting.core import error_resources
             raise ValueError(
-                f"Invalid type for channel_id: {type(value)}. "
-                "Expected ChannelId or str."
+                error_resources.InvalidChannelIdType.format(type(value))
             )
 
     def _set_validated_channel_id(self, data: Any) -> None:

libraries/microsoft-agents-activity/microsoft_agents/activity/activity.py

@@ -218,8 +218,9 @@ def _validate_channel_id(
                     activity.channel_id.sub_channel
                     and activity.channel_id.sub_channel != product_info.id
                 ):
+                    from microsoft_agents.hosting.core import error_resources
                     raise Exception(
-                        "Conflict between channel_id.sub_channel and productInfo entity"
+                        str(error_resources.ChannelIdProductInfoConflict)
                     )
                 activity.channel_id = ChannelId(
                     channel=activity.channel_id.channel,
@@ -256,8 +257,9 @@ def _serialize_sub_channel_data(
         # self.channel_id is the source of truth for serialization
         if self.channel_id and self.channel_id.sub_channel:
             if product_info and product_info.get("id") != self.channel_id.sub_channel:
+                from microsoft_agents.hosting.core import error_resources
                 raise Exception(
-                    "Conflict between channel_id.sub_channel and productInfo entity"
+                    str(error_resources.ChannelIdProductInfoConflict)
                 )
             elif not product_info:
                 if not serialized.get("entities"):

libraries/microsoft-agents-activity/microsoft_agents/activity/channel_id.py

@@ -52,14 +52,16 @@ def __new__(
         """
         if isinstance(value, str):
             if channel or sub_channel:
+                from microsoft_agents.hosting.core import error_resources
                 raise ValueError(
-                    "If value is provided, channel and sub_channel must be None"
+                    str(error_resources.ChannelIdValueConflict)
                 )
 
             value = value.strip()
             if value:
                 return str.__new__(cls, value)
-            raise TypeError("value must be a non empty string if provided")
+            from microsoft_agents.hosting.core import error_resources
+            raise TypeError(str(error_resources.ChannelIdValueMustBeNonEmpty))
         else:
             if (
                 not isinstance(channel, str)

libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_auth.py

@@ -25,6 +25,7 @@
     AuthTypes,
     AccessTokenProviderBase,
     AgentAuthConfiguration,
+    error_resources,
 )
 
 logger = logging.getLogger(__name__)
@@ -65,7 +66,7 @@ async def get_access_token(
         )
         valid_uri, instance_uri = self._uri_validator(resource_url)
         if not valid_uri:
-            raise ValueError("Invalid instance URL")
+            raise ValueError(str(error_resources.InvalidInstanceUrl))
 
         local_scopes = self._resolve_scopes_list(instance_uri, scopes)
         self._create_client_application()
@@ -86,7 +87,7 @@ async def get_access_token(
         res = auth_result_payload.get("access_token") if auth_result_payload else None
         if not res:
             logger.error("Failed to acquire token for resource %s", auth_result_payload)
-            raise ValueError(f"Failed to acquire token. {str(auth_result_payload)}")
+            raise ValueError(error_resources.FailedToAcquireToken.format(str(auth_result_payload)))
 
         return res
 
@@ -106,7 +107,7 @@ async def acquire_token_on_behalf_of(
                 "Attempted on-behalf-of flow with Managed Identity authentication."
             )
             raise NotImplementedError(
-                "On-behalf-of flow is not supported with Managed Identity authentication."
+                str(error_resources.OnBehalfOfFlowNotSupportedManagedIdentity)
             )
         elif isinstance(self._msal_auth_client, ConfidentialClientApplication):
             # TODO: Handling token error / acquisition failed
@@ -123,15 +124,17 @@ async def acquire_token_on_behalf_of(
                 logger.error(
                     f"Failed to acquire token on behalf of user: {user_assertion}"
                 )
-                raise ValueError(f"Failed to acquire token. {str(token)}")
+                raise ValueError(error_resources.FailedToAcquireToken.format(str(token)))
 
             return token["access_token"]
 
         logger.error(
             f"On-behalf-of flow is not supported with the current authentication type: {self._msal_auth_client.__class__.__name__}"
         )
         raise NotImplementedError(
-            f"On-behalf-of flow is not supported with the current authentication type: {self._msal_auth_client.__class__.__name__}"
+            error_resources.OnBehalfOfFlowNotSupportedAuthType.format(
+                self._msal_auth_client.__class__.__name__
+            )
         )
 
     def _create_client_application(self) -> None:
@@ -187,7 +190,7 @@ def _create_client_application(self) -> None:
                 logger.error(
                     f"Unsupported authentication type: {self._msal_configuration.AUTH_TYPE}"
                 )
-                raise NotImplementedError("Authentication type not supported")
+                raise NotImplementedError(str(error_resources.AuthenticationTypeNotSupported))
 
             self._msal_auth_client = ConfidentialClientApplication(
                 client_id=self._msal_configuration.CLIENT_ID,
@@ -233,7 +236,7 @@ async def get_agentic_application_token(
         """
 
         if not agent_app_instance_id:
-            raise ValueError("Agent application instance Id must be provided.")
+            raise ValueError(str(error_resources.AgentApplicationInstanceIdRequired))
 
         logger.info(
             "Attempting to get agentic application token from agent_app_instance_id %s",
@@ -267,7 +270,7 @@ async def get_agentic_instance_token(
         """
 
         if not agent_app_instance_id:
-            raise ValueError("Agent application instance Id must be provided.")
+            raise ValueError(str(error_resources.AgentApplicationInstanceIdRequired))
 
         logger.info(
             "Attempting to get agentic instance token from agent_app_instance_id %s",
@@ -283,7 +286,7 @@ async def get_agentic_instance_token(
                 agent_app_instance_id,
             )
             raise Exception(
-                f"Failed to acquire agentic instance token or agent token for agent_app_instance_id {agent_app_instance_id}"
+                error_resources.FailedToAcquireAgenticInstanceToken.format(agent_app_instance_id)
             )
 
         authority = (
@@ -306,7 +309,7 @@ async def get_agentic_instance_token(
                 agent_app_instance_id,
             )
             raise Exception(
-                f"Failed to acquire agentic instance token or agent token for agent_app_instance_id {agent_app_instance_id}"
+                error_resources.FailedToAcquireAgenticInstanceToken.format(agent_app_instance_id)
             )
 
         # future scenario where we don't know the blueprint id upfront
@@ -316,7 +319,7 @@ async def get_agentic_instance_token(
             logger.error(
                 "Failed to acquire agentic instance token, %s", agentic_instance_token
             )
-            raise ValueError(f"Failed to acquire token. {str(agentic_instance_token)}")
+            raise ValueError(error_resources.FailedToAcquireToken.format(str(agentic_instance_token)))
 
         logger.debug(
             "Agentic blueprint id: %s",
@@ -345,7 +348,7 @@ async def get_agentic_user_token(
         """
         if not agent_app_instance_id or not agentic_user_id:
             raise ValueError(
-                "Agent application instance Id and agentic user Id must be provided."
+                str(error_resources.AgentApplicationInstanceIdAndUserIdRequired)
             )
 
         logger.info(
@@ -364,7 +367,9 @@ async def get_agentic_user_token(
                 agentic_user_id,
             )
             raise Exception(
-                f"Failed to acquire instance token or agent token for agent_app_instance_id {agent_app_instance_id} and agentic_user_id {agentic_user_id}"
+                error_resources.FailedToAcquireInstanceOrAgentToken.format(
+                    agent_app_instance_id, agentic_user_id
+                )
             )
 
         authority = (

libraries/microsoft-agents-copilotstudio-client/microsoft_agents/copilotstudio/client/power_platform_environment.py

@@ -1,3 +1,4 @@
+from microsoft_agents.hosting.core import error_resources
 from urllib.parse import urlparse, urlunparse
 from typing import Optional
 from .connection_settings import ConnectionSettings
@@ -23,12 +24,12 @@ def get_copilot_studio_connection_url(
     ) -> str:
         if cloud == PowerPlatformCloud.OTHER and not cloud_base_address:
             raise ValueError(
-                "cloud_base_address must be provided when PowerPlatformCloud is Other"
+                str(error_resources.CloudBaseAddressRequired)
             )
         if not settings.environment_id:
-            raise ValueError("EnvironmentId must be provided")
+            raise ValueError(str(error_resources.EnvironmentIdRequired))
         if not settings.agent_identifier:
-            raise ValueError("AgentIdentifier must be provided")
+            raise ValueError(str(error_resources.AgentIdentifierRequired))
         if settings.cloud and settings.cloud != PowerPlatformCloud.UNKNOWN:
             cloud = settings.cloud
         if cloud == PowerPlatformCloud.OTHER:
@@ -40,7 +41,7 @@ def get_copilot_studio_connection_url(
                 cloud_base_address = settings.custom_power_platform_cloud
             else:
                 raise ValueError(
-                    "Either CustomPowerPlatformCloud or cloud_base_address must be provided when PowerPlatformCloud is Other"
+                    str(error_resources.CustomCloudOrBaseAddressRequired)
                 )
         if settings.copilot_agent_type:
             agent_type = settings.copilot_agent_type
@@ -61,7 +62,7 @@ def get_token_audience(
     ) -> str:
         if cloud == PowerPlatformCloud.OTHER and not cloud_base_address:
             raise ValueError(
-                "cloud_base_address must be provided when PowerPlatformCloud is Other"
+                str(error_resources.CloudBaseAddressRequired)
             )
         if not settings and cloud == PowerPlatformCloud.UNKNOWN:
             raise ValueError("Either settings or cloud must be provided")
@@ -79,7 +80,7 @@ def get_token_audience(
                 cloud_base_address = settings.custom_power_platform_cloud
             else:
                 raise ValueError(
-                    "Either CustomPowerPlatformCloud or cloud_base_address must be provided when PowerPlatformCloud is Other"
+                    str(error_resources.CustomCloudOrBaseAddressRequired)
                 )
 
         cloud_base_address = cloud_base_address or "api.unknown.powerplatform.com"
@@ -117,7 +118,7 @@ def get_environment_endpoint(
     ) -> str:
         if cloud == PowerPlatformCloud.OTHER and not cloud_base_address:
             raise ValueError(
-                "cloud_base_address must be provided when PowerPlatformCloud is Other"
+                str(error_resources.CloudBaseAddressRequired)
             )
         cloud_base_address = cloud_base_address or "api.unknown.powerplatform.com"
         normalized_resource_id = environment_id.lower().replace("-", "")

libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/_start_agent_process.py

@@ -1,5 +1,6 @@
 from typing import Optional
 from aiohttp.web import Request, Response
+from microsoft_agents.hosting.core import error_resources
 from microsoft_agents.hosting.core.app import AgentApplication
 from .cloud_adapter import CloudAdapter
 
@@ -15,9 +16,9 @@ async def start_agent_process(
         agent_application (AgentApplication): The agent application to run.
     """
     if not adapter:
-        raise TypeError("start_agent_process: adapter can't be None")
+        raise TypeError(str(error_resources.AdapterRequired))
     if not agent_application:
-        raise TypeError("start_agent_process: agent_application can't be None")
+        raise TypeError(str(error_resources.AgentApplicationRequired))
 
     # Start the agent application with the provided adapter
     return await adapter.process(

libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/app/streaming/streaming_response.py

@@ -16,8 +16,8 @@
     SensitivityUsageInfo,
 )
 
-if TYPE_CHECKING:
-    from microsoft_agents.hosting.core.turn_context import TurnContext
+from microsoft_agents.hosting.core import error_resources
+from microsoft_agents.hosting.core.turn_context import TurnContext
 
 from .citation import Citation
 from .citation_util import CitationUtil
@@ -99,7 +99,7 @@ def queue_informative_update(self, text: str) -> None:
             return
 
         if self._ended:
-            raise RuntimeError("The stream has already ended.")
+            raise RuntimeError(str(error_resources.StreamAlreadyEnded))
 
         # Queue a typing activity
         def create_activity():
@@ -135,7 +135,7 @@ def queue_text_chunk(
         if self._cancelled:
             return
         if self._ended:
-            raise RuntimeError("The stream has already ended.")
+            raise RuntimeError(str(error_resources.StreamAlreadyEnded))
 
         # Update full message text
         self._message += text
@@ -151,7 +151,7 @@ async def end_stream(self) -> None:
         Ends the stream by sending the final message to the client.
         """
         if self._ended:
-            raise RuntimeError("The stream has already ended.")
+            raise RuntimeError(str(error_resources.StreamAlreadyEnded))
 
         # Queue final message
         self._ended = True

libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/cloud_adapter.py

@@ -12,6 +12,7 @@
     HTTPUnauthorized,
     HTTPUnsupportedMediaType,
 )
+from microsoft_agents.hosting.core import error_resources
 from microsoft_agents.hosting.core.authorization import (
     ClaimsIdentity,
     Connections,
@@ -70,9 +71,9 @@ async def on_turn_error(context: TurnContext, error: Exception):
 
     async def process(self, request: Request, agent: Agent) -> Optional[Response]:
         if not request:
-            raise TypeError("CloudAdapter.process: request can't be None")
+            raise TypeError(str(error_resources.RequestRequired))
         if not agent:
-            raise TypeError("CloudAdapter.process: agent can't be None")
+            raise TypeError(str(error_resources.AgentRequired))
 
         if request.method == "POST":
             # Deserialize the incoming Activity

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/__init__.py

@@ -82,6 +82,9 @@
 from .storage import Storage
 from .storage.memory_storage import MemoryStorage
 
+# Error Resources
+from .errors import error_resources, ErrorMessage, ErrorResources
+
 
 # Define the package's public interface
 __all__ = [
@@ -148,4 +151,7 @@
     "MemoryStorage",
     "AgenticUserAuthorization",
     "Authorization",
+    "error_resources",
+    "ErrorMessage",
+    "ErrorResources",
 ]