Core message structure for all communication. Separates WHO sent it (sender_type) from WHAT it contains (message_type).
| 95 | |
| 96 | @dataclass |
| 97 | class Message: |
| 98 | """ |
| 99 | Core message structure for all communication. |
| 100 | Separates WHO sent it (sender_type) from WHAT it contains (message_type). |
| 101 | """ |
| 102 | message_id: str = field(default_factory=lambda: str(uuid.uuid4())) |
| 103 | sender_type: SenderType = SenderType.USER |
| 104 | message_type: MessageType = MessageType.QUERY |
| 105 | conversation_id: str | None = None |
| 106 | timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat()) |
| 107 | content: Union[str, UserQuery, dict[str, Any], list[Any]] = "" |
| 108 | sender_info: dict[str, Any] | None = None |
| 109 | metadata: dict[str, Any] | None = None |
| 110 | |
| 111 | def to_dict(self) -> dict[str, Any]: |
| 112 | """Convert message to dictionary.""" |
| 113 | result = { |
| 114 | "message_id": self.message_id, |
| 115 | "sender_type": self.sender_type.value if isinstance(self.sender_type, SenderType) else self.sender_type, |
| 116 | "message_type": self.message_type.value if isinstance(self.message_type, MessageType) else self.message_type, |
| 117 | "timestamp": self.timestamp |
| 118 | } |
| 119 | |
| 120 | # Handle content serialization |
| 121 | if isinstance(self.content, UserQuery): |
| 122 | result["content"] = self.content.to_dict() |
| 123 | elif isinstance(self.content, (str, dict, list)): |
| 124 | result["content"] = self.content |
| 125 | else: |
| 126 | result["content"] = str(self.content) |
| 127 | |
| 128 | # Add optional fields |
| 129 | if self.conversation_id is not None: |
| 130 | result["conversation_id"] = self.conversation_id |
| 131 | if self.sender_info is not None: |
| 132 | result["sender_info"] = self.sender_info |
| 133 | if self.metadata is not None: |
| 134 | result["metadata"] = self.metadata |
| 135 | |
| 136 | return result |
| 137 | |
| 138 | def to_json(self) -> str: |
| 139 | """Convert message to JSON string.""" |
| 140 | return json.dumps(self.to_dict()) |
| 141 | |
| 142 | @classmethod |
| 143 | def from_dict(cls, data: dict[str, Any]) -> 'Message': |
| 144 | """Create Message from dictionary.""" |
| 145 | # Determine content type |
| 146 | content = data.get("content", "") |
| 147 | sender_type = data.get("sender_type", "user") |
| 148 | message_type = data.get("message_type", "query") |
| 149 | |
| 150 | # Keep content as dict to preserve all parameters |
| 151 | # We don't convert to UserQuery anymore because it loses extra fields like 'db' |
| 152 | # The dict already has everything we need |
| 153 | |
| 154 | # Convert string types to enums if valid |
no outgoing calls
no test coverage detected