microsoft · Rodrigo Brandão (rodrigobr-msft) · Aug 12, 2026 · Aug 13, 2026 · Aug 13, 2026 · Aug 14, 2026
@@ -30,7 +30,7 @@ async def test_cloud_adapter_configures_user_token_client_endpoint(
             "token_service_endpoint": token_service_endpoint
         },
     )
-    identity = ClaimsIdentity({"aud": "test-app-id"}, True)
+    identity = ClaimsIdentity({"aud": "test-app-id"})
     context = TurnContext(
         adapter,
         Activity(

@@ -30,7 +30,7 @@ async def test_cloud_adapter_configures_user_token_client_endpoint(
             "token_service_endpoint": token_service_endpoint
         },
     )
-    identity = ClaimsIdentity({"aud": "test-app-id"}, True)
+    identity = ClaimsIdentity({"aud": "test-app-id"})
     context = TurnContext(
         adapter,
         Activity(

@@ -41,7 +41,7 @@ async def _claims_handler(request):
     identity = request["claims_identity"]
     return web.json_response(
         {
-            "authenticated": identity.is_authenticated,
+            "allow_anonymous": identity.allow_anonymous,
             "authentication_type": identity.authentication_type,
         }
     )
@@ -73,8 +73,8 @@ async def test_aiohttp_global_middleware_allows_anonymous_request_from_env_confi
 
     assert response.status == 200
     assert await response.json() == {
-        "authenticated": False,
-        "authentication_type": "Anonymous",
+        "allow_anonymous": True,
+        "authentication_type": None,
     }
 
 
@@ -105,7 +105,7 @@ async def test_aiohttp_global_middleware_accepts_real_service_connection_token(
     response = await client.get("/", headers={"Authorization": f"Bearer {token}"})
 
     assert response.status == 200
-    assert (await response.json())["authenticated"] is True
+    assert (await response.json())["allow_anonymous"] is False
 
 
 @_requires_real_service_connection
@@ -138,8 +138,8 @@ async def test_aiohttp_decorator_allows_anonymous_request_from_env_config(
 
     assert response.status == 200
     assert await response.json() == {
-        "authenticated": False,
-        "authentication_type": "Anonymous",
+        "allow_anonymous": True,
+        "authentication_type": None,
     }
 
 

@@ -41,7 +41,7 @@
 def _claims_payload(request: Request):
     identity = request.state.claims_identity
     return {
-        "authenticated": identity.is_authenticated,
+        "allow_anonymous": identity.allow_anonymous,
         "authentication_type": identity.authentication_type,
     }
 
@@ -78,8 +78,8 @@ def test_fastapi_global_middleware_allows_anonymous_request_from_env_config():
 
     assert response.status_code == 200
     assert response.json() == {
-        "authenticated": False,
-        "authentication_type": "Anonymous",
+        "allow_anonymous": True,
+        "authentication_type": None,
     }
 
 
@@ -105,7 +105,7 @@ async def test_fastapi_global_middleware_accepts_real_service_connection_token()
     response = client.get("/", headers={"Authorization": f"Bearer {token}"})
 
     assert response.status_code == 200
-    assert response.json()["authenticated"] is True
+    assert response.json()["allow_anonymous"] is False
 
 
 @_requires_real_service_connection
@@ -134,8 +134,8 @@ def test_fastapi_decorator_allows_anonymous_request_from_env_config():
 
     assert response.status_code == 200
     assert response.json() == {
-        "authenticated": False,
-        "authentication_type": "Anonymous",
+        "allow_anonymous": True,
+        "authentication_type": None,
     }
 
 

@@ -112,10 +112,8 @@ async def process_request(
             span.share(activity=activity)
 
             # Get claims identity (default to anonymous if not set by middleware)
-            claims_identity: (
-                ClaimsIdentity
-            ) = request.get_claims_identity() or ClaimsIdentity(
-                {}, False, authentication_type="Anonymous"
+            claims_identity: ClaimsIdentity = (
+                request.get_claims_identity() or ClaimsIdentity()
             )
 
             # Validate required activity fields

@@ -97,7 +97,9 @@ def identity_from_claims(claims: dict[str, str]) -> ClaimsIdentity:
         :return: Reconstituted claims identity.
         :rtype: :class:`~microsoft_agents.hosting.core.authorization.ClaimsIdentity`
         """
-        return ClaimsIdentity(claims=dict(claims), is_authenticated=True)
+        if not claims:
+            return ClaimsIdentity()
+        return ClaimsIdentity(claims=dict(claims))
 
     # ------------------------------------------------------------------
     # Validation

@@ -1,27 +1,86 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
-from typing import Optional
+
+import warnings
 
 from .authentication_constants import AuthenticationConstants
 
 
 class ClaimsIdentity:
+    """Represents an identity with associated claims and authentication information.
+
+    For context, this class merges the functionality of ClaimsIdentity and AgentClaims from .NET
+    """
+
+    claims: dict[str, str]
+    authentication_type: str | None
+    security_token: str | None  # deprecated, will be removed in future versions
+
     def __init__(
         self,
-        claims: dict[str, str],
-        is_authenticated: bool,
-        authentication_type: Optional[str] = None,
-        security_token: Optional[str] = None,
+        claims: dict[str, str] | None = None,
+        is_authenticated: bool | None = None,
+        authentication_type: str | None = None,
+        security_token: str | None = None,
     ):
+        """Creates a new instance of the ClaimsIdentity class.
+
+        :param claims: A dictionary of claims associated with the identity.
+        :param is_authenticated: A boolean indicating whether the identity is authenticated. (Deprecated)
+        :param authentication_type: A string representing the type of authentication used.
+            None values indicate that the identity is not authenticated.
+        :param security_token: The security token associated with the identity.
+        """
+        if claims is None:
+            claims = {}
         self.claims = claims
-        self.is_authenticated = is_authenticated
+        if is_authenticated is not None:
+            warnings.warn(
+                "The 'is_authenticated' parameter is deprecated and will be removed in future versions.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
         self.authentication_type = authentication_type
         self.security_token = security_token
+        self._is_authenticated = is_authenticated
+
+    def get_claim_value(self, claim_type: str) -> str | None:
+        """Gets the value of a specific claim type from the claims dictionary.
 
-    def get_claim_value(self, claim_type: str) -> Optional[str]:
+        :param claim_type: The type of claim to retrieve.
+        :return: The value of the claim if found, otherwise None.
+        """
         return self.claims.get(claim_type)
 
-    def get_app_id(self) -> Optional[str]:
+    @property
+    def allow_anonymous(self) -> bool:
+        """Returns True if the identity allows anonymous access, otherwise False."""
+        return (
+            not self.authentication_type
+            or self.authentication_type.lower() == "anonymous"
+        ) and not self.claims
+
+    @property
+    def is_authenticated(self) -> bool:
+        """Returns True if the identity is authenticated, otherwise False."""
+        warnings.warn(
+            "The 'is_authenticated' property is deprecated and will be removed in future versions.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return bool(self.claims)
+
+    @is_authenticated.setter
+    def is_authenticated(self, value: bool) -> None:
+        """(Deprecated). This is now a no-op."""
+        warnings.warn(
+            "The 'is_authenticated' property is deprecated and will be removed in future versions.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
+    def get_app_id(self) -> str | None:
         """
         Gets the AppId from the current ClaimsIdentity.
 
@@ -32,7 +91,7 @@ def get_app_id(self) -> Optional[str]:
             AuthenticationConstants.AUDIENCE_CLAIM, None
         ) or self.claims.get(AuthenticationConstants.APP_ID_CLAIM, None)
 
-    def get_outgoing_app_id(self) -> Optional[str]:
+    def get_outgoing_app_id(self) -> str | None:
         """
         Gets the outgoing AppId from current claims.
 

@@ -154,12 +154,12 @@ async def validate_token(self, token: str) -> ClaimsIdentity:
         )
 
         logger.debug("JWT token validated successfully.")
-        return ClaimsIdentity(decoded_token, True, security_token=token)
+        return ClaimsIdentity(decoded_token, security_token=token)
 
     def get_anonymous_claims(self) -> ClaimsIdentity:
         """Returns a ClaimsIdentity for an anonymous user."""
         logger.debug("Returning anonymous claims identity.")
-        return ClaimsIdentity({}, False, authentication_type="Anonymous")
+        return ClaimsIdentity()
 
 
 def _build_jwks_uri(

@@ -252,10 +252,16 @@ async def create_conversation(  # pylint: disable=arguments-differ
         claims_identity = self.create_claims_identity(agent_app_id)
         claims_identity.claims[AuthenticationConstants.SERVICE_URL_CLAIM] = service_url
 
+        use_anonymous_auth_callback = claims_identity.allow_anonymous
+
         # Create the connector client to use for outbound requests.
         connector_client = (
             await self._channel_service_client_factory.create_connector_client(
-                None, claims_identity, service_url, audience
+                None,
+                claims_identity,
+                service_url,
+                audience,
+                use_anonymous=use_anonymous_auth_callback,
             )
         )
 
@@ -285,7 +291,7 @@ async def create_conversation(  # pylint: disable=arguments-differ
         # Create a UserTokenClient instance for the application to use. (For example, in the OAuthPrompt.)
         user_token_client = (
             await self._channel_service_client_factory.create_user_token_client(
-                context, claims_identity
+                context, claims_identity, use_anonymous_auth_callback
             )
         )
         context.services.set(UserTokenClientBase, user_token_client)
@@ -307,6 +313,8 @@ async def process_proactive(
         callback: Callable[[TurnContext], Awaitable],
     ):
 
+        use_anonymous_auth_callback = claims_identity.allow_anonymous
+
         # Create a turn context and run the pipeline.
         context = self._create_turn_context(
             claims_identity,
@@ -316,7 +324,7 @@ async def process_proactive(
 
         user_token_client = (
             await self._channel_service_client_factory.create_user_token_client(
-                context, claims_identity
+                context, claims_identity, use_anonymous_auth_callback
             )
         )
         context.services.set(UserTokenClientBase, user_token_client)
@@ -327,7 +335,11 @@ async def process_proactive(
         # Create the connector client to use for outbound requests.
         connector_client = (
             await self._channel_service_client_factory.create_connector_client(
-                context, claims_identity, continuation_activity.service_url, audience
+                context,
+                claims_identity,
+                continuation_activity.service_url,
+                audience,
+                use_anonymous=use_anonymous_auth_callback,
             )
         )
         context.services.set(ConnectorClientBase, connector_client)
@@ -390,12 +402,7 @@ async def process_activity(
         else:
             outgoing_audience = AuthenticationConstants.AGENTS_SDK_SCOPE
 
-        use_anonymous_auth_callback = False
-        if (
-            not claims_identity.is_authenticated
-            and claims_identity.authentication_type == "Anonymous"
-        ):
-            use_anonymous_auth_callback = True
+        use_anonymous_auth_callback = claims_identity.allow_anonymous
 
         # Create a turn context and run the pipeline.
         context = self._create_turn_context(
@@ -456,7 +463,6 @@ def create_claims_identity(self, agent_app_id: str = "") -> ClaimsIdentity:
                 AuthenticationConstants.AUDIENCE_CLAIM: agent_app_id,
                 AuthenticationConstants.APP_ID_CLAIM: agent_app_id,
             },
-            False,
         )
 
     @staticmethod

@@ -11,6 +11,8 @@ filterwarnings =
     ignore::DeprecationWarning:setuptools.*
     ignore::PendingDeprecationWarning
     ignore:The bot client is deprecated and will be removed in a future release\.:DeprecationWarning
+    ignore:The 'is_authenticated' parameter is deprecated and will be removed in future versions\.:DeprecationWarning
+    ignore:The 'is_authenticated' property is deprecated and will be removed in future versions\.:DeprecationWarning
     # pytest-asyncio warnings that are safe to ignore
     ignore:.*deprecated.*asyncio.*:DeprecationWarning:pytest_asyncio.*
 

@@ -113,6 +113,12 @@ def test_identity_from_claims_is_authenticated(self):
         identity = Conversation.identity_from_claims(claims)
         assert identity.is_authenticated is True
 
+    def test_identity_from_empty_claims_allows_anonymous(self):
+        identity = Conversation.identity_from_claims({})
+
+        assert identity.claims == {}
+        assert identity.allow_anonymous is True
+
     def test_identity_from_claims_preserves_values(self):
         claims = {"aud": "app-id", "tid": "tenant", "ver": "2.0"}
         identity = Conversation.identity_from_claims(claims)