feat: expose per-turn state (ctx.state, App(state=...)) and wire into dispatch

examples/state/README.md

@@ -0,0 +1,15 @@
+# State
+
+Demonstrates the per-turn state layer: enabling it with `App(state=True)` and
+reading/writing the `conversation` and `user` scopes through `ctx.state`.
+
+State is loaded before each turn, saved automatically after it, and then sealed
+(post-turn access raises `TurnStateSealedError`). With no storage configured the
+app uses in-memory `LocalStorage`; pass `App(state=StateOptions(storage=...))`
+for a durable backing store.
+
+## Run
+
+```bash
+uv run --directory examples/state src/main.py
+```

examples/state/pyproject.toml

@@ -0,0 +1,14 @@
+[project]
+name = "state"
+version = "0.1.0"
+description = "Per-turn state app"
+readme = "README.md"
+requires-python = ">=3.11,<4.0"
+dependencies = [
+    "dotenv>=0.9.9",
+    "microsoft-teams-apps",
+    "microsoft-teams-api",
+]
+
+[tool.uv.sources]
+microsoft-teams-apps = { workspace = true }

examples/state/src/main.py

@@ -0,0 +1,50 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+import asyncio
+import logging
+
+from microsoft_teams.api import MessageActivity
+from microsoft_teams.apps import ActivityContext, App
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# `state=True` enables the per-turn state layer using the app's storage.
+#
+# With no other storage configured the app falls back to in-memory
+# `LocalStorage`, so state is scoped to a single process and lost on restart —
+# the SDK logs a warning to that effect. For production, pass a durable store
+# via `App(state=StateOptions(storage=...))`
+#
+# State is loaded before each turn and saved automatically after it, then
+# sealed — reading or writing `ctx.state` after the handler returns raises
+# `TurnStateSealedError`.
+app = App(state=True)
+
+
+@app.on_message
+async def handle_message(ctx: ActivityContext[MessageActivity]) -> None:
+    """Track a per-conversation message count and a per-user first-seen name."""
+    assert ctx.state is not None
+
+    # Conversation scope: shared by everyone in the chat/channel.
+    count = ctx.state.conversation.get("message_count", 0) + 1
+    ctx.state.conversation["message_count"] = count
+
+    # User scope: per-sender. `user` is None only when the activity has no sender.
+    greeting = ""
+    if ctx.state.user is not None:
+        if "name" not in ctx.state.user:
+            ctx.state.user["name"] = ctx.activity.from_.name
+            greeting = f"Nice to meet you, {ctx.activity.from_.name}! "
+        else:
+            greeting = f"Welcome back, {ctx.state.user['name']}! "
+
+    await ctx.send(f"{greeting}This conversation has seen {count} message(s).")
+
+
+if __name__ == "__main__":
+    asyncio.run(app.start())

packages/apps/src/microsoft_teams/apps/__init__.py

@@ -17,6 +17,7 @@
 from .options import AppOptions, AppTelemetryOptions
 from .plugins import *  # noqa: F401, F403
 from .routing import ActivityContext
+from .state import StateOptions, TurnState, TurnStateContainer, TurnStateSealedError, create_state_loader
 from .token_provider import AppTokenProvider
 from .utils.html_widget import (
     DisplayMode,
@@ -45,6 +46,11 @@
     "HttpStream",
     "ActivityContext",
     "AppTokenProvider",
+    "StateOptions",
+    "TurnState",
+    "TurnStateContainer",
+    "TurnStateSealedError",
+    "create_state_loader",
     "to_threaded_conversation_id",
     "build_html_widget_markdown",
     "build_html_widget_message",

packages/apps/src/microsoft_teams/apps/app.py

@@ -60,6 +60,7 @@
 from .plugins import PluginBase, PluginStartEvent
 from .routing import ActivityHandlerMixin, ActivityRouter
 from .routing.activity_context import ActivityContext
+from .state import create_state_loader
 from .token_manager import DEFAULT_TENANT_FOR_GRAPH_TOKEN, TokenManager
 from .token_provider import AppTokenProvider
 from .utils import create_graph_client
@@ -92,6 +93,8 @@ def __init__(self, **options: Unpack[AppOptions]):
 
         self.storage = self.options.storage or LocalStorage()
 
+        self._state_loader = create_state_loader(self.options.state, self.storage)
+
         self.http_client = self._init_http_client()
 
         self._events = EventEmitter[EventType]()
@@ -143,6 +146,7 @@ def __init__(self, **options: Unpack[AppOptions]):
             self.cloud,
             fetch_user_token=self.options.fetch_user_token,
             agent365_baggage_options=self.options.telemetry.get("agent365") if self.options.telemetry else None,
+            state_loader=self._state_loader,
         )
         self.event_manager = EventManager(self._events)
         self.activity_processor.event_manager = self.event_manager

packages/apps/src/microsoft_teams/apps/app_process.py

@@ -46,6 +46,7 @@
 from .plugins import PluginActivityEvent, PluginBase, StreamCancelledError
 from .routing.activity_context import ActivityContext
 from .routing.router import ActivityHandler, ActivityRouter
+from .state import TurnStateLoader
 from .token_provider import AppTokenProvider
 from .utils import extract_tenant_id
 
@@ -74,6 +75,7 @@ def __init__(
         cloud: CloudEnvironment = PUBLIC,
         fetch_user_token: bool = True,
         agent365_baggage_options: Agent365BaggageOptions | bool | None = None,
+        state_loader: Optional[TurnStateLoader] = None,
     ) -> None:
         self.router = router
         self.id = id
@@ -86,6 +88,7 @@ def __init__(
         self.cloud = cloud
         self.fetch_user_token = fetch_user_token
         self.agent365_baggage_options = agent365_baggage_options
+        self.state_loader = state_loader
 
         # This will be set after the EventManager is initialized due to
         # a circular dependency
@@ -288,6 +291,8 @@ async def route(ctx: ActivityContext[ActivityBase]) -> Optional[Any]:
             raise ValueError("EventManager was not initialized properly")
 
         try:
+            await self._load_turn_state(activityCtx, activity)
+
             # If no registered handlers, middleware_result is set to None
             middleware_result = await self.execute_middleware_chain(activityCtx, handlers)
 
@@ -312,12 +317,40 @@ async def route(ctx: ActivityContext[ActivityBase]) -> Optional[Any]:
             response = InvokeResponse[Any](status=200)
         except Exception as error:
             await self.event_manager.on_error(ErrorEvent(error=error, activity=activity), plugins)
-            raise error
+            raise
+        finally:
+            await self._persist_turn_state(activityCtx)
 
         logger.debug("Completed processing activity")
 
         return response
 
+    async def _load_turn_state(self, ctx: ActivityContext[ActivityBase], activity: ValidatedActivity) -> None:
+        """Load per-turn state onto ``ctx.state`` when state is enabled.
+
+        Loads both the conversation scope and the user scope (keyed by the
+        activity's ``from`` identity). A no-op when state is disabled, leaving
+        ``ctx.state`` as ``None``.
+        """
+        if self.state_loader is None:
+            return
+        ctx.state = await self.state_loader.load(activity.conversation.id, activity.from_.id)
+
+    async def _persist_turn_state(self, ctx: ActivityContext[ActivityBase]) -> None:
+        """Save dirty scopes and seal state at the end of the turn.
+
+        Runs in a ``finally`` so dirty state is persisted even when the handler
+        raised. Sealing makes any post-turn access raise, guarding against use of
+        per-turn state in background work.
+        """
+        container = ctx.state
+        if self.state_loader is None or container is None:
+            return
+        try:
+            await self.state_loader.save(container)
+        finally:
+            container.seal()
+
     def _activity_attributes(self, activity: ActivityBase) -> dict[str, str]:
         attributes = {
             APP_ATTRIBUTE_NAMES.activity_type: activity.type,

packages/apps/src/microsoft_teams/apps/options.py

@@ -19,6 +19,7 @@
 from .diagnostics import Agent365BaggageOptions
 from .http.adapter import HttpServerAdapter
 from .plugins import PluginBase
+from .state import StateOptions
 
 DANGEROUSLY_ALLOW_UNAUTHENTICATED_REQUESTS_ENV_VAR = "DANGEROUSLY_ALLOW_UNAUTHENTICATED_REQUESTS"
 _TRUE_ENV_VALUES = {"1", "true", "yes", "on"}
@@ -88,6 +89,14 @@ class AppOptions(TypedDict, total=False):
     # Infrastructure
     storage: Optional[Storage[str, Any]]
     plugins: Optional[List[PluginBase]]
+    state: Optional[Union[bool, StateOptions]]
+    """Per-turn state opt-in. Off by default (``None``/``False``).
+
+    ``state=True`` enables state on the app's shared ``storage`` (in-memory by
+    default). Pass a ``StateOptions`` to configure the
+    key prefix, TTL, or a dedicated ``Storage`` backend. When enabled, handlers
+    read/write ``ctx.state.conversation`` and ``ctx.state.user``; when off,
+    ``ctx.state`` is ``None``."""
     dangerously_allow_unauthenticated_requests: Optional[bool]
     """
     Whether to accept incoming requests without JWT validation.
@@ -183,6 +192,9 @@ class InternalAppOptions:
     If not set or equals client_id, uses direct managed identity (no federation).
     """
     storage: Optional[Storage[str, Any]] = None
+    state: Optional[Union[bool, StateOptions]] = None
+    """Per-turn state opt-in. ``None``/``False`` disables it; ``True`` enables it on
+    the app's shared storage; a ``StateOptions`` configures prefix/TTL/backend."""
     service_url: Optional[str] = None
     """
     Base Service URL for BotBackend.

packages/apps/src/microsoft_teams/apps/routing/activity_context.py

@@ -44,6 +44,7 @@
 from ..files import FilesAccessor
 from ..http_stream import HttpStream
 from ..plugins.streamer import StreamerProtocol
+from ..state import TurnStateContainer
 from ..utils import create_graph_client
 
 if TYPE_CHECKING:
@@ -101,6 +102,7 @@ def __init__(
         self.connection_name = connection_name
         self.is_signed_in = is_signed_in
         self.cloud = cloud
+        self.state: Optional[TurnStateContainer] = None
         self._app_token = app_token
         self._stream: Optional[StreamerProtocol] = None
         self._files: Optional[FilesAccessor] = None

packages/apps/src/microsoft_teams/apps/state/__init__.py

@@ -4,7 +4,7 @@
 """
 
 from .container import TurnStateContainer
-from .loader import TurnStateLoader
+from .loader import TurnStateLoader, create_state_loader
 from .options import StateOptions
 from .turn_state import TurnState, TurnStateSealedError
 
@@ -14,4 +14,5 @@
     "TurnStateContainer",
     "TurnStateLoader",
     "StateOptions",
+    "create_state_loader",
 ]

packages/apps/src/microsoft_teams/apps/state/loader.py

@@ -8,10 +8,10 @@
 import json
 import logging
 import time
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional, Union, cast
 from urllib.parse import quote
 
-from microsoft_teams.common import Storage
+from microsoft_teams.common import LocalStorage, Storage
 
 from .container import TurnStateContainer
 from .options import StateOptions
@@ -159,3 +159,28 @@ def _is_expired(self, blob: Dict[str, Any]) -> bool:
 
         saved_at = blob.get("ts")
         return not isinstance(saved_at, (int, float)) or (time.time() - saved_at) > self._options.ttl
+
+
+def create_state_loader(
+    state: Optional[Union[bool, "StateOptions"]],
+    fallback_storage: Storage[str, Any],
+) -> Optional[TurnStateLoader]:
+    """Resolve the ``App(state=...)`` option into a loader (or ``None`` when off).
+
+    ``state`` is the opt-in value: falsy disables state; ``True`` enables it with
+    defaults; a ``StateOptions`` configures it. The loader's storage is the one on
+    ``StateOptions`` when provided, otherwise the app's shared ``fallback_storage``.
+    A warning is logged when that resolves to in-memory ``LocalStorage``.
+    """
+    if not state:
+        return None
+
+    options = StateOptions() if state is True else state
+    storage: Storage[str, Any] = options.storage if options.storage is not None else fallback_storage
+
+    if isinstance(storage, LocalStorage):
+        logger.warning(
+            "State is enabled with in-memory storage (LocalStorage): per-turn state is lost on "
+            + "restart and is not shared across instances."
+        )
+    return TurnStateLoader(storage=cast(Storage[str, str], storage), options=options)