Serializers

Message serialization for network transport

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.

Parameter	Type	Description
message	BusMessage	The bus message to serialize

Returns: The serialized bytes.

deserialize

@abstractmethod
def deserialize(self, data: bytes) -> BusMessage | None

Reconstruct a bus message from bytes.

Parameter	Type	Description
data	bytes	The 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.

Parameter	Type	Description
type_	type	The type to handle
adapter	TypeAdapter	The 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.

Parameter	Type	Description
obj	Any	The object to serialize
serialize_value	Callable[[Any], Any]	Callback to recursively serialize nested values

deserialize

@abstractmethod
def deserialize(
    self,
    data: dict[str, Any],
    deserialize_value: DeserializeFunc,
    target_type: type | None = None,
) -> Any

Reconstruct an object from a dict.

Parameter	Type	Default	Description
data	dict[str, Any]		The dict representation produced by serialize()
deserialize_value	Callable[[Any], Any]		Callback to recursively deserialize nested values
target_type	`type	None`	None	The resolved target class