add AU support for socket mode

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

@@ -235,10 +235,28 @@ def _build_socket_mode(self) -> Optional[SocketModeAdapter]:
                 "Socket Mode is not supported in this cloud environment. Use the HTTP inbound transport instead."
             )
 
+        socket_options = option if isinstance(option, SocketModeOptions) else SocketModeOptions()
+        get_token = self._get_bot_token
+        agentic_app_id = socket_options.agentic_app_id
+        scope = socket_options.agentic_token_scope
+        if agentic_app_id is None:
+            if scope is not None or socket_options.agentic_tenant_id is not None:
+                raise ValueError("agentic_app_id is required with agentic socket authentication options")
+        else:
+            if not scope:
+                raise ValueError("agentic_token_scope is required with agentic_app_id")
+
+            async def get_agentic_token() -> Optional[TokenProtocol]:
+                return await self._token_provider.get_agentic_app_token(
+                    scope, agentic_app_id, socket_options.agentic_tenant_id
+                )
+
+            get_token = get_agentic_token
+
         return SocketModeAdapter(
-            option if isinstance(option, SocketModeOptions) else SocketModeOptions(),
+            socket_options,
             process_activity=self._process_activity_event,
-            get_app_token=self._get_bot_token,
+            get_app_token=get_token,
             messaging_endpoint=self.options.messaging_endpoint,
             client_id=self.credentials.client_id if self.credentials else None,
             on_error=self._on_socket_mode_error,

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

@@ -120,7 +120,7 @@ class AppOptions(TypedDict, total=False):
     """Experimental: receive activities over an outbound WebSocket instead of an inbound HTTPS endpoint.
 
     ``True`` enables it with defaults; pass a ``SocketModeOptions`` to tune geos and
-    timeouts. The bot dials the Teams service, so no public endpoint or dev tunnel is
+    timeouts or opt into agentic identity auth. The bot dials the Teams service, so no public endpoint or dev tunnel is
     needed — but HTTP-only features (``app.function()``, tabs, OAuth redirect routes)
     have no listener to serve them. Mutually exclusive with ``http_server_adapter``,
     and supported only in the public cloud.

packages/apps/src/microsoft_teams/apps/socket_mode/adapter.py

@@ -22,7 +22,13 @@
 
 from ..events import ActivityEvent, CoreActivity
 from ..http.adapter import HttpMethod, HttpRouteHandler
-from .envelope import build_reply_frame, is_invoke_envelope, read_envelope_activity
+from .envelope import (
+    EnvelopeError,
+    build_reply_frame,
+    is_invoke_envelope,
+    read_envelope_activity,
+    validate_agentic_recipient,
+)
 from .transport import (
     DEFAULT_GEOS,
     DEFAULT_SOCKET_MODE_NEGOTIATE_BASE_URL,
@@ -109,6 +115,15 @@ class SocketModeOptions:
     server_timeout: float = 30.0
     """How long without an inbound message before the connection is considered lost."""
 
+    agentic_app_id: Optional[str] = None
+    """Agent instance used to authenticate the socket; App.client_id remains the blueprint ID."""
+
+    agentic_token_scope: Optional[str] = None
+    """Backend-agreed negotiate scope, required with agentic_app_id; no audience is assumed."""
+
+    agentic_tenant_id: Optional[str] = None
+    """Tenant for socket authentication. Defaults to the app's credential tenant."""
+
     def to_transport_options(self) -> SocketModeTransportOptions:
         return SocketModeTransportOptions(
             negotiate_base_url=self.negotiate_base_url,
@@ -176,7 +191,7 @@ def __init__(
     ):
         self.options = options or SocketModeOptions()
         self._process_activity = process_activity
-        self._get_app_token = get_app_token
+        self._get_token = get_app_token
         self._messaging_endpoint = messaging_endpoint
         self._client_id = client_id
         self._on_error = on_error
@@ -263,8 +278,10 @@ async def stop(self) -> None:
         await self._transport.stop()
 
     async def _acquire_bot_token(self) -> str:
-        token = await self._get_app_token()
-        if token is None:
+        token = await self._get_token()
+        if token is None or not str(token).strip():
+            if self.options.agentic_app_id is not None:
+                raise RuntimeError("Socket Mode could not acquire an agent-instance token")
             raise RuntimeError(
                 "Socket Mode could not acquire a Bot Framework app token. "
                 "Check that the app credentials (client_id / client_secret / tenant_id) are configured."
@@ -306,6 +323,22 @@ async def _handle_envelope(self, envelope: SocketActivityEnvelope) -> Optional[R
             logger.warning("socket-mode: inbound envelope %s carried no activity; dropping", envelope.envelope_id)
             return None
 
+        try:
+            if self.options.agentic_app_id is not None:
+                if envelope.bot_key is not None and envelope.bot_key != self._client_id:
+                    raise EnvelopeError("Socket Mode botKey does not match the configured blueprint ID")
+                validate_agentic_recipient(activity.get("recipient"), blueprint_id=self._client_id)
+        except EnvelopeError as error:
+            logger.warning("socket-mode: rejecting envelope %s: %s", envelope.envelope_id, error)
+            await self._report_error(error)
+            return build_reply_frame(
+                envelope,
+                bot_key=self._client_id,
+                status=400,
+                body={"error": str(error)} if is_invoke_envelope(envelope) else None,
+                received_at=received_at,
+            )
+
         invoke = is_invoke_envelope(envelope)
         logger.debug(
             "socket-mode: recv kind=%s type=%s envelope=%s",

packages/apps/src/microsoft_teams/apps/socket_mode/envelope.py

@@ -86,6 +86,20 @@ def is_invoke_envelope(envelope: SocketActivityEnvelope) -> bool:
     return activity is not None and str(activity.get("type", "")).lower() == "invoke"
 
 
+def validate_agentic_recipient(value: object, *, blueprint_id: Optional[str]) -> None:
+    if not isinstance(value, Mapping):
+        raise EnvelopeError("Agentic Socket Mode requires a recipient identity")
+    recipient = cast(Mapping[str, object], value)
+    required_fields = ["agenticAppBlueprintId", "agenticAppId", "tenantId"]
+    if recipient.get("role") == "agenticUser" or recipient.get("agenticUserId") is not None:
+        required_fields.append("agenticUserId")
+    values = (recipient.get(field) for field in required_fields)
+    if not all(isinstance(value, str) and value and value == value.strip() for value in values):
+        raise EnvelopeError(f"Agentic recipient requires non-empty, unpadded strings for {', '.join(required_fields)}")
+    if recipient["agenticAppBlueprintId"] != blueprint_id:
+        raise EnvelopeError("Agentic recipient blueprint does not match the configured client_id")
+
+
 def build_reply_frame(
     envelope: SocketActivityEnvelope,
     *,