From 69f881894871754e8dd63ae3a440340dbcb2f433 Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Wed, 1 Jul 2026 18:04:06 -0700 Subject: [PATCH 01/10] Support resetting response streams Add a reset option to stream close so handlers can finalize one streamed message and reuse the same stream instance for a later streamed message. Update the stream example with a multi-stream flow and cover the lifecycle in HttpStream tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/README.md | 3 ++ examples/stream/src/main.py | 40 ++++++++++++++++++ .../src/microsoft_teams/apps/http_stream.py | 27 +++++++++--- .../microsoft_teams/apps/plugins/streamer.py | 17 ++++++-- packages/apps/tests/test_http_stream.py | 42 +++++++++++++++++++ 5 files changed, 119 insertions(+), 10 deletions(-) diff --git a/examples/stream/README.md b/examples/stream/README.md index ab6e52b88..85a19f04e 100644 --- a/examples/stream/README.md +++ b/examples/stream/README.md @@ -2,6 +2,9 @@ A test application that demonstrates streaming functionality. +- Send any message for the normal single-stream demo with suggested actions. +- Send `multi-stream` to test finalizing the current stream with `close(reset=True)`, sending a normal message, 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..385a0753e 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -31,11 +31,51 @@ "✨ 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 with reset=True.", +] + +SECOND_STREAM_MESSAGES = [ + "[stream 2] Reusing ctx.stream after close(reset=True). ", + "[stream 2] This should render after the non-stream checkpoint 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 + @app.on_message async def handle_message(ctx: ActivityContext[MessageActivity]): """Stream messages to the user on any message activity.""" + 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) + + await ctx.stream.close(reset=True) + await asyncio.sleep(1) + + sent_message = await ctx.send("NON-STREAM MESSAGE BETWEEN STREAMS") + logger.info("Sent checkpoint message: %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..b9d246cbb 100644 --- a/packages/apps/src/microsoft_teams/apps/http_stream.py +++ b/packages/apps/src/microsoft_teams/apps/http_stream.py @@ -100,7 +100,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: """ @@ -162,11 +162,25 @@ async def _poll(): except asyncio.TimeoutError: return False - async def close(self) -> Optional[SentActivity]: - # wait for lock to be free + async def close(self, reset: bool = False) -> Optional[SentActivity]: + """ + Finalize the current streamed message. + + By default, closing is terminal and idempotent: subsequent calls return + the same final activity. When reset is True, the current streamed + message is finalized and the stream is reset so the same instance can + be used for another streamed message. + + Args: + reset: Whether to reset the stream after finalizing the current + streamed message. + """ if self._result is not None: logger.debug("stream already closed with result") - return self._result + result = self._result + if reset: + self._result = None + return result if self._canceled: logger.debug("stream was cancelled, nothing to close") @@ -203,9 +217,10 @@ async def close(self) -> Optional[SentActivity]: # Emit close event self._events.emit("close", res) - # Reset state + # Reset per-message state; keep event handlers and cancellation state. self._reset_state() - self._result = res + if not reset: + self._result = res logger.debug("stream closed with result: %s", res) return 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..811dbc2bb 100644 --- a/packages/apps/src/microsoft_teams/apps/plugins/streamer.py +++ b/packages/apps/src/microsoft_teams/apps/plugins/streamer.py @@ -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 """ ... @@ -87,8 +87,17 @@ def clear_text(self) -> None: """ ... - async def close(self) -> Optional[SentActivity]: + async def close(self, reset: bool = False) -> Optional[SentActivity]: """ - Close the stream. + Finalize the current streamed message. + + By default, closing is terminal and idempotent: subsequent calls return + the same final activity. When reset is True, the current streamed + message is finalized and the stream is reset so the same instance can + be used for another streamed message. + + Args: + reset: Whether to reset the stream after finalizing the current + streamed message. """ ... diff --git a/packages/apps/tests/test_http_stream.py b/packages/apps/tests/test_http_stream.py index 53173c4cb..411d9c0ac 100644 --- a/packages/apps/tests/test_http_stream.py +++ b/packages/apps/tests/test_http_stream.py @@ -453,3 +453,45 @@ 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_close_reset_reuses_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(reset=True) + await close_event.wait() + close_event.clear() + assert first_result is not None + assert stream.closed is False + assert stream.sequence == 1 + + stream.emit("Second streamed message") + await asyncio.sleep(0) + await self._run_scheduled_flushes(scheduled) + second_result = await stream.close() + 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] From 76c2ead20f7d01bca2d443489c44d887b1879052 Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Wed, 1 Jul 2026 18:04:48 -0700 Subject: [PATCH 02/10] Update workspace lockfile Include the formatted-messaging workspace member and refreshed dependency metadata in uv.lock. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- uv.lock | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) 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" }, From 054c9d76868f8d536500082d7e79fc8fe5e3ae3d Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Wed, 1 Jul 2026 18:10:03 -0700 Subject: [PATCH 03/10] Add simple adaptive card stream example Add a standalone simple-card path to the stream example for validating Adaptive Card delivery independently of streaming reset behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/README.md | 1 + examples/stream/src/main.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/examples/stream/README.md b/examples/stream/README.md index 85a19f04e..b3421f7c3 100644 --- a/examples/stream/README.md +++ b/examples/stream/README.md @@ -3,6 +3,7 @@ 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 finalizing the current stream with `close(reset=True)`, sending a normal message, and then reusing `ctx.stream` for another streamed response. ## Running diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index 385a0753e..ffab7be79 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 # Surface SDK INFO/WARNING logs (including the anonymous-mode startup warning # emitted when CLIENT_ID / CLIENT_SECRET / TENANT_ID are not configured). @@ -49,10 +50,45 @@ def should_run_multi_stream(text: str | None) -> bool: 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.model_validate( + { + "type": "AdaptiveCard", + "version": "1.4", + "body": [ + { + "type": "TextBlock", + "text": "Simple Adaptive Card", + "weight": "Bolder", + "size": "Large", + "wrap": True, + }, + { + "type": "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) From 4b93eccc396e08ffe62ab39e1a56db24dcdb378c Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Wed, 1 Jul 2026 18:12:00 -0700 Subject: [PATCH 04/10] Use adaptive card in multi-stream example Update the multi-stream stream example to send the simple Adaptive Card between streamed response segments. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/README.md | 2 +- examples/stream/src/main.py | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/examples/stream/README.md b/examples/stream/README.md index b3421f7c3..bb461f1e9 100644 --- a/examples/stream/README.md +++ b/examples/stream/README.md @@ -4,7 +4,7 @@ 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 finalizing the current stream with `close(reset=True)`, sending a normal message, and then reusing `ctx.stream` for another streamed response. +- Send `multi-stream` to test finalizing the current stream with `close(reset=True)`, sending an Adaptive Card, and then reusing `ctx.stream` for another streamed response. ## Running diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index ffab7be79..fdf739384 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -100,8 +100,9 @@ async def handle_message(ctx: ActivityContext[MessageActivity]): await ctx.stream.close(reset=True) await asyncio.sleep(1) - sent_message = await ctx.send("NON-STREAM MESSAGE BETWEEN STREAMS") - logger.info("Sent checkpoint message: %s", sent_message.id) + card_message = MessageActivityInput(text="Adaptive Card between streams.").add_card(create_simple_card()) + sent_message = await ctx.send(card_message) + logger.info("Sent checkpoint adaptive card: %s", sent_message.id) await asyncio.sleep(2) ctx.stream.update("Starting stream 2...") From 91a0afcee9bf7444ebf933d511e94665f93b66a8 Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Wed, 1 Jul 2026 18:13:28 -0700 Subject: [PATCH 05/10] Emit card in multi-stream final message Update the multi-stream example to emit the Adaptive Card through ctx.stream before close(reset=True), so the card is part of the first stream's final message. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/README.md | 2 +- examples/stream/src/main.py | 13 +++++++------ 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/examples/stream/README.md b/examples/stream/README.md index bb461f1e9..05b84c389 100644 --- a/examples/stream/README.md +++ b/examples/stream/README.md @@ -4,7 +4,7 @@ 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 finalizing the current stream with `close(reset=True)`, sending an Adaptive Card, and then reusing `ctx.stream` for another streamed response. +- Send `multi-stream` to test emitting an Adaptive Card as part of the first stream final message, finalizing with `close(reset=True)`, and then reusing `ctx.stream` for another streamed response. ## Running diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index fdf739384..b59bf20ac 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -97,12 +97,13 @@ async def handle_message(ctx: ActivityContext[MessageActivity]): await asyncio.sleep(0.5) ctx.stream.emit(message) - await ctx.stream.close(reset=True) - await asyncio.sleep(1) - - card_message = MessageActivityInput(text="Adaptive Card between streams.").add_card(create_simple_card()) - sent_message = await ctx.send(card_message) - logger.info("Sent checkpoint adaptive card: %s", sent_message.id) + 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(reset=True) + 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...") From 19c15d7d5cbe92952f1e8398c651c238e779293c Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Thu, 2 Jul 2026 14:18:52 -0700 Subject: [PATCH 06/10] Reopen stream on emit after close Make close finalize the current streamed message while a later emit or update starts a new stream cycle on the same instance. Keep repeated close calls idempotent until another emit occurs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/README.md | 2 +- examples/stream/src/main.py | 6 ++--- .../src/microsoft_teams/apps/http_stream.py | 27 ++++++++----------- .../microsoft_teams/apps/plugins/streamer.py | 13 +++------ packages/apps/tests/test_http_stream.py | 14 ++++++---- 5 files changed, 28 insertions(+), 34 deletions(-) diff --git a/examples/stream/README.md b/examples/stream/README.md index 05b84c389..99d0d7192 100644 --- a/examples/stream/README.md +++ b/examples/stream/README.md @@ -4,7 +4,7 @@ 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(reset=True)`, and then reusing `ctx.stream` for another streamed response. +- 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 diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index b59bf20ac..51a6476ba 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -35,11 +35,11 @@ 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 with reset=True.", + "[stream 1] Next the handler will close the current streamed message.", ] SECOND_STREAM_MESSAGES = [ - "[stream 2] Reusing ctx.stream after close(reset=True). ", + "[stream 2] Reusing ctx.stream after emit reopens the closed stream. ", "[stream 2] This should render after the non-stream checkpoint message. ", "[stream 2] The app processor will close this stream when the handler returns.", ] @@ -101,7 +101,7 @@ async def handle_message(ctx: ActivityContext[MessageActivity]): create_simple_card() ) ctx.stream.emit(card_message) - sent_message = await ctx.stream.close(reset=True) + 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) diff --git a/packages/apps/src/microsoft_teams/apps/http_stream.py b/packages/apps/src/microsoft_teams/apps/http_stream.py index b9d246cbb..c78e64ffb 100644 --- a/packages/apps/src/microsoft_teams/apps/http_stream.py +++ b/packages/apps/src/microsoft_teams/apps/http_stream.py @@ -83,7 +83,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 @@ -113,6 +113,10 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) if self._canceled: raise StreamCancelledError("Stream has been cancelled.") + if self._result is not None: + logger.debug("starting a new streamed message after close") + self._result = None + if isinstance(activity, str): activity = MessageActivityInput(text=activity, type="message") self._queue.append(activity) @@ -162,25 +166,17 @@ async def _poll(): except asyncio.TimeoutError: return False - async def close(self, reset: bool = False) -> Optional[SentActivity]: + async def close(self) -> Optional[SentActivity]: """ Finalize the current streamed message. - By default, closing is terminal and idempotent: subsequent calls return - the same final activity. When reset is True, the current streamed - message is finalized and the stream is reset so the same instance can - be used for another streamed message. - - Args: - reset: Whether to reset the stream after finalizing 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._result is not None: logger.debug("stream already closed with result") - result = self._result - if reset: - self._result = None - return result + return self._result if self._canceled: logger.debug("stream was cancelled, nothing to close") @@ -219,8 +215,7 @@ async def close(self, reset: bool = False) -> Optional[SentActivity]: # Reset per-message state; keep event handlers and cancellation state. self._reset_state() - if not reset: - self._result = res + self._result = res logger.debug("stream closed with result: %s", res) return res diff --git a/packages/apps/src/microsoft_teams/apps/plugins/streamer.py b/packages/apps/src/microsoft_teams/apps/plugins/streamer.py index 811dbc2bb..4e0bf1765 100644 --- a/packages/apps/src/microsoft_teams/apps/plugins/streamer.py +++ b/packages/apps/src/microsoft_teams/apps/plugins/streamer.py @@ -87,17 +87,12 @@ def clear_text(self) -> None: """ ... - async def close(self, reset: bool = False) -> Optional[SentActivity]: + async def close(self) -> Optional[SentActivity]: """ Finalize the current streamed message. - By default, closing is terminal and idempotent: subsequent calls return - the same final activity. When reset is True, the current streamed - message is finalized and the stream is reset so the same instance can - be used for another streamed message. - - Args: - reset: Whether to reset the stream after finalizing 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 411d9c0ac..802109608 100644 --- a/packages/apps/tests/test_http_stream.py +++ b/packages/apps/tests/test_http_stream.py @@ -455,7 +455,7 @@ async def test_close_waits_for_flush_to_complete(self, mock_api_client, conversa assert mock_api_client.sent_activities[0].text == "Response text" @pytest.mark.asyncio - async def test_close_reset_reuses_stream_for_next_message( + 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() @@ -475,17 +475,21 @@ async def handle_close(activity: SentActivity) -> None: await asyncio.sleep(0) await self._run_scheduled_flushes(scheduled) - first_result = await stream.close(reset=True) + first_result = await stream.close() await close_event.wait() close_event.clear() assert first_result is not None - assert stream.closed is False - assert stream.sequence == 1 + 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() + second_result = await stream.close() await close_event.wait() assert second_result is not None From 316f1df0cd0c93359be758d0bdd0c322673d8f22 Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Thu, 2 Jul 2026 16:06:14 -0700 Subject: [PATCH 07/10] Clarify stream reset helper names Rename the per-message reset helper and add a separate helper for preparing the stream instance for the next stream cycle. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../apps/src/microsoft_teams/apps/http_stream.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/http_stream.py b/packages/apps/src/microsoft_teams/apps/http_stream.py index c78e64ffb..630010db9 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: """ @@ -115,7 +120,7 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) if self._result is not None: logger.debug("starting a new streamed message after close") - self._result = None + self._reset_stream_for_next_stream() if isinstance(activity, str): activity = MessageActivityInput(text=activity, type="message") @@ -214,7 +219,7 @@ async def close(self) -> Optional[SentActivity]: self._events.emit("close", res) # Reset per-message state; keep event handlers and cancellation state. - self._reset_state() + self._reset_current_stream() self._result = res logger.debug("stream closed with result: %s", res) From 809279f87dc6a9a5f21065eb8533cefbf24a8598 Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Thu, 2 Jul 2026 16:08:58 -0700 Subject: [PATCH 08/10] Use closed property in stream lifecycle checks Prefer the HttpStream.closed property where the code is checking whether the current stream cycle has been finalized. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- packages/apps/src/microsoft_teams/apps/http_stream.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/apps/src/microsoft_teams/apps/http_stream.py b/packages/apps/src/microsoft_teams/apps/http_stream.py index 630010db9..a70cf2fac 100644 --- a/packages/apps/src/microsoft_teams/apps/http_stream.py +++ b/packages/apps/src/microsoft_teams/apps/http_stream.py @@ -118,7 +118,7 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) if self._canceled: raise StreamCancelledError("Stream has been cancelled.") - if self._result is not None: + if self.closed: logger.debug("starting a new streamed message after close") self._reset_stream_for_next_stream() @@ -179,8 +179,9 @@ async def close(self) -> Optional[SentActivity]: updating after close starts a new streamed message using the same stream instance. """ - if self._result is not None: + if self.closed: logger.debug("stream already closed with result") + assert self._result is not None return self._result if self._canceled: From 73bf5ed84080e05c273250e855fc231a95dc00b7 Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Thu, 2 Jul 2026 16:11:59 -0700 Subject: [PATCH 09/10] Clarify stream cycle documentation Align protocol documentation and example text with the emit-after-close stream cycle behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/src/main.py | 2 +- packages/apps/src/microsoft_teams/apps/plugins/streamer.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index 51a6476ba..414a30057 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -40,7 +40,7 @@ SECOND_STREAM_MESSAGES = [ "[stream 2] Reusing ctx.stream after emit reopens the closed stream. ", - "[stream 2] This should render after the non-stream checkpoint message. ", + "[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.", ] diff --git a/packages/apps/src/microsoft_teams/apps/plugins/streamer.py b/packages/apps/src/microsoft_teams/apps/plugins/streamer.py index 4e0bf1765..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 From 78f98a7e1949602f92222c9510a67f7785e0dcdc Mon Sep 17 00:00:00 2001 From: heyitsaamir Date: Thu, 2 Jul 2026 16:13:59 -0700 Subject: [PATCH 10/10] Use card builder in stream example Replace AdaptiveCard.model_validate in the stream example with the builder-style AdaptiveCard and TextBlock classes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- examples/stream/src/main.py | 27 +++++++-------------------- 1 file changed, 7 insertions(+), 20 deletions(-) diff --git a/examples/stream/src/main.py b/examples/stream/src/main.py index 414a30057..16e305b4b 100644 --- a/examples/stream/src/main.py +++ b/examples/stream/src/main.py @@ -9,7 +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 +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). @@ -56,25 +56,12 @@ def should_send_simple_card(text: str | None) -> bool: def create_simple_card() -> AdaptiveCard: - return AdaptiveCard.model_validate( - { - "type": "AdaptiveCard", - "version": "1.4", - "body": [ - { - "type": "TextBlock", - "text": "Simple Adaptive Card", - "weight": "Bolder", - "size": "Large", - "wrap": True, - }, - { - "type": "TextBlock", - "text": "If you can see this card, basic Adaptive Card delivery is working.", - "wrap": True, - }, - ], - } + 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), + ], )