diff --git a/dev/integration/tests/adapter/test_aiohttp_cloud_adapter.py b/dev/integration/tests/adapter/test_aiohttp_cloud_adapter.py
index 154b6529..d1a21492 100644
--- a/dev/integration/tests/adapter/test_aiohttp_cloud_adapter.py
+++ b/dev/integration/tests/adapter/test_aiohttp_cloud_adapter.py
@@ -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(
diff --git a/dev/integration/tests/adapter/test_fastapi_cloud_adapter.py b/dev/integration/tests/adapter/test_fastapi_cloud_adapter.py
index 4c79c1a7..54bf339f 100644
--- a/dev/integration/tests/adapter/test_fastapi_cloud_adapter.py
+++ b/dev/integration/tests/adapter/test_fastapi_cloud_adapter.py
@@ -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(
diff --git a/dev/integration/tests/jwt_validation/test_aiohttp_jwt_validation.py b/dev/integration/tests/jwt_validation/test_aiohttp_jwt_validation.py
index a3a4c73e..842206ff 100644
--- a/dev/integration/tests/jwt_validation/test_aiohttp_jwt_validation.py
+++ b/dev/integration/tests/jwt_validation/test_aiohttp_jwt_validation.py
@@ -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,
     }
 
 
diff --git a/dev/integration/tests/jwt_validation/test_fastapi_jwt_validation.py b/dev/integration/tests/jwt_validation/test_fastapi_jwt_validation.py
index c1ce1133..b5f4c8f4 100644
--- a/dev/integration/tests/jwt_validation/test_fastapi_jwt_validation.py
+++ b/dev/integration/tests/jwt_validation/test_fastapi_jwt_validation.py
@@ -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,
     }
 
 
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/_http_adapter_base.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/_http_adapter_base.py
index 8add7c0c..5d6cfaa1 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/_http_adapter_base.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/_http_adapter_base.py
@@ -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
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/conversation.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/conversation.py
index 079136dc..95eb579e 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/conversation.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/proactive/conversation.py
@@ -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
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/claims_identity.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/claims_identity.py
index dff15a25..0a6b26f2 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/claims_identity.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/claims_identity.py
@@ -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.
 
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py
index 1a9a14d6..00265a9e 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py
@@ -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(
diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py
index 77fa371f..6cbc700f 100644
--- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py
+++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py
@@ -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
diff --git a/pytest.ini b/pytest.ini
index f2a7ece8..b2692554 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -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.*
 
diff --git a/tests/hosting_core/app/proactive/test_conversation.py b/tests/hosting_core/app/proactive/test_conversation.py
index d6ca5adf..64814ad5 100644
--- a/tests/hosting_core/app/proactive/test_conversation.py
+++ b/tests/hosting_core/app/proactive/test_conversation.py
@@ -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)
diff --git a/tests/hosting_core/authorization/test_claims_identity.py b/tests/hosting_core/authorization/test_claims_identity.py
new file mode 100644
index 00000000..e29725f3
--- /dev/null
+++ b/tests/hosting_core/authorization/test_claims_identity.py
@@ -0,0 +1,113 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import pytest
+
+from microsoft_agents.hosting.core.authorization import ClaimsIdentity
+
+
+class TestClaimsIdentityConstructor:
+    def test_default_identity_is_anonymous(self):
+        identity = ClaimsIdentity()
+
+        assert identity.claims == {}
+        assert identity.authentication_type is None
+        assert identity.security_token is None
+        assert identity.allow_anonymous is True
+        assert identity.is_authenticated is False
+
+    def test_default_claims_are_not_shared(self):
+        first = ClaimsIdentity()
+        second = ClaimsIdentity()
+
+        first.claims["aud"] = "app-id"
+
+        assert second.claims == {}
+
+    def test_constructor_preserves_values(self):
+        claims = {"aud": "app-id"}
+
+        identity = ClaimsIdentity(
+            claims=claims,
+            authentication_type="Bearer",
+            security_token="token",
+        )
+
+        assert identity.claims is claims
+        assert identity.authentication_type == "Bearer"
+        assert identity.security_token == "token"
+        assert identity.is_authenticated is True
+
+    def test_is_authenticated_parameter_is_deprecated(self):
+        with pytest.warns(DeprecationWarning, match="is_authenticated"):
+            identity = ClaimsIdentity(is_authenticated=True)
+
+        assert identity.allow_anonymous is True
+
+
+class TestClaimsIdentityAnonymousAccess:
+    @pytest.mark.parametrize(
+        ("claims", "is_authenticated", "authentication_type", "expected"),
+        [
+            (None, None, None, True),
+            ({}, False, None, True),
+            ({}, True, None, True),
+            ({"aud": "app-id"}, None, None, False),
+            ({}, None, "Bearer", False),
+        ],
+    )
+    def test_allow_anonymous(
+        self,
+        claims,
+        is_authenticated,
+        authentication_type,
+        expected,
+    ):
+        identity = ClaimsIdentity(
+            claims=claims,
+            is_authenticated=is_authenticated,
+            authentication_type=authentication_type,
+        )
+
+        assert identity.allow_anonymous is expected
+
+    @pytest.mark.parametrize("is_authenticated", [False, True])
+    def test_deprecated_is_authenticated_does_not_affect_allow_anonymous(
+        self, is_authenticated
+    ):
+        identity = ClaimsIdentity(
+            claims={},
+            is_authenticated=is_authenticated,
+        )
+
+        assert identity.allow_anonymous is True
+
+
+class TestClaimsIdentityAuthenticationCompatibility:
+    @pytest.mark.parametrize(
+        ("claims", "expected"),
+        [
+            ({}, False),
+            ({"aud": "app-id"}, True),
+        ],
+    )
+    def test_is_authenticated_is_derived_from_claims(self, claims, expected):
+        identity = ClaimsIdentity(claims=claims)
+
+        assert identity.is_authenticated is expected
+
+    def test_is_authenticated_setter_is_deprecated_no_op(self):
+        identity = ClaimsIdentity(claims={"aud": "app-id"})
+
+        with pytest.warns(DeprecationWarning, match="is_authenticated"):
+            identity.is_authenticated = False
+
+        with pytest.warns(DeprecationWarning, match="is_authenticated"):
+            assert identity.is_authenticated is True
+
+
+def test_get_claim_value_returns_matching_claim():
+    identity = ClaimsIdentity(claims={"aud": "app-id"})
+
+    assert identity.get_claim_value("aud") == "app-id"
+    assert identity.get_claim_value("missing") is None
diff --git a/tests/hosting_core/test_channel_service_adapter.py b/tests/hosting_core/test_channel_service_adapter.py
index aa7267f2..805a0782 100644
--- a/tests/hosting_core/test_channel_service_adapter.py
+++ b/tests/hosting_core/test_channel_service_adapter.py
@@ -240,3 +240,46 @@ async def callback(context: TurnContext):
         assert context_arg.activity.service_url == "service_url"
         assert context_arg.services.get(UserTokenClientBase) is user_token_client
         assert context_arg.services.get(ConnectorClientBase) is connector_client
+
+    @pytest.mark.asyncio
+    async def test_process_proactive_uses_anonymous_clients(self, mocker):
+        factory = mocker.Mock(spec=ChannelServiceClientFactoryBase)
+        user_token_client = mocker.Mock(spec=UserTokenClient)
+        user_token_client.close = mocker.AsyncMock()
+        connector_client = mocker.Mock(spec=TeamsConnectorClient)
+        connector_client.close = mocker.AsyncMock()
+        factory.create_user_token_client = mocker.AsyncMock(
+            return_value=user_token_client
+        )
+        factory.create_connector_client = mocker.AsyncMock(
+            return_value=connector_client
+        )
+        adapter = MyChannelServiceAdapter(factory)
+        adapter.run_pipeline = mocker.AsyncMock()
+        identity = ClaimsIdentity()
+        activity = Activity(
+            type="message",
+            conversation={"id": "conversation123"},
+            channel_id="channel_id",
+            service_url="service_url",
+        )
+        callback = mocker.AsyncMock()
+
+        await adapter.process_proactive(
+            identity,
+            activity,
+            "audience",
+            callback,
+        )
+
+        context = adapter.run_pipeline.await_args.args[0]
+        factory.create_user_token_client.assert_awaited_once_with(
+            context, identity, True
+        )
+        factory.create_connector_client.assert_awaited_once_with(
+            context,
+            identity,
+            "service_url",
+            "audience",
+            use_anonymous=True,
+        )