from typing import Literal, Optional import re def note( command: str, path: Optional[str] = None, content: Optional[str] = None, old_str: Optional[str] = None, new_str: Optional[str] = None, new_path: Optional[str] = None, insert_line: Optional[int] = None, query: Optional[str] = None, search_type: str = "label", ) -> str: """ Manage notes in your vault. All notes are automatically scoped to your agent. Commands: create - create new note (not attached) view - read note contents attach [content] - load into context (supports /folder/*) detach - remove from context (supports /folder/*) insert [line] - insert before line (0-indexed) or append append - add content to end of note replace - find/replace, shows diff rename - move/rename note to new path copy - duplicate note to new path delete - permanently remove list [query] - list notes (prefix filter, * for all) search [label|content] - grep notes by label or content attached - show notes currently in context Args: command: The operation to perform path: Path to the note (e.g., /projects/webapp, /todo) content: Content to insert or initial content when creating old_str: Text to find (for replace) new_str: Text to replace with (for replace) new_path: Destination path (for rename/copy) insert_line: Line number to insert before (0-indexed, omit to append) query: Search query (for list/search) search_type: Search by "label" or "content" Returns: str: Result of the operation """ import os agent_id = os.environ.get("LETTA_AGENT_ID") # Check enabled commands ("all" or "*" enables everything) all_commands = ["create", "view", "attach", "detach", "insert", "append", "replace", "rename", "copy", "delete", "list", "search", "attached"] enabled_env = os.environ.get("ENABLED_COMMANDS", "all") enabled = all_commands if enabled_env in ("all", "*") else enabled_env.split(",") if command not in enabled: return f"Error: '{command}' is disabled. Enabled: {enabled}" # Pattern to filter out legacy UUID paths uuid_pattern = re.compile(r'/\[?agent-[a-f0-9-]+\]?/') # Parameter validation path_required = ["create", "view", "attach", "detach", "insert", "append", "replace", "rename", "copy", "delete"] if command in path_required and not path: return f"Error: '{command}' requires path parameter" if command == "replace" and (not old_str or new_str is None): return "Error: 'replace' requires old_str and new_str parameters" if command in ["create", "insert", "append"] and not content: return f"Error: '{command}' requires content parameter" if command in ["rename", "copy"] and not new_path: return f"Error: '{command}' requires new_path parameter" if command == "search" and not query: return "Error: 'search' requires query parameter" # Track if directory needs updating update_directory = False result = None try: if command == "create": # Check for existing note with same path existing = list(client.blocks.list(label=path, description_search=agent_id).items) if existing: return f"Error: Note already exists: {path}" client.blocks.create( label=path, value=content, description=f"owner:{agent_id}" ) update_directory = True result = f"Created: {path}" elif command == "view": blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}" return blocks[0].value elif command == "attach": # Get currently attached block IDs to avoid duplicate attach errors agent = client.agents.retrieve(agent_id=agent_id) attached_ids = {b.id for b in agent.memory.blocks} # Handle bulk wildcard: /folder/* if path.endswith("/*"): prefix = path[:-1] # "/folder/*" → "/folder/" all_blocks = list(client.blocks.list(description_search=agent_id).items) blocks = [b for b in all_blocks if b.label and b.label.startswith(prefix) and not uuid_pattern.search(b.label)] if not blocks: return f"No notes matching: {path}" to_attach = [b for b in blocks if b.id not in attached_ids] skipped = len(blocks) - len(to_attach) for block in to_attach: client.agents.blocks.attach(agent_id=agent_id, block_id=block.id) msg = f"Attached {len(to_attach)} notes matching {path}" if skipped: msg += f" ({skipped} already attached)" return msg # Single note attach existing = list(client.blocks.list(label=path, description_search=agent_id).items) if existing: block_id = existing[0].id if block_id in attached_ids: return f"Already attached: {path}" else: new_block = client.blocks.create( label=path, value=content or "", description=f"owner:{agent_id}" ) block_id = new_block.id update_directory = True # New note created client.agents.blocks.attach(agent_id=agent_id, block_id=block_id) result = f"Attached: {path}" elif command == "detach": # Handle bulk wildcard: /folder/* if path.endswith("/*"): prefix = path[:-1] # Get currently attached block IDs agent = client.agents.retrieve(agent_id=agent_id) attached_ids = {b.id for b in agent.memory.blocks} all_blocks = list(client.blocks.list(description_search=agent_id).items) blocks = [b for b in all_blocks if b.label and b.label.startswith(prefix) and not uuid_pattern.search(b.label) and b.id in attached_ids] # Only detach if actually attached if not blocks: return f"No attached notes matching: {path}" for block in blocks: client.agents.blocks.detach(agent_id=agent_id, block_id=block.id) return f"Detached {len(blocks)} notes matching {path}" # Single note detach blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}" client.agents.blocks.detach(agent_id=agent_id, block_id=blocks[0].id) return f"Detached: {path}" elif command == "insert": blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}. Use 'attach' first." block = blocks[0] lines = block.value.split("\n") if block.value else [] if insert_line is not None: lines.insert(insert_line, content) line_info = f"line {insert_line}" else: lines.append(content) line_info = "end" client.blocks.update(block_id=block.id, value="\n".join(lines)) preview = content[:80] + "..." if len(content) > 80 else content return f"Inserted at {line_info} in {path}:\n + {preview}" elif command == "append": blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}. Use 'attach' first." block = blocks[0] if block.value: new_value = block.value + "\n" + content else: new_value = content client.blocks.update(block_id=block.id, value=new_value) preview = content[:80] + "..." if len(content) > 80 else content return f"Appended to {path}:\n + {preview}" elif command == "rename": # Check source exists blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}" # Check destination doesn't exist dest_blocks = list(client.blocks.list(label=new_path, description_search=agent_id).items) if dest_blocks: return f"Error: Destination already exists: {new_path}" # Update the label block = blocks[0] client.blocks.update(block_id=block.id, label=new_path) update_directory = True result = f"Renamed: {path} → {new_path}" elif command == "copy": # Check source exists blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}" # Check destination doesn't exist dest_blocks = list(client.blocks.list(label=new_path, description_search=agent_id).items) if dest_blocks: return f"Error: Destination already exists: {new_path}" # Create copy (not attached) source = blocks[0] client.blocks.create( label=new_path, value=source.value, description=f"owner:{agent_id}" ) update_directory = True result = f"Copied: {path} → {new_path}" elif command == "replace": blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}" block = blocks[0] if old_str not in block.value: return f"Error: old_str not found in note. Exact match required." new_value = block.value.replace(old_str, new_str, 1) client.blocks.update(block_id=block.id, value=new_value) return f"Replaced in {path}:\n - {old_str}\n + {new_str}" elif command == "delete": blocks = list(client.blocks.list(label=path, description_search=agent_id).items) if not blocks: return f"Note not found: {path}" client.blocks.delete(block_id=blocks[0].id) update_directory = True result = f"Deleted: {path}" elif command == "list": all_blocks = list(client.blocks.list(description_search=agent_id).items) # Filter to path-like labels, exclude legacy UUID paths blocks = [b for b in all_blocks if b.label and b.label.startswith("/") and not uuid_pattern.search(b.label)] # Apply prefix filter if query provided if query and query != "*": blocks = [b for b in blocks if b.label.startswith(query)] if not blocks: return "No notes found" if not query or query == "*" else f"No notes matching: {query}" # Deduplicate and sort labels = sorted(set(b.label for b in blocks)) return "\n".join(labels) elif command == "search": all_blocks = list(client.blocks.list(description_search=agent_id).items) # Filter out legacy UUID paths all_blocks = [b for b in all_blocks if b.label and b.label.startswith("/") and not uuid_pattern.search(b.label)] if search_type == "label": blocks = [b for b in all_blocks if query in b.label] else: blocks = [b for b in all_blocks if b.value and query in b.value] if not blocks: return f"No notes matching: {query}" results = [] for b in blocks: preview = b.value[:100].replace("\n", " ") if b.value else "" if len(b.value or "") > 100: preview += "..." results.append(f"{b.label}: {preview}") return "\n".join(results) elif command == "attached": agent = client.agents.retrieve(agent_id=agent_id) note_blocks = [b for b in agent.memory.blocks if b.label and b.label.startswith("/") and not uuid_pattern.search(b.label)] if not note_blocks: return "No notes currently attached" return "\n".join(sorted(b.label for b in note_blocks)) else: return f"Error: Unknown command '{command}'" # Update note_directory if needed if update_directory: dir_label = "/note_directory" # Get all notes all_blocks = list(client.blocks.list(description_search=agent_id).items) notes = [b for b in all_blocks if b.label and b.label.startswith("/") and b.label != dir_label and not uuid_pattern.search(b.label)] # Header for the directory header = "External storage. Attach to load into context, detach when done.\nFolders are also notes (e.g., /projects and /projects/task1 can both have content).\nCommands: view, attach, detach, insert, append, replace, rename, copy, delete, list, search\nBulk: attach /folder/*, detach /folder/*" if notes: # Group notes by folder folders = {} for b in sorted(notes, key=lambda x: x.label): parts = b.label.rsplit("/", 1) if len(parts) == 2: folder, name = parts[0] + "/", parts[1] else: folder, name = "/", b.label[1:] # Root level if folder not in folders: folders[folder] = [] first_line = (b.value or "").split("\n")[0][:80] if len((b.value or "").split("\n")[0]) > 80: first_line += "..." folders[folder].append((name, first_line)) # Build tree view lines = [] for folder in sorted(folders.keys()): lines.append(folder) items = folders[folder] max_name_len = max(len(name) for name, _ in items) for name, summary in items: lines.append(f" {name.ljust(max_name_len)} | {summary}") dir_content = header + "\n\n" + "\n".join(lines) else: dir_content = header + "\n\n(no notes)" # Find or create directory block dir_blocks = list(client.blocks.list(label=dir_label, description_search=agent_id).items) if dir_blocks: client.blocks.update(block_id=dir_blocks[0].id, value=dir_content) else: # Create and attach directory block dir_block = client.blocks.create( label=dir_label, value=dir_content, description=f"owner:{agent_id}" ) client.agents.blocks.attach(agent_id=agent_id, block_id=dir_block.id) if result: return result except Exception as e: return f"Error executing '{command}': {str(e)}"