Add a message to a session
(self,
session_id: str,
role: str,
content: str,
message_type: str = 'text',
metadata: Dict[str, Any] = None)
| 79 | return list(self._sessions.values()) |
| 80 | |
| 81 | def add_message(self, |
| 82 | session_id: str, |
| 83 | role: str, |
| 84 | content: str, |
| 85 | message_type: str = 'text', |
| 86 | metadata: Dict[str, Any] = None) -> bool: |
| 87 | """Add a message to a session""" |
| 88 | if session_id not in self._sessions: |
| 89 | return False |
| 90 | |
| 91 | message = { |
| 92 | 'id': str(uuid.uuid4()), |
| 93 | 'role': role, # user, assistant, system, tool |
| 94 | 'content': content, |
| 95 | 'type': message_type, # text, tool_call, tool_result, error, log |
| 96 | 'timestamp': datetime.now().isoformat(), |
| 97 | 'metadata': metadata or {} |
| 98 | } |
| 99 | |
| 100 | with self._lock: |
| 101 | if session_id not in self._messages: |
| 102 | self._messages[session_id] = [] |
| 103 | self._messages[session_id].append(message) |
| 104 | |
| 105 | return True |
| 106 | |
| 107 | def get_messages(self, session_id: str) -> Optional[List[Dict[str, Any]]]: |
| 108 | """Get all messages for a session""" |
no outgoing calls
no test coverage detected