Skip to main content

​
MessageSerializer

Abstract base class for bus message serialization. Network bus implementations use a MessageSerializer to convert messages to bytes for transmission and reconstruct them on the receiving end. 
from pipecat_subagents.bus.serializers.base import MessageSerializer

​
Methods

​
serialize

@abstractmethod
def serialize(self, message: BusMessage) -> bytes
Convert a bus message to bytes.
ParameterTypeDescription
messageBusMessageThe bus message to serialize
Returns: The serialized bytes.

​
deserialize

@abstractmethod
def deserialize(self, data: bytes) -> Optional[BusMessage]
Reconstruct a bus message from bytes.
ParameterTypeDescription
databytesThe serialized bytes produced by serialize()
Returns: The reconstructed BusMessage, or None if deserialization fails.

​
JSONMessageSerializer

Serialize bus messages as JSON with pluggable type adapters. Handles JSON-native types, enums, bytes, dataclasses, and any type with a registered TypeAdapter. Adapters for common Pipecat types (LLMContext, ToolsSchema) are registered by default. 
from pipecat_subagents.bus.serializers import JSONMessageSerializer

serializer = JSONMessageSerializer()

data = serializer.serialize(message)
restored = serializer.deserialize(data)
This is the default serializer used by RedisBus and the proxy agents.

​
Methods

​
register_adapter

def register_adapter(self, type_: type, adapter: TypeAdapter) -> None
Register a custom type adapter for types not supported by default.
ParameterTypeDescription
type_typeThe type to handle
adapterTypeAdapterThe adapter that serializes/deserializes instances

​
TypeAdapter

Abstract base class for type-specific serialization adapters. Each adapter handles one or more types, converting them to/from a JSON-compatible dict. 
from pipecat_subagents.bus.adapters.base import TypeAdapter

​
Methods

​
serialize

@abstractmethod
def serialize(self, obj: Any, serialize_value: SerializeFunc) -> dict[str, Any]
Convert an object to a JSON-compatible dict.
ParameterTypeDescription
objAnyThe object to serialize
serialize_valueCallable[[Any], Any]Callback to recursively serialize nested values

​
deserialize

@abstractmethod
def deserialize(
    self,
    data: dict[str, Any],
    deserialize_value: DeserializeFunc,
    target_type: Optional[type] = None,
) -> Any
Reconstruct an object from a dict.
ParameterTypeDefaultDescription
datadict[str, Any]The dict representation produced by serialize()
deserialize_valueCallable[[Any], Any]Callback to recursively deserialize nested values
target_typeOptional[type]NoneThe resolved target class