diff --git a/examples/stream/README.md b/examples/stream/README.md index ab6e52b88..99d0d7192 100644 --- a/examples/stream/README.md +++ b/examples/stream/README.md @@ -2,6 +2,10 @@ A test application that demonstrates streaming functionality. +- Send any message for the normal single-stream demo with suggested actions. +- Send `simple-card` to send a minimal Adaptive Card outside the streaming flow. +- Send `multi-stream` to test emitting an Adaptive Card as part of the first stream final message, finalizing with `close()`, and then reusing `ctx.stream` for another streamed response. + ## Running ```bash diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index 27e69f202..16e305b4b 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -9,6 +9,7 @@ from microsoft_teams.api import CardAction, CardActionType, MessageActivity, MessageActivityInput, SuggestedActions from microsoft_teams.apps import ActivityContext, App +from microsoft_teams.cards import AdaptiveCard, TextBlock # Surface SDK INFO/WARNING logs (including the anonymous-mode startup warning # emitted when CLIENT_ID / CLIENT_SECRET / TENANT_ID are not configured). @@ -31,11 +32,75 @@ "✨ Stream test complete!", ] +FIRST_STREAM_MESSAGES = [ + "[stream 1] Starting the first streamed response. ", + "[stream 1] This is using the default ctx.stream instance. ", + "[stream 1] Next the handler will close the current streamed message.", +] + +SECOND_STREAM_MESSAGES = [ + "[stream 2] Reusing ctx.stream after emit reopens the closed stream. ", + "[stream 2] This should render after stream 1's final Adaptive Card message. ", + "[stream 2] The app processor will close this stream when the handler returns.", +] + + +def should_run_multi_stream(text: str | None) -> bool: + normalized = (text or "").lower().replace("-", " ") + return "multi stream" in normalized + + +def should_send_simple_card(text: str | None) -> bool: + normalized = (text or "").lower().replace("-", " ") + return "simple card" in normalized + + +def create_simple_card() -> AdaptiveCard: + return AdaptiveCard( + version="1.4", + body=[ + TextBlock(text="Simple Adaptive Card", weight="Bolder", size="Large", wrap=True), + TextBlock(text="If you can see this card, basic Adaptive Card delivery is working.", wrap=True), + ], + ) + @app.on_message async def handle_message(ctx: ActivityContext[MessageActivity]): """Stream messages to the user on any message activity.""" + if should_send_simple_card(ctx.activity.text): + sent_card = await ctx.send( + MessageActivityInput(text="Sending a simple Adaptive Card.").add_card(create_simple_card()) + ) + logger.info("Sent simple adaptive card: %s", sent_card.id) + return + + if should_run_multi_stream(ctx.activity.text): + ctx.stream.update("Starting stream 1...") + await asyncio.sleep(1) + + for message in FIRST_STREAM_MESSAGES: + await asyncio.sleep(0.5) + ctx.stream.emit(message) + + card_message = MessageActivityInput(text="Adaptive Card emitted as part of stream 1.").add_card( + create_simple_card() + ) + ctx.stream.emit(card_message) + sent_message = await ctx.stream.close() + if sent_message: + logger.info("Sent stream 1 final message with adaptive card: %s", sent_message.id) + await asyncio.sleep(2) + + ctx.stream.update("Starting stream 2...") + await asyncio.sleep(1) + + for message in SECOND_STREAM_MESSAGES: + await asyncio.sleep(0.5) + ctx.stream.emit(message) + return + ctx.stream.update("Stream starting...") await asyncio.sleep(1) diff --git a/packages/apps/src/microsoft_teams/apps/http_stream.py b/packages/apps/src/microsoft_teams/apps/http_stream.py index 05f188fad..a70cf2fac 100644 --- a/packages/apps/src/microsoft_teams/apps/http_stream.py +++ b/packages/apps/src/microsoft_teams/apps/http_stream.py @@ -62,10 +62,10 @@ def __init__(self, client: ApiClient, ref: ConversationReference): self._state_changed = asyncio.Event() self._canceled = False - self._reset_state() + self._reset_current_stream() - def _reset_state(self) -> None: - """Reset the stream state to initial values.""" + def _reset_current_stream(self) -> None: + """Reset per-message state for the current stream cycle.""" self._index = 1 self._id: Optional[str] = None self._text: str = "" @@ -73,6 +73,11 @@ def _reset_state(self) -> None: self._final_activity: Optional[MessageActivityInput] = None self._queue: deque[Union[MessageActivityInput, TypingActivityInput, str]] = deque() + def _reset_stream_for_next_stream(self) -> None: + """Prepare the stream instance to start a new stream cycle.""" + self._reset_current_stream() + self._result = None + @property def canceled(self) -> bool: """ @@ -83,7 +88,7 @@ def canceled(self) -> bool: @property def closed(self) -> bool: - """Whether the final stream message has been sent.""" + """Whether the current streamed message has been finalized.""" return self._result is not None @property @@ -100,7 +105,7 @@ def on_chunk(self, handler: Callable[[SentActivity], Awaitable[None]]): self._events.on("chunk", handler) def on_close(self, handler: Callable[[SentActivity], Awaitable[None]]): - self._events.once("close", handler) + self._events.on("close", handler) def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) -> None: """ @@ -113,6 +118,10 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) if self._canceled: raise StreamCancelledError("Stream has been cancelled.") + if self.closed: + logger.debug("starting a new streamed message after close") + self._reset_stream_for_next_stream() + if isinstance(activity, str): activity = MessageActivityInput(text=activity, type="message") self._queue.append(activity) @@ -163,9 +172,16 @@ async def _poll(): return False async def close(self) -> Optional[SentActivity]: - # wait for lock to be free - if self._result is not None: + """ + Finalize the current streamed message. + + Closing is idempotent until the next emit or update. Emitting or + updating after close starts a new streamed message using the same + stream instance. + """ + if self.closed: logger.debug("stream already closed with result") + assert self._result is not None return self._result if self._canceled: @@ -203,8 +219,8 @@ async def close(self) -> Optional[SentActivity]: # Emit close event self._events.emit("close", res) - # Reset state - self._reset_state() + # Reset per-message state; keep event handlers and cancellation state. + self._reset_current_stream() self._result = res logger.debug("stream closed with result: %s", res) diff --git a/packages/apps/src/microsoft_teams/apps/plugins/streamer.py b/packages/apps/src/microsoft_teams/apps/plugins/streamer.py index 20a05ea1f..4f44895e2 100644 --- a/packages/apps/src/microsoft_teams/apps/plugins/streamer.py +++ b/packages/apps/src/microsoft_teams/apps/plugins/streamer.py @@ -30,7 +30,7 @@ def canceled(self) -> bool: @property def closed(self) -> bool: - """Whether the final stream message has been sent.""" + """Whether the current streamed message has been finalized.""" ... @property @@ -59,10 +59,10 @@ def on_chunk(self, handler: Callable[[SentActivity], Awaitable[None]]) -> None: def on_close(self, handler: Callable[[SentActivity], Awaitable[None]]) -> None: """ - Register a handler for close events. + Register a handler for stream close events. Args: - handler: Async function that will be called when the stream closes + handler: Async function that will be called each time the stream closes """ ... @@ -89,6 +89,10 @@ def clear_text(self) -> None: async def close(self) -> Optional[SentActivity]: """ - Close the stream. + Finalize the current streamed message. + + Closing is idempotent until the next emit or update. Emitting or + updating after close starts a new streamed message using the same + stream instance. """ ... diff --git a/packages/apps/tests/test_http_stream.py b/packages/apps/tests/test_http_stream.py index 53173c4cb..802109608 100644 --- a/packages/apps/tests/test_http_stream.py +++ b/packages/apps/tests/test_http_stream.py @@ -453,3 +453,49 @@ async def test_close_waits_for_flush_to_complete(self, mock_api_client, conversa assert result is not None assert mock_api_client.send_call_count == 1 assert mock_api_client.sent_activities[0].text == "Response text" + + @pytest.mark.asyncio + async def test_emit_after_close_reopens_stream_for_next_message( + self, mock_api_client, conversation_reference, patch_loop_call_later + ): + loop = asyncio.get_running_loop() + patcher, scheduled = patch_loop_call_later(loop) + close_results: list[SentActivity] = [] + close_event = asyncio.Event() + + async def handle_close(activity: SentActivity) -> None: + close_results.append(activity) + close_event.set() + + with patcher: + stream = HttpStream(mock_api_client, conversation_reference) + stream.on_close(handle_close) + + stream.emit("First streamed message") + await asyncio.sleep(0) + await self._run_scheduled_flushes(scheduled) + + first_result = await stream.close() + await close_event.wait() + close_event.clear() + assert first_result is not None + assert stream.closed is True + + repeated_result = await stream.close() + assert repeated_result is first_result + assert stream.closed is True + + stream.emit("Second streamed message") + assert stream.closed is False + await asyncio.sleep(0) + await self._run_scheduled_flushes(scheduled) + + second_result = await stream.close() + await close_event.wait() + assert second_result is not None + assert stream.closed is True + + assert first_result.id != second_result.id + assert first_result.activity_params.text == "First streamed message" + assert second_result.activity_params.text == "Second streamed message" + assert [result.id for result in close_results] == [first_result.id, second_result.id] diff --git a/uv.lock b/uv.lock index 3ef2ba7ac..9cd6c7189 100644 --- a/uv.lock +++ b/uv.lock @@ -14,6 +14,7 @@ members = [ "cards", "dialogs", "echo", + "formatted-messaging", "graph", "http-adapters", "mcp-server", @@ -967,6 +968,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b5/36/7fb70f04bf00bc646cd5bb45aa9eddb15e19437a28b8fb2b4a5249fac770/filelock-3.20.3-py3-none-any.whl", hash = "sha256:4b0dda527ee31078689fc205ec4f1c1bf7d56cf88b6dc9426c4f230e46c2dce1", size = 16701, upload-time = "2026-01-09T17:55:04.334Z" }, ] +[[package]] +name = "formatted-messaging" +version = "0.1.0" +source = { virtual = "examples/formatted-messaging" } +dependencies = [ + { name = "dotenv" }, + { name = "microsoft-teams-api" }, + { name = "microsoft-teams-apps" }, +] + +[package.metadata] +requires-dist = [ + { name = "dotenv", specifier = ">=0.9.9" }, + { name = "microsoft-teams-api", editable = "packages/api" }, + { name = "microsoft-teams-apps", editable = "packages/apps" }, +] + [[package]] name = "frozenlist" version = "1.8.0" @@ -1816,7 +1834,7 @@ test = [ [package.metadata] requires-dist = [ - { name = "cryptography", specifier = ">=48.0.1" }, + { name = "cryptography", specifier = ">=3.4.0" }, { name = "dependency-injector", specifier = ">=4.48.1" }, { name = "fastapi", specifier = ">=0.115.13" }, { name = "microsoft-teams-api", editable = "packages/api" },