Documentation Index
Fetch the complete documentation index at: https://docs.pipecat.ai/llms.txt
Use this file to discover all available pages before exploring further.
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
| Parameter | Type | Description |
|---|
message | BusMessage | The bus message to serialize |
deserialize
@abstractmethod
def deserialize(self, data: bytes) -> BusMessage | None
| Parameter | Type | Description |
|---|
data | bytes | The serialized bytes produced by serialize() |
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)
RedisBus and the proxy agents.
Methods
register_adapter
def register_adapter(self, type_: type, adapter: TypeAdapter) -> None
| 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]
| 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
| 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 |
