Get message
(
self,
user_id: str,
conversation_id: str | None = None,
conversation_limit_number: int = 6,
message_limit_number: int = 6,
source: str | None = None,
)
| 66 | raise ValueError(f"{param_name} is required") |
| 67 | |
| 68 | def get_message( |
| 69 | self, |
| 70 | user_id: str, |
| 71 | conversation_id: str | None = None, |
| 72 | conversation_limit_number: int = 6, |
| 73 | message_limit_number: int = 6, |
| 74 | source: str | None = None, |
| 75 | ) -> MemOSGetMessagesResponse | None: |
| 76 | """Get message""" |
| 77 | # Validate required parameters |
| 78 | self._validate_required_params(user_id=user_id) |
| 79 | |
| 80 | url = f"{self.base_url}/get/message" |
| 81 | payload = { |
| 82 | "user_id": user_id, |
| 83 | "conversation_id": conversation_id, |
| 84 | "conversation_limit_number": conversation_limit_number, |
| 85 | "message_limit_number": message_limit_number, |
| 86 | "source": source, |
| 87 | } |
| 88 | for retry in range(MAX_RETRY_COUNT): |
| 89 | try: |
| 90 | response = requests.post( |
| 91 | url, data=json.dumps(payload), headers=self.headers, timeout=30 |
| 92 | ) |
| 93 | response.raise_for_status() |
| 94 | response_data = response.json() |
| 95 | |
| 96 | return MemOSGetMessagesResponse(**response_data) |
| 97 | except Exception as e: |
| 98 | logger.error(f"Failed to get messages (retry {retry + 1}/3): {e}") |
| 99 | if retry == MAX_RETRY_COUNT - 1: |
| 100 | raise |
| 101 | |
| 102 | def add_message( |
| 103 | self, |
nothing calls this directly
no test coverage detected