Fixing _FlowStorageClient write cache usage

README.md

@@ -42,18 +42,14 @@ The packages should target Python 3.10 or greater, and can be used with modern P
 
 > Note: We recommend using Python 3.11 or later for optimal performance and compatibility with all features. The SDK supports Python 3.10, 3.11, 3.12, 3.13, and 3.14.
 
-### Debugging
-
-The packages include source code to allow debugging in your preferred Python IDE or debugger.
-
 ### Code Style
 
 We are using `black` and `flake8` for code formatting and linting.
 
 ## Contributing
 
-#### Note for Microsoft intenral developers: 
-- Internal Micrsoft Developers should join the Core identity group [Agents SDK Contrib](https://coreidentity.microsoft.com/manage/Entitlement/entitlement/agentssdkint-upyj)
+#### Note for Microsoft internal developers: 
+- Internal Microsoft Developers should join the Core identity group [Agents SDK Contrib](https://coreidentity.microsoft.com/manage/Entitlement/entitlement/agentssdkint-upyj)
 
 #### Non-Microsoft internal developers:
 

changelog.md

@@ -7,6 +7,9 @@
 ## New Models & APIs
 - **Regionalized UserTokenClient Support**: Added optional argument to `CloudAdapter` to configure Token Service endpoint used by `RestChannelServiceClientFactory` when creating `UserTokenClient` instances. 
 
+## Bug Fixes
+- **OAuth Flow Storage**: Avoided redundant writes when the flow state is unchanged in the cache.
+
 # Microsoft 365 Agents SDK for Python - Release Notes v1.3.0
 
 **Release Date:** 2026-07-30

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/_oauth/_flow_storage_client.py

@@ -1,8 +1,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
-from typing import Optional
-
 from ..storage import Storage
 from ._flow_state import _FlowState
 
@@ -34,7 +32,7 @@ def __init__(
         channel_id: str,
         user_id: str,
         storage: Storage,
-        cache_class: Optional[type[Storage]] = None,
+        cache_class: type[Storage] | None = None,
     ):
         """
         Args:
@@ -65,7 +63,7 @@ def key(self, auth_handler_id: str) -> str:
         """Creates a storage key for a specific sign-in handler."""
         return f"{self._base_key}{auth_handler_id}"
 
-    async def read(self, auth_handler_id: str) -> Optional[_FlowState]:
+    async def read(self, auth_handler_id: str) -> _FlowState | None:
         """Reads the flow state for a specific authentication handler."""
         key: str = self.key(auth_handler_id)
         data = await self._cache.read([key], target_cls=_FlowState)
@@ -74,13 +72,13 @@ async def read(self, auth_handler_id: str) -> Optional[_FlowState]:
             if key not in data:
                 return None
             await self._cache.write({key: data[key]})
-        return _FlowState.model_validate(data.get(key))
+        return data.get(key)
 
     async def write(self, value: _FlowState) -> None:
         """Saves the flow state for a specific authentication handler."""
         key: str = self.key(value.auth_handler_id)
         cached_state = await self._cache.read([key], target_cls=_FlowState)
-        if not cached_state or cached_state != value:
+        if not cached_state or cached_state.get(key, None) != value:
             await self._cache.write({key: value})
             await self._storage.write({key: value})
 

tests/hosting_core/_oauth/test_flow_storage_client.py

@@ -86,6 +86,25 @@ async def test_write(self, mocker, auth_handler_id):
         await client.write(flow_state)
         storage.write.assert_called_once_with({client.key(auth_handler_id): flow_state})
 
+    @pytest.mark.asyncio
+    async def test_write_skips_unchanged_cached_state(self, mocker):
+        storage = mocker.AsyncMock()
+        cache = mocker.AsyncMock()
+        flow_state = _FlowState(auth_handler_id="handler")
+        key = f"auth/{DEFAULTS.channel_id}/{DEFAULTS.user_id}/handler"
+        cache.read.return_value = {key: flow_state.model_copy()}
+        client = _FlowStorageClient(
+            DEFAULTS.channel_id,
+            DEFAULTS.user_id,
+            storage,
+            cache_class=mocker.Mock(return_value=cache),
+        )
+
+        await client.write(flow_state)
+
+        cache.write.assert_not_called()
+        storage.write.assert_not_called()
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("auth_handler_id", ["handler", "auth_handler"])
     async def test_delete(self, mocker, auth_handler_id):