Utility functions to build and sort realtime feedback events for a pipeline.
# Source: https://github.com/dograh-hq/dograh | License: BSD-2-Clause | Copyright 2025, Zansat Technologies Private Limited
# Provided via justclaw.it.com; keep this notice with the code.
"""Shared helpers for building and ordering realtime feedback events."""
from typing import Any
from pipecat.utils.enums import RealtimeFeedbackType
def build_node_transition_event(
*,
node_id: str | None,
node_name: str | None,
previous_node_id: str | None,
previous_node_name: str | None,
allow_interrupt: bool = False,
) -> dict[str, Any]:
return {
"type": RealtimeFeedbackType.NODE_TRANSITION.value,
"payload": {
"node_id": node_id,
"node_name": node_name,
"previous_node_id": previous_node_id,
"previous_node_na
... (truncated -- full source via MCP)
See the full source, get the GitHub permalink, and search for more like it.
Get a free API key