Remove ActivitySender send abstraction

packages/apps/src/microsoft_teams/apps/activity_send.py

@@ -0,0 +1,58 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from microsoft_teams.api import (
+    ActivityParams,
+    AgenticIdentity,
+    ApiClient,
+    ConversationReference,
+    MessageActivityInput,
+    SentActivity,
+)
+
+
+async def send_or_update_activity(
+    api: ApiClient,
+    activity: ActivityParams,
+    ref: ConversationReference,
+    *,
+    agentic_identity: AgenticIdentity | None = None,
+) -> SentActivity:
+    """Send or update an activity using the same routing rules as the removed ActivitySender."""
+    is_targeted = (
+        isinstance(activity, MessageActivityInput)
+        and activity.recipient is not None
+        and activity.recipient.is_targeted is True
+    )
+
+    if is_targeted and ref.conversation.conversation_type == "personal":
+        raise ValueError("Targeted messages are not supported in 1:1 (personal) chats.")
+
+    activity.from_ = ref.bot
+    activity.conversation = ref.conversation
+
+    activities = api.conversations.activities(ref.conversation.id)
+    if activity.id:
+        activity_id = activity.id
+        if is_targeted:
+            res = await activities.update_targeted(
+                activity_id,
+                activity,
+                service_url=ref.service_url,
+            )
+        else:
+            res = await activities.update(
+                activity_id,
+                activity,
+                service_url=ref.service_url,
+                agentic_identity=agentic_identity,
+            )
+        return SentActivity.merge(activity, res)
+
+    if is_targeted:
+        res = await activities.create_targeted(activity, service_url=ref.service_url)
+    else:
+        res = await activities.create(activity, service_url=ref.service_url, agentic_identity=agentic_identity)
+    return SentActivity.merge(activity, res)

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

@@ -35,7 +35,7 @@
 if TYPE_CHECKING:
     from msgraph.graph_service_client import GraphServiceClient
 
-from .activity_sender import ActivitySender
+from .activity_send import send_or_update_activity
 from .app_events import EventManager
 from .app_oauth import OauthHandlers
 from .app_plugins import PluginProcessor
@@ -128,9 +128,6 @@ def __init__(self, **options: Unpack[AppOptions]):
         self._port: Optional[int] = None
         self._initialized = False
 
-        # initialize ActivitySender for sending activities
-        self.activity_sender = ActivitySender(self.http_client.clone(ClientOptions(token=self._get_bot_token)))
-
         # initialize all event, activity, and plugin processors
         self.activity_processor = ActivityProcessor(
             self._router,
@@ -140,7 +137,6 @@ def __init__(self, **options: Unpack[AppOptions]):
             self.http_client,
             self._token_manager,
             self.options.api_client_settings,
-            self.activity_sender,
             self.cloud,
         )
         self.event_manager = EventManager(self._events)
@@ -321,7 +317,7 @@ async def send(self, conversation_id: str, activity: str | ActivityParams | Adap
         else:
             activity = activity
 
-        return await self.activity_sender.send(activity, conversation_ref)
+        return await send_or_update_activity(self.api, activity, conversation_ref)
 
     @overload
     async def reply(
@@ -574,7 +570,6 @@ async def handler(request: HttpRequest) -> HttpResponse:
                 ctx = FunctionContext(
                     id=self.id,
                     api=self.api,
-                    activity_sender=self.activity_sender,
                     data=request["body"],
                     **client_context.__dict__,
                 )

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

@@ -26,7 +26,6 @@
 if TYPE_CHECKING:
     from .app_events import EventManager
 
-from .activity_sender import ActivitySender
 from .events import ActivityEvent, ActivityResponseEvent, ActivitySentEvent, ErrorEvent
 from .plugins import PluginActivityEvent, PluginBase, StreamCancelledError
 from .routing.activity_context import ActivityContext
@@ -49,7 +48,6 @@ def __init__(
         http_client: Client,
         token_manager: TokenManager,
         api_client_settings: Optional[ApiClientSettings],
-        activity_sender: ActivitySender,
         cloud: CloudEnvironment = PUBLIC,
     ) -> None:
         self.router = router
@@ -59,7 +57,6 @@ def __init__(
         self.http_client = http_client
         self.token_manager = token_manager
         self.api_client_settings = api_client_settings
-        self.activity_sender = activity_sender
         self.cloud = cloud
 
         # This will be set after the EventManager is initialized due to
@@ -127,7 +124,6 @@ async def _build_context(
             conversation_ref,
             is_signed_in,
             self.default_connection_name,
-            activity_sender=self.activity_sender,
             app_token=lambda: self.token_manager.get_graph_token(tenant_id),
             cloud=self.cloud,
         )

packages/apps/src/microsoft_teams/apps/contexts/function_context.py

@@ -21,7 +21,7 @@
 )
 from microsoft_teams.cards import AdaptiveCard
 
-from ..activity_sender import ActivitySender
+from ..activity_send import send_or_update_activity
 from .client_context import ClientContext
 
 T = TypeVar("T")
@@ -44,9 +44,6 @@ class FunctionContext(ClientContext, Generic[T]):
     api: ApiClient
     """The API client instance for conversation client."""
 
-    activity_sender: ActivitySender
-    """The activity sender instance for sending messages."""
-
     data: T
     """The function payload."""
 
@@ -65,21 +62,20 @@ async def send(self, activity: str | ActivityParams | AdaptiveCard) -> Optional[
             logger.warning("Cannot send activity: conversation ID could not be resolved")
             return None
 
-        conversation_ref = ConversationReference(
-            channel_id="msteams",
-            service_url=self.api.service_url,
-            bot=Account(id=self.id, name=self.name),
-            conversation=ConversationAccount(id=conversation_id, conversation_type="personal"),
-        )
-
         if isinstance(activity, str):
             activity = MessageActivityInput(text=activity)
         elif isinstance(activity, AdaptiveCard):
             activity = MessageActivityInput().add_card(activity)
         else:
             activity = activity
 
-        return await self.activity_sender.send(activity, conversation_ref)
+        conversation_ref = ConversationReference(
+            channel_id="msteams",
+            service_url=self.api.service_url,
+            bot=Account(id=self.id, name=self.name),
+            conversation=ConversationAccount(id=conversation_id, conversation_type="personal"),
+        )
+        return await send_or_update_activity(self.api, activity, conversation_ref)
 
     async def _resolve_conversation_id(self, activity: str | ActivityParams | AdaptiveCard) -> Optional[str]:
         """Resolve or create a conversation ID for the current user/context.

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

@@ -41,7 +41,9 @@
 from microsoft_teams.common.experimental import ExperimentalWarning
 from microsoft_teams.common.http.client_token import Token
 
-from ..activity_sender import ActivitySender
+from ..activity_send import send_or_update_activity
+from ..http_stream import HttpStream
+from ..plugins.streamer import StreamerProtocol
 from ..utils import create_graph_client
 
 if TYPE_CHECKING:
@@ -86,7 +88,6 @@ def __init__(
         conversation_ref: ConversationReference,
         is_signed_in: bool,
         connection_name: str,
-        activity_sender: ActivitySender,
         app_token: Token,
         cloud: CloudEnvironment = PUBLIC,
     ):
@@ -100,16 +101,21 @@ def __init__(
         self.connection_name = connection_name
         self.is_signed_in = is_signed_in
         self.cloud = cloud
-        self._activity_sender = activity_sender
         self._app_token = app_token
-        self.stream = activity_sender.create_stream(conversation_ref)
+        self._stream: Optional[StreamerProtocol] = None
 
         self._next_handler: Optional[Callable[[], Awaitable[None]]] = None
 
         # Initialize graph clients as None - they'll be created lazily
         self._user_graph: Optional["GraphServiceClient"] = None
         self._app_graph: Optional["GraphServiceClient"] = None
 
+    @property
+    def stream(self) -> StreamerProtocol:
+        if self._stream is None:
+            self._stream = HttpStream(self.api, self.conversation_ref)
+        return self._stream
+
     @property
     def user_graph(self) -> "GraphServiceClient":
         """
@@ -191,8 +197,7 @@ async def send(
         self._add_targeted_message_info_entity(activity)
 
         ref = conversation_ref or self.conversation_ref
-        res = await self._activity_sender.send(activity, ref)
-        return res
+        return await send_or_update_activity(self.api, activity, ref)
 
     async def reply(self, input: str | ActivityParams) -> SentActivity:
         """Send a message in the current conversation with a visual quote of the inbound message.