""" Session management with JSONL format """ import json import os from datetime import datetime from typing import List, Dict, Any, Optional class Session: """Session manager for conversation history""" def __init__(self, session_file: Optional[str] = None): self.session_file = session_file self.messages: List[Dict[str, Any]] = [] def add_message(self, role: str, content: str, timestamp: Optional[str] = None): """Add a message to the session""" if timestamp is None: timestamp = datetime.utcnow().isoformat() message = { 'role': role, 'content': content, 'timestamp': timestamp } self.messages.append(message) def add_tool_use(self, tool_name: str, tool_input: Dict[str, Any], tool_id: str): """Add a tool use message""" content = [{ 'type': 'tool_use', 'id': tool_id, 'name': tool_name, 'input': tool_input }] self.add_message('assistant', content) def add_tool_result(self, tool_id: str, result: str): """Add a tool result message""" content = [{ 'type': 'tool_result', 'tool_use_id': tool_id, 'content': result }] self.add_message('user', content) def get_messages(self) -> List[Dict[str, Any]]: """Get all messages in the session""" return self.messages.copy() def save(self, filename: Optional[str] = None): """Save session to JSONL file""" if filename is None and self.session_file is None: raise ValueError("No filename specified for saving session") save_file = filename or self.session_file with open(save_file, 'w') as f: for message in self.messages: f.write(json.dumps(message) + '\n') def load(self, filename: Optional[str] = None): """Load session from JSONL file""" load_file = filename or self.session_file if not os.path.exists(load_file): return # Empty session if file doesn't exist self.messages = [] with open(load_file, 'r') as f: for line in f: line = line.strip() if line: message = json.loads(line) self.messages.append(message) def clear(self): """Clear all messages from the session""" self.messages = [] def get_anthropic_messages(self) -> List[Dict[str, Any]]: """Format messages for Anthropic API""" anthropic_messages = [] for message in self.messages: # Skip system messages (handled separately) if message['role'] == 'system': continue anthropic_messages.append({ 'role': message['role'], 'content': message['content'] }) return anthropic_messages