refactor: Migrate from programmatic to streaming subscriptions by SoulSniper-V2 · Pull Request #349 · dapr/dapr-agents

SoulSniper-V2 · 2026-01-07T15:06:04Z

Description

Migrates message subscriptions from the legacy subscribe_with_handler (programmatic) to dapr_client.subscribe (streaming).

This fixes #348 by using persistent gRPC connections, which removes the requirement for exposing app ports and improves reliability.

Issue reference

Closes: #348

Checklist

Created/updated tests
Tested this change against all the quickstarts
Extended the documentation
- Created the dapr/docs PR: N/A

Implementation

Replaced subscribe_with_handler with dapr_client.subscribe and a background consumer thread.
Added graceful shutdown logic to join threads on exit.
Added error logging for failed retries.

Verification

Unit Tests: Updated tests/workflow/test_message_router.py to mock streaming subscriptions. All 77 unit tests passed.
Static Analysis: ruff passed.

sicoyle

Thank you so much for your interest in contributing and for adding tests—this is great 🙌

I have a few comments so far. Also, would you mind splitting the streaming logic out of registration.py into separate functions, or possibly a separate file (e.g., subscription.py)? With these great additions, that file is starting to get a bit heavy, and this would help keep things more maintainable.

Did you test these changes against some of the quickstarts/examples by chance?

dapr_agents/workflow/utils/registration.py

SoulSniper-V2 · 2026-01-08T22:38:56Z

Thanks for the review @sicoyle! 🙌

I moved the streaming logic into subscription.py like you suggested - definitely makes registration.py much cleaner. Also went ahead and removed the legacy code path and addressed the variable naming/logging nits.

Ran the message router tests locally and everything looks good.

SoulSniper-V2 · 2026-01-24T16:35:11Z

Hi @sicoyle, just checking in on this. I have addressed the feedback regarding the refactor into subscription.py and the cleanup of the legacy paths. Let me know if there’s anything else needed to move this toward a final review or to start the CI builds

sicoyle · 2026-01-27T17:54:24Z

Hi @sicoyle, just checking in on this. I have addressed the feedback regarding the refactor into subscription.py and the cleanup of the legacy paths. Let me know if there’s anything else needed to move this toward a final review or to start the CI builds

Thank you! Could you please correct the lint/build failures? I just need to test these changes locally before merge (since I don't have automation yet running on PRs 🙃 )

sicoyle

Would you mind splitting up subscribe_message_bindings a bit? At almost 400 lines with several nested functions, it’s getting a little hard to follow and review. Breaking it into smaller pieces would make it much easier to maintain and reason about 🙏

dapr_agents/workflow/utils/registration.py

dapr_agents/workflow/utils/subscription.py

SoulSniper-V2 · 2026-01-28T02:24:16Z

Addressed all feedback. Split the function into smaller helpers, added constants, fixed exception handling, and renamed variables for clarity. Tests pass.

Signed-off-by: Arush Wadhawan <soulsniper@Arushs-MacBook-Air.local>

Signed-off-by: Arush Wadhawan <warush23+github@gmail.com>

- Split subscribe_message_bindings into focused helper functions - Add module-level constants for delivery modes and status codes - Use WorkflowStatus enum instead of magic strings - Add proper validation for dead_letter_topics (ValueError if multiple) - Fix _resolve_event_loop to properly raise on missing loop - Add zombie thread detection (raise RuntimeError on timeout) - Move json import to top level - Rename variables for clarity: - grouped -> bindings_by_topic_key - b -> binding - plan -> binding_schema_pairs - preferred -> matching_ce_type_pairs - Replace logger.info with logger.debug for internal operations - Restore underscore prefixes for internal functions - Remove nested try/except where possible - Add proper docstrings explaining function purposes Signed-off-by: Arush Wadhawan <warush23+github@gmail.com>

CasperGN

Thank you for this!

Left an ask to use f-strings consistently rather than %s.

CasperGN · 2026-01-28T18:16:05Z

dapr_agents/workflow/utils/subscription.py

+        else:
+            setattr(parsed, METADATA_KEY, metadata)
+    except Exception:
+        logger.debug("Could not attach %s to payload; continuing.", METADATA_KEY)


Can you convert all these strings (including the error message ones further down below) to f strings so we're consistent?

There's a mix in this file as well.

Signed-off-by: Arush Wadhawan <warush23+github@gmail.com>

SoulSniper-V2 · 2026-01-28T20:14:19Z

Converted all logging to f-strings.

CasperGN · 2026-01-29T09:45:44Z

@sicoyle this lgtm now.

Thank you @SoulSniper-V2!

github-actions · 2026-02-26T23:27:07Z

Metadata Schema Compatibility Report

Breaking Metadata Schema Changes

Removed property agent_metadata from (root)
Removed property max_iterations from (root)
Removed property schema_version from (root)
Removed property tool_choice from (root)
Removed property statestore from AgentMetadata
Removed property component_name from LLMMetadata
Removed property statestore from MemoryMetadata
Removed property type from MemoryMetadata
Removed property name from PubSubMetadata
Removed property statestore from RegistryMetadata
Removed property tool_args from ToolMetadata
Removed property tool_description from ToolMetadata
Removed property tool_name from ToolMetadata
New required field version in (root) (did not exist in previous version)
New required field type in MemoryStoreMetadata (did not exist in previous version)
New required field resource_name in PubSubMetadata (did not exist in previous version)
New required field args in ToolMetadata (did not exist in previous version)
New required field description in ToolMetadata (did not exist in previous version)
New required field name in ToolMetadata (did not exist in previous version)

This is informational — breaking changes may be intentional for a new release.

Copilot

Pull request overview

Migrates workflow message routing from legacy programmatic pub/sub subscriptions (subscribe_with_handler) to streaming subscriptions (dapr_client.subscribe) to use persistent gRPC connections and improve reliability.

Changes:

Added a new workflow/utils/subscription.py module implementing streaming subscription consumption in background threads with optional async delivery mode.
Refactored workflow/utils/registration.py to build MessageRouteBindings and delegate subscription setup to the new streaming utilities.
Updated tests/workflow/test_message_router.py to mock dapr_client.subscribe instead of subscribe_with_handler.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 12 comments.

File	Description
`tests/workflow/test_message_router.py`	Updates unit tests to mock streaming subscriptions and validate subscription call parameters.
`dapr_agents/workflow/utils/subscription.py`	New streaming subscription implementation: composite routing per topic, consumer threads, async worker queue option, shutdown helpers.
`dapr_agents/workflow/utils/registration.py`	Refactors binding collection and replaces legacy subscription logic with calls into the new streaming subscription module.

Comments suppressed due to low confidence (1)

dapr_agents/workflow/utils/registration.py:316

register_message_routes no longer accepts a subscribe override, but there are still internal call sites passing subscribe=... (e.g. dapr_agents/workflow/runners/base.py calls it with subscribe=subscribe). This will raise TypeError: got an unexpected keyword argument 'subscribe' at runtime. Either update all call sites to stop passing subscribe, or keep the parameter (possibly deprecated) and handle/ignore it for streaming subscriptions.

def register_message_routes(
    *,
    dapr_client: DaprClient,
    targets: Optional[Iterable[Any]] = None,
    routes: Optional[Iterable[PubSubRouteSpec]] = None,
    loop: Optional[asyncio.AbstractEventLoop] = None,
    delivery_mode: Literal["sync", "async"] = "sync",
    queue_maxsize: int = 1024,
    deduper: Optional[DedupeBackend] = None,
    scheduler: Optional[SchedulerFn] = None,
    wf_client: Optional[wf.DaprWorkflowClient] = None,
    await_result: bool = False,
    await_timeout: Optional[int] = None,
    fetch_payloads: bool = True,
    log_outcome: bool = True,
) -> List[Callable[[], None]]:

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-02-26T23:31:50Z

dapr_agents/workflow/utils/registration.py

+    _ = _resolve_event_loop(
        loop
    )  # Parity with message registrar; FastAPI does not require it yet.


_mount_http_bindings calls _resolve_event_loop(loop) purely for “parity”, but this can raise in normal FastAPI setups where routes are registered from a synchronous context (common with Python 3.12+ where asyncio.get_event_loop() can fail). Since HTTP route mounting doesn’t require an event loop here, consider removing this call or making it non-fatal (e.g., only validate in truly async-required paths).

Suggested change

_ = _resolve_event_loop(

loop

) # Parity with message registrar; FastAPI does not require it yet.

Copilot · 2026-02-26T23:31:50Z

dapr_agents/workflow/utils/subscription.py

+                                loop.call_soon_threadsafe(
+                                    queue.put_nowait,
+                                    (binding.handler, parsed),
+                                )
+                                return TopicEventResponse(STATUS_SUCCESS)


In async delivery mode the handler schedules queue.put_nowait(...) via loop.call_soon_threadsafe(...) and immediately returns success. If the queue is full, put_nowait will raise QueueFull in the event loop, and the message will still be ACKed (lost) with no retry. Add explicit backpressure handling: catch QueueFull and return retry (or block with an awaited put), and ensure failures to enqueue do not ACK the message.

Suggested change

loop.call_soon_threadsafe(

queue.put_nowait,

(binding.handler, parsed),

)

return TopicEventResponse(STATUS_SUCCESS)

if loop.is_running():

# Backpressure-aware enqueue: block until the item is queued

fut = asyncio.run_coroutine_threadsafe(

queue.put((binding.handler, parsed)),

loop,

)

try:

fut.result()

except Exception:

logger.exception(

"Failed to enqueue workflow task for handler %s; "

"requesting retry.",

binding.name,

)

return TopicEventResponse(STATUS_RETRY)

return TopicEventResponse(STATUS_SUCCESS)

# If the loop is not running, fall through to the sync path below.

Copilot · 2026-02-26T23:31:50Z

dapr_agents/workflow/utils/subscription.py

+    _validate_delivery_mode(delivery_mode)
+    _validate_dead_letter_topics(bindings)
+
+    resolved_loop = _resolve_event_loop(loop)


subscribe_message_bindings resolves an event loop unconditionally (resolved_loop = _resolve_event_loop(loop)), even for delivery_mode='sync'. In Python 3.12+ this can raise when called from a synchronous context with loop=None, despite sync mode being able to operate via asyncio.run(...). Consider only requiring an existing/running loop for delivery_mode='async', and otherwise creating a new loop or bypassing loop resolution in sync mode.

Suggested change

resolved_loop = _resolve_event_loop(loop)

if delivery_mode == DELIVERY_MODE_ASYNC:

resolved_loop = _resolve_event_loop(loop)

else:

# In sync mode we can rely on asyncio.run(...) and do not require

# an existing/running event loop; avoid resolving it unconditionally.

resolved_loop = loop

Copilot · 2026-02-26T23:31:51Z

tests/workflow/test_message_router.py

+    mock_sub = MagicMock()
+    mock_sub.__iter__.return_value = []
+    mock_client.subscribe.return_value = mock_sub


mock_sub.__iter__.return_value is set to [], which is iterable but not an iterator; iterating the subscription in the consumer thread may raise a TypeError and the test won’t reliably exercise the streaming loop. Prefer mock_sub.__iter__.return_value = iter([]).

Copilot · 2026-02-26T23:31:51Z

tests/workflow/test_message_router.py

+    mock_sub = MagicMock()
+    mock_sub.__iter__.return_value = []
+    mock_client.subscribe.return_value = mock_sub


mock_sub.__iter__.return_value = [] returns a list rather than an iterator; the consumer thread’s for msg in sub: can fail with TypeError and exit. Set this to iter([]) to accurately model an empty subscription stream.

i addressed

dapr_agents/workflow/utils/subscription.py

tests/workflow/test_message_router.py

sicoyle · 2026-03-05T14:50:35Z

@SoulSniper-V2 this is super close! Would you mind addressing the copilot feedback please? 🙏

github-actions · 2026-03-05T14:51:00Z

Metadata Schema Compatibility Report

Breaking Metadata Schema Changes

Removed property agent_metadata from (root)
Removed property max_iterations from (root)
Removed property schema_version from (root)
Removed property tool_choice from (root)
Removed property statestore from AgentMetadata
Removed property component_name from LLMMetadata
Removed property statestore from MemoryMetadata
Removed property type from MemoryMetadata
Removed property name from PubSubMetadata
Removed property statestore from RegistryMetadata
Removed property tool_args from ToolMetadata
Removed property tool_description from ToolMetadata
Removed property tool_name from ToolMetadata
New required field version in (root) (did not exist in previous version)
New required field type in MemoryStoreMetadata (did not exist in previous version)
New required field resource_name in PubSubMetadata (did not exist in previous version)
New required field args in ToolMetadata (did not exist in previous version)
New required field description in ToolMetadata (did not exist in previous version)
New required field name in ToolMetadata (did not exist in previous version)

This is informational — breaking changes may be intentional for a new release.

1Ninad · 2026-03-09T15:05:48Z

Hi @SoulSniper-V2 @sicoyle, just checking - are you still planning to address the Copilot feedback on this PR?
If you are unable to continue, I would be happy to help work on the requested changes and push an update. Please let me know.

sicoyle · 2026-03-10T16:23:56Z

thank you so much for your contributions @SoulSniper-V2 🤗 I went ahead and cherry picked your changes in this PR and addressed all of the remaining PR feedback here so we can get this change into the next release 🎉
#500

SoulSniper-V2 requested review from a team as code owners January 7, 2026 15:06

SoulSniper-V2 force-pushed the fix/migrate-programmatic-subscriptions branch from 37c2439 to 46666cb Compare January 7, 2026 15:10

sicoyle reviewed Jan 8, 2026

View reviewed changes

sicoyle requested changes Jan 27, 2026

View reviewed changes

sicoyle changed the title ~~refactor: Migrate from programmatic to streaming pub/sub subscriptions~~ refactor: Migrate from programmatic to streamin subscriptions Jan 27, 2026

sicoyle changed the title ~~refactor: Migrate from programmatic to streamin subscriptions~~ refactor: Migrate from programmatic to streaming subscriptions Jan 27, 2026

Arush Wadhawan and others added 5 commits January 27, 2026 21:24

refactor: migrate from programmatic to streaming pub/sub subscriptions

d1bad08

Signed-off-by: Arush Wadhawan <soulsniper@Arushs-MacBook-Air.local>

style: apply ruff formatting fixes

7c9efe9

Signed-off-by: Arush Wadhawan <soulsniper@Arushs-MacBook-Air.local>

refactor: extract streaming subscription logic to subscription.py

02326aa

Signed-off-by: Arush Wadhawan <soulsniper@Arushs-MacBook-Air.local>

style: apply ruff formatting

51999e5

Signed-off-by: Arush Wadhawan <warush23+github@gmail.com>

SoulSniper-V2 force-pushed the fix/migrate-programmatic-subscriptions branch from 478053b to 9143489 Compare January 28, 2026 02:26

CasperGN requested changes Jan 28, 2026

View reviewed changes

style: convert logging to f-strings for consistency

1f3e21a

Signed-off-by: Arush Wadhawan <warush23+github@gmail.com>

sicoyle added 2 commits February 4, 2026 09:44

Merge branch 'main' into fix/migrate-programmatic-subscriptions

253d567

Merge branch 'main' into fix/migrate-programmatic-subscriptions

b0c7737

Copilot AI review requested due to automatic review settings February 26, 2026 23:26

Copilot started reviewing on behalf of sicoyle February 26, 2026 23:27 View session

Copilot AI reviewed Feb 26, 2026

View reviewed changes

Merge branch 'main' into fix/migrate-programmatic-subscriptions

46f07c0

CasperGN and others added 3 commits March 9, 2026 17:22

Merge branch 'main' into fix/migrate-programmatic-subscriptions

3092dab

Merge branch 'main' into fix/migrate-programmatic-subscriptions

a7c3847

Merge branch 'main' into fix/migrate-programmatic-subscriptions

bd24edb

sicoyle mentioned this pull request Mar 10, 2026

feat: use streaming subscriptions #500

Merged

4 tasks

sicoyle closed this Mar 10, 2026

	_ = _resolve_event_loop(
	loop
	) # Parity with message registrar; FastAPI does not require it yet.

-                                loop.call_soon_threadsafe(
-                                    queue.put_nowait,
-                                    (binding.handler, parsed),
-                                )
-                                return TopicEventResponse(STATUS_SUCCESS)
+                                if loop.is_running():
+                                    # Backpressure-aware enqueue: block until the item is queued
+                                    fut = asyncio.run_coroutine_threadsafe(
+                                        queue.put((binding.handler, parsed)),
+                                        loop,
+                                    )
+                                    try:
+                                        fut.result()
+                                    except Exception:
+                                        logger.exception(
+                                            "Failed to enqueue workflow task for handler %s; "
+                                            "requesting retry.",
+                                            binding.name,
+                                        )
+                                        return TopicEventResponse(STATUS_RETRY)
+                                    return TopicEventResponse(STATUS_SUCCESS)
+                                # If the loop is not running, fall through to the sync path below.

-    resolved_loop = _resolve_event_loop(loop)
+    if delivery_mode == DELIVERY_MODE_ASYNC:
+        resolved_loop = _resolve_event_loop(loop)
+    else:
+        # In sync mode we can rely on asyncio.run(...) and do not require
+        # an existing/running event loop; avoid resolving it unconditionally.
+        resolved_loop = loop

Conversation

SoulSniper-V2 commented Jan 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Issue reference

Checklist

Implementation

Verification

Uh oh!

sicoyle left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SoulSniper-V2 commented Jan 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SoulSniper-V2 commented Jan 24, 2026

Uh oh!

sicoyle commented Jan 27, 2026

Uh oh!

sicoyle left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SoulSniper-V2 commented Jan 28, 2026

Uh oh!

CasperGN left a comment

Choose a reason for hiding this comment

Uh oh!

CasperGN Jan 28, 2026

Choose a reason for hiding this comment

Uh oh!

SoulSniper-V2 commented Jan 28, 2026

Uh oh!

CasperGN commented Jan 29, 2026

Uh oh!

github-actions bot commented Feb 26, 2026

Metadata Schema Compatibility Report

Breaking Metadata Schema Changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Feb 26, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 26, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 26, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 26, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Feb 26, 2026

Choose a reason for hiding this comment

Uh oh!

sicoyle Mar 10, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SoulSniper-V2 commented Jan 7, 2026 •

edited

Loading

SoulSniper-V2 commented Jan 8, 2026 •

edited

Loading