Added new message router workflow quickstart

Signed-off-by: Roberto Rodriguez <[email protected]>

Author: Cyb3rWard0g

quickstarts/04-message-router-workflow/README.md

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import signal
+
+import dapr.ext.workflow as wf
+from dapr.clients import DaprClient
+from dotenv import load_dotenv
+from handlers import start_blog_workflow
+from workflow import (
+    blog_workflow,
+    create_outline,
+    write_post,
+)
+
+from dapr_agents.workflow.utils.registration import register_message_handlers
+
+load_dotenv()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+async def _wait_for_shutdown() -> None:
+    """Block until Ctrl+C or SIGTERM."""
+    loop = asyncio.get_running_loop()
+    stop = asyncio.Event()
+
+    def _set_stop(*_: object) -> None:
+        stop.set()
+
+    try:
+        loop.add_signal_handler(signal.SIGINT, _set_stop)
+        loop.add_signal_handler(signal.SIGTERM, _set_stop)
+    except NotImplementedError:
+        # Windows fallback
+        signal.signal(signal.SIGINT, lambda *_: _set_stop())
+        signal.signal(signal.SIGTERM, lambda *_: _set_stop())
+
+    await stop.wait()
+
+
+async def main() -> None:
+    runtime = wf.WorkflowRuntime()
+    
+    runtime.register_workflow(blog_workflow)
+    runtime.register_activity(create_outline)
+    runtime.register_activity(write_post)
+
+    runtime.start()
+
+    try:
+        with DaprClient() as client:
+            # Wire streaming subscriptions for our router(s)
+            closers = register_message_handlers(
+                targets=[start_blog_workflow],
+                dapr_client=client,
+            )
+
+            try:
+                await _wait_for_shutdown()
+            finally:
+                for close in closers:
+                    try:
+                        close()
+                    except Exception:
+                        logger.exception("Error while closing subscription")
+    finally:
+        runtime.shutdown()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass

quickstarts/04-message-router-workflow/app.py

@@ -0,0 +1,14 @@
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: openai
+spec:
+  type: conversation.openai
+  version: v1
+  metadata:
+  - name: key
+    value: "{{OPENAI_API_KEY}}"
+  - name: model
+    value: gpt-5-mini
+  - name: temperature
+    value: 1

quickstarts/04-message-router-workflow/components/openai.yaml

@@ -0,0 +1,10 @@
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: messagepubsub
+spec:
+  type: pubsub.redis
+  version: v1
+  metadata:
+    - name: redisHost
+      value: localhost:6379

quickstarts/04-message-router-workflow/components/pubsub.yaml

@@ -0,0 +1,12 @@
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: workflowstatestore
+spec:
+  type: state.redis
+  version: v1
+  metadata:
+    - name: redisHost
+      value: localhost:6379
+    - name: actorStateStore
+      value: "true"

quickstarts/04-message-router-workflow/components/statestore.yaml

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import logging
+
+import dapr.ext.workflow as wf
+from dapr.clients.grpc._response import TopicEventResponse
+from pydantic import BaseModel, Field
+
+from dapr_agents.workflow.decorators.routers import message_router
+
+logger = logging.getLogger(__name__)
+
+
+class StartBlogMessage(BaseModel):
+    topic: str = Field(min_length=1, description="Blog topic/title")
+
+
+# Import the workflow after defining models to avoid circular import surprises
+from workflow import blog_workflow  # noqa: E402
+
+
+@message_router(pubsub="messagepubsub", topic="blog.requests")
+def start_blog_workflow(message: StartBlogMessage) -> TopicEventResponse:
+    """
+    Triggered by pub/sub. Validates payload via Pydantic and schedules the workflow.
+    """
+    try:
+        client = wf.DaprWorkflowClient()
+        instance_id = client.schedule_new_workflow(
+            workflow=blog_workflow,
+            input=message.model_dump(),
+        )
+        logger.info("Scheduled blog_workflow instance=%s topic=%s", instance_id, message.topic)
+        return TopicEventResponse("success")
+    except Exception as exc:  # transient infra error → retry
+        logger.exception("Failed to schedule blog workflow: %s", exc)
+        return TopicEventResponse("retry")

quickstarts/04-message-router-workflow/handlers.py

@@ -0,0 +1,155 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import random
+import signal
+import sys
+from typing import Any, Dict
+
+from dapr.clients import DaprClient
+
+# ---------------------------
+# Config via environment vars
+# ---------------------------
+PUBSUB_NAME = os.getenv("PUBSUB_NAME", "messagepubsub")
+TOPIC_NAME = os.getenv("TOPIC_NAME", "blog.requests")
+BLOG_TOPIC = os.getenv("BLOG_TOPIC", "AI Agents")  # used when RAW_DATA is not provided
+RAW_DATA = os.getenv("RAW_DATA")  # if set, must be a JSON object (string)
+CONTENT_TYPE = os.getenv("CONTENT_TYPE", "application/json")
+CE_TYPE = os.getenv("CLOUDEVENT_TYPE")  # optional CloudEvent 'type' metadata
+
+# Publish behavior
+PUBLISH_ONCE = os.getenv("PUBLISH_ONCE", "true").lower() in {"1", "true", "yes"}
+INTERVAL_SEC = float(os.getenv("INTERVAL_SEC", "0"))  # used when PUBLISH_ONCE=false
+MAX_ATTEMPTS = int(os.getenv("MAX_ATTEMPTS", "8"))
+INITIAL_DELAY = float(os.getenv("INITIAL_DELAY", "0.5"))
+BACKOFF_FACTOR = float(os.getenv("BACKOFF_FACTOR", "2.0"))
+JITTER_FRAC = float(os.getenv("JITTER_FRAC", "0.2"))
+
+# Optional warmup (give sidecar/broker a moment)
+STARTUP_DELAY = float(os.getenv("STARTUP_DELAY", "1.0"))
+
+logger = logging.getLogger("publisher")
+
+
+async def _backoff_sleep(delay: float, jitter: float, factor: float) -> float:
+    """Sleep for ~delay seconds with ±jitter% randomness, then return the next delay."""
+    actual = max(0.0, delay * (1 + random.uniform(-jitter, jitter)))
+    if actual:
+        await asyncio.sleep(actual)
+    return delay * factor
+
+
+def _build_payload() -> Dict[str, Any]:
+    """
+    Build the JSON payload:
+      - if RAW_DATA is set → parse as JSON (must be an object)
+      - else               → {"topic": BLOG_TOPIC}
+    """
+    if RAW_DATA:
+        try:
+            data = json.loads(RAW_DATA)
+        except Exception as exc:  # noqa: BLE001
+            raise ValueError(f"Invalid RAW_DATA JSON: {exc}") from exc
+        if not isinstance(data, dict):
+            raise ValueError("RAW_DATA must be a JSON object")
+        return data
+
+    return {"topic": BLOG_TOPIC}
+
+
+def _encode_payload(payload: Dict[str, Any]) -> bytes:
+    """Encode the payload as UTF-8 JSON bytes."""
+    return json.dumps(payload, ensure_ascii=False).encode("utf-8")
+
+
+async def publish_once(client: DaprClient, payload: Dict[str, Any]) -> None:
+    """Publish once with retries and exponential backoff."""
+    delay = INITIAL_DELAY
+    body = _encode_payload(payload)
+
+    for attempt in range(1, MAX_ATTEMPTS + 1):
+        try:
+            logger.info("publish attempt %d → %s/%s", attempt, PUBSUB_NAME, TOPIC_NAME)
+            client.publish_event(
+                pubsub_name=PUBSUB_NAME,
+                topic_name=TOPIC_NAME,
+                data=body,
+                data_content_type=CONTENT_TYPE,
+                publish_metadata=({"cloudevent.type": CE_TYPE} if CE_TYPE else None),
+            )
+            logger.info("published successfully")
+            return
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("publish failed: %s", exc)
+            if attempt == MAX_ATTEMPTS:
+                raise
+            logger.info("retrying in ~%.2fs …", delay)
+            delay = await _backoff_sleep(delay, JITTER_FRAC, BACKOFF_FACTOR)
+
+
+async def main() -> int:
+    logging.basicConfig(level=logging.INFO)
+    stop_event = asyncio.Event()
+
+    # Signal-aware shutdown
+    loop = asyncio.get_running_loop()
+
+    def _stop(*_: object) -> None:
+        stop_event.set()
+
+    try:
+        loop.add_signal_handler(signal.SIGINT, _stop)
+        loop.add_signal_handler(signal.SIGTERM, _stop)
+    except NotImplementedError:
+        signal.signal(signal.SIGINT, lambda *_: _stop())
+        signal.signal(signal.SIGTERM, lambda *_: _stop())
+
+    # Optional warmup
+    if STARTUP_DELAY > 0:
+        await asyncio.sleep(STARTUP_DELAY)
+
+    payload = _build_payload()
+    logger.info("payload: %s", payload)
+
+    try:
+        with DaprClient() as client:
+            if PUBLISH_ONCE:
+                await publish_once(client, payload)
+                # brief wait so logs flush nicely under dapr
+                await asyncio.sleep(0.2)
+                return 0
+
+            # periodic mode
+            if INTERVAL_SEC <= 0:
+                logger.error("INTERVAL_SEC must be > 0 when PUBLISH_ONCE=false")
+                return 2
+
+            logger.info("starting periodic publisher every %.2fs", INTERVAL_SEC)
+            while not stop_event.is_set():
+                try:
+                    await publish_once(client, payload)
+                except Exception as exc:  # noqa: BLE001
+                    logger.error("giving up after %d attempts: %s", MAX_ATTEMPTS, exc)
+
+                # wait for next tick or shutdown
+                try:
+                    await asyncio.wait_for(stop_event.wait(), timeout=INTERVAL_SEC)
+                except asyncio.TimeoutError:
+                    pass
+
+            logger.info("shutdown requested; exiting")
+            return 0
+
+    except KeyboardInterrupt:
+        return 130
+    except Exception as exc:  # noqa: BLE001
+        logger.exception("fatal error: %s", exc)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

quickstarts/04-message-router-workflow/message_client.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from dapr.ext.workflow import DaprWorkflowContext
+from dotenv import load_dotenv
+
+from dapr_agents.llm.dapr import DaprChatClient
+from dapr_agents.workflow.decorators import llm_activity
+
+load_dotenv()
+
+# Initialize the LLM client and workflow runtime
+llm = DaprChatClient(component_name="openai")
+
+
+def blog_workflow(ctx: DaprWorkflowContext, wf_input: dict) -> str:
+    """
+    Workflow input must be JSON-serializable. We accept a dict like:
+      {"topic": "<string>"}
+    """
+    topic = wf_input["topic"]
+    outline = yield ctx.call_activity(create_outline, input={"topic": topic})
+    post = yield ctx.call_activity(write_post, input={"outline": outline})
+    return post
+
+
+@llm_activity(
+    prompt="Create a short outline about {topic}. Output 3-5 bullet points.",
+    llm=llm,
+)
+async def create_outline(ctx, topic: str) -> str:
+    # Implemented by the decorator; body can be empty.
+    pass
+
+
+@llm_activity(
+    prompt="Write a short blog post following this outline:\n{outline}",
+    llm=llm,
+)
+async def write_post(ctx, outline: str) -> str:
+    # Implemented by the decorator; body can be empty.
+    pass

quickstarts/04-message-router-workflow/workflow.py

@@ -117,6 +117,19 @@ This quickstart demonstrates how to design and run **agent-based workflows**, st
 
 [Go to Agent-based Workflow Patterns](./04-agent-based-workflows/)
 
+### Message Router Workflow
+
+Learn how to trigger Dapr Workflows via Pub/Sub messages using the `@message_router` decorator.
+This pattern connects event-driven systems with LLM-powered workflows, validating and routing structured messages to durable workflow executions.
+
+- **Event-Driven Orchestration**: Start workflows automatically when messages arrive on a topic
+- **Edge Validation**: Enforce schema integrity with Pydantic before invoking workflows
+- **Seamless Integration**: Combine Dapr Pub/Sub, Workflow Runtime, and LLM activities for resilient automation
+
+This quickstart demonstrates how to design a message-driven workflow where each published event triggers a workflow instance such as creating a blog post outline and draft powered by an LLM.
+
+[Go to Message Router Workflow](./04-message-router-workflow/)
+
 ### Multi-Agent Workflows
 
 Advanced example of event-driven workflows with multiple autonomous agents: