cameron.stream
[READ-ONLY] Mirror of https://github.com/just-cameron/note. External storage tool for Letta agents to manage persistent notes
1from typing import Literal, Optional
2import re
3
4
5def note(
6 command: str,
7 path: Optional[str] = None,
8 content: Optional[str] = None,
9 old_str: Optional[str] = None,
10 new_str: Optional[str] = None,
11 new_path: Optional[str] = None,
12 insert_line: Optional[int] = None,
13 query: Optional[str] = None,
14 search_type: str = "label",
15) -> str:
16 """
17 Manage notes in your vault. All notes are automatically scoped to your agent.
18
19 Commands:
20 create <path> <content> - create new note (not attached)
21 view <path> - read note contents
22 attach <path> [content] - load into context (supports /folder/*)
23 detach <path> - remove from context (supports /folder/*)
24 insert <path> <content> [line] - insert before line (0-indexed) or append
25 append <path> <content> - add content to end of note
26 replace <path> <old_str> <new_str> - find/replace, shows diff
27 rename <path> <new_path> - move/rename note to new path
28 copy <path> <new_path> - duplicate note to new path
29 delete <path> - permanently remove
30 list [query] - list notes (prefix filter, * for all)
31 search <query> [label|content] - grep notes by label or content
32 attached - show notes currently in context
33
34 Args:
35 command: The operation to perform
36 path: Path to the note (e.g., /projects/webapp, /todo)
37 content: Content to insert or initial content when creating
38 old_str: Text to find (for replace)
39 new_str: Text to replace with (for replace)
40 new_path: Destination path (for rename/copy)
41 insert_line: Line number to insert before (0-indexed, omit to append)
42 query: Search query (for list/search)
43 search_type: Search by "label" or "content"
44
45 Returns:
46 str: Result of the operation
47 """
48 import os
49
50 agent_id = os.environ.get("LETTA_AGENT_ID")
51
52 # Check enabled commands ("all" or "*" enables everything)
53 all_commands = ["create", "view", "attach", "detach", "insert", "append", "replace", "rename", "copy", "delete", "list", "search", "attached"]
54 enabled_env = os.environ.get("ENABLED_COMMANDS", "all")
55 enabled = all_commands if enabled_env in ("all", "*") else enabled_env.split(",")
56 if command not in enabled:
57 return f"Error: '{command}' is disabled. Enabled: {enabled}"
58
59 # Pattern to filter out legacy UUID paths
60 uuid_pattern = re.compile(r'/\[?agent-[a-f0-9-]+\]?/')
61
62 # Parameter validation
63 path_required = ["create", "view", "attach", "detach", "insert", "append", "replace", "rename", "copy", "delete"]
64 if command in path_required and not path:
65 return f"Error: '{command}' requires path parameter"
66
67 if command == "replace" and (not old_str or new_str is None):
68 return "Error: 'replace' requires old_str and new_str parameters"
69
70 if command in ["create", "insert", "append"] and not content:
71 return f"Error: '{command}' requires content parameter"
72
73 if command in ["rename", "copy"] and not new_path:
74 return f"Error: '{command}' requires new_path parameter"
75
76 if command == "search" and not query:
77 return "Error: 'search' requires query parameter"
78
79 # Track if directory needs updating
80 update_directory = False
81 result = None
82
83 try:
84 if command == "create":
85 # Check for existing note with same path
86 existing = list(client.blocks.list(label=path, description_search=agent_id).items)
87 if existing:
88 return f"Error: Note already exists: {path}"
89
90 client.blocks.create(
91 label=path,
92 value=content,
93 description=f"owner:{agent_id}"
94 )
95 update_directory = True
96 result = f"Created: {path}"
97
98 elif command == "view":
99 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
100 if not blocks:
101 return f"Note not found: {path}"
102 return blocks[0].value
103
104 elif command == "attach":
105 # Get currently attached block IDs to avoid duplicate attach errors
106 agent = client.agents.retrieve(agent_id=agent_id)
107 attached_ids = {b.id for b in agent.memory.blocks}
108
109 # Handle bulk wildcard: /folder/*
110 if path.endswith("/*"):
111 prefix = path[:-1] # "/folder/*" → "/folder/"
112 all_blocks = list(client.blocks.list(description_search=agent_id).items)
113 blocks = [b for b in all_blocks if b.label and b.label.startswith(prefix)
114 and not uuid_pattern.search(b.label)]
115 if not blocks:
116 return f"No notes matching: {path}"
117
118 to_attach = [b for b in blocks if b.id not in attached_ids]
119 skipped = len(blocks) - len(to_attach)
120
121 for block in to_attach:
122 client.agents.blocks.attach(agent_id=agent_id, block_id=block.id)
123
124 msg = f"Attached {len(to_attach)} notes matching {path}"
125 if skipped:
126 msg += f" ({skipped} already attached)"
127 return msg
128
129 # Single note attach
130 existing = list(client.blocks.list(label=path, description_search=agent_id).items)
131 if existing:
132 block_id = existing[0].id
133 if block_id in attached_ids:
134 return f"Already attached: {path}"
135 else:
136 new_block = client.blocks.create(
137 label=path,
138 value=content or "",
139 description=f"owner:{agent_id}"
140 )
141 block_id = new_block.id
142 update_directory = True # New note created
143
144 client.agents.blocks.attach(agent_id=agent_id, block_id=block_id)
145 result = f"Attached: {path}"
146
147 elif command == "detach":
148 # Handle bulk wildcard: /folder/*
149 if path.endswith("/*"):
150 prefix = path[:-1]
151 # Get currently attached block IDs
152 agent = client.agents.retrieve(agent_id=agent_id)
153 attached_ids = {b.id for b in agent.memory.blocks}
154
155 all_blocks = list(client.blocks.list(description_search=agent_id).items)
156 blocks = [b for b in all_blocks if b.label and b.label.startswith(prefix)
157 and not uuid_pattern.search(b.label)
158 and b.id in attached_ids] # Only detach if actually attached
159 if not blocks:
160 return f"No attached notes matching: {path}"
161 for block in blocks:
162 client.agents.blocks.detach(agent_id=agent_id, block_id=block.id)
163 return f"Detached {len(blocks)} notes matching {path}"
164
165 # Single note detach
166 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
167 if not blocks:
168 return f"Note not found: {path}"
169
170 client.agents.blocks.detach(agent_id=agent_id, block_id=blocks[0].id)
171 return f"Detached: {path}"
172
173 elif command == "insert":
174 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
175 if not blocks:
176 return f"Note not found: {path}. Use 'attach' first."
177
178 block = blocks[0]
179 lines = block.value.split("\n") if block.value else []
180
181 if insert_line is not None:
182 lines.insert(insert_line, content)
183 line_info = f"line {insert_line}"
184 else:
185 lines.append(content)
186 line_info = "end"
187
188 client.blocks.update(block_id=block.id, value="\n".join(lines))
189
190 preview = content[:80] + "..." if len(content) > 80 else content
191 return f"Inserted at {line_info} in {path}:\n + {preview}"
192
193 elif command == "append":
194 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
195 if not blocks:
196 return f"Note not found: {path}. Use 'attach' first."
197
198 block = blocks[0]
199 if block.value:
200 new_value = block.value + "\n" + content
201 else:
202 new_value = content
203
204 client.blocks.update(block_id=block.id, value=new_value)
205
206 preview = content[:80] + "..." if len(content) > 80 else content
207 return f"Appended to {path}:\n + {preview}"
208
209 elif command == "rename":
210 # Check source exists
211 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
212 if not blocks:
213 return f"Note not found: {path}"
214
215 # Check destination doesn't exist
216 dest_blocks = list(client.blocks.list(label=new_path, description_search=agent_id).items)
217 if dest_blocks:
218 return f"Error: Destination already exists: {new_path}"
219
220 # Update the label
221 block = blocks[0]
222 client.blocks.update(block_id=block.id, label=new_path)
223 update_directory = True
224 result = f"Renamed: {path} → {new_path}"
225
226 elif command == "copy":
227 # Check source exists
228 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
229 if not blocks:
230 return f"Note not found: {path}"
231
232 # Check destination doesn't exist
233 dest_blocks = list(client.blocks.list(label=new_path, description_search=agent_id).items)
234 if dest_blocks:
235 return f"Error: Destination already exists: {new_path}"
236
237 # Create copy (not attached)
238 source = blocks[0]
239 client.blocks.create(
240 label=new_path,
241 value=source.value,
242 description=f"owner:{agent_id}"
243 )
244 update_directory = True
245 result = f"Copied: {path} → {new_path}"
246
247 elif command == "replace":
248 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
249 if not blocks:
250 return f"Note not found: {path}"
251
252 block = blocks[0]
253 if old_str not in block.value:
254 return f"Error: old_str not found in note. Exact match required."
255
256 new_value = block.value.replace(old_str, new_str, 1)
257 client.blocks.update(block_id=block.id, value=new_value)
258
259 return f"Replaced in {path}:\n - {old_str}\n + {new_str}"
260
261 elif command == "delete":
262 blocks = list(client.blocks.list(label=path, description_search=agent_id).items)
263 if not blocks:
264 return f"Note not found: {path}"
265
266 client.blocks.delete(block_id=blocks[0].id)
267 update_directory = True
268 result = f"Deleted: {path}"
269
270 elif command == "list":
271 all_blocks = list(client.blocks.list(description_search=agent_id).items)
272
273 # Filter to path-like labels, exclude legacy UUID paths
274 blocks = [b for b in all_blocks
275 if b.label and b.label.startswith("/")
276 and not uuid_pattern.search(b.label)]
277
278 # Apply prefix filter if query provided
279 if query and query != "*":
280 blocks = [b for b in blocks if b.label.startswith(query)]
281
282 if not blocks:
283 return "No notes found" if not query or query == "*" else f"No notes matching: {query}"
284
285 # Deduplicate and sort
286 labels = sorted(set(b.label for b in blocks))
287 return "\n".join(labels)
288
289 elif command == "search":
290 all_blocks = list(client.blocks.list(description_search=agent_id).items)
291
292 # Filter out legacy UUID paths
293 all_blocks = [b for b in all_blocks
294 if b.label and b.label.startswith("/")
295 and not uuid_pattern.search(b.label)]
296
297 if search_type == "label":
298 blocks = [b for b in all_blocks if query in b.label]
299 else:
300 blocks = [b for b in all_blocks if b.value and query in b.value]
301
302 if not blocks:
303 return f"No notes matching: {query}"
304
305 results = []
306 for b in blocks:
307 preview = b.value[:100].replace("\n", " ") if b.value else ""
308 if len(b.value or "") > 100:
309 preview += "..."
310 results.append(f"{b.label}: {preview}")
311
312 return "\n".join(results)
313
314 elif command == "attached":
315 agent = client.agents.retrieve(agent_id=agent_id)
316 note_blocks = [b for b in agent.memory.blocks
317 if b.label and b.label.startswith("/")
318 and not uuid_pattern.search(b.label)]
319
320 if not note_blocks:
321 return "No notes currently attached"
322
323 return "\n".join(sorted(b.label for b in note_blocks))
324
325 else:
326 return f"Error: Unknown command '{command}'"
327
328 # Update note_directory if needed
329 if update_directory:
330 dir_label = "/note_directory"
331 # Get all notes
332 all_blocks = list(client.blocks.list(description_search=agent_id).items)
333 notes = [b for b in all_blocks
334 if b.label and b.label.startswith("/")
335 and b.label != dir_label
336 and not uuid_pattern.search(b.label)]
337
338 # Header for the directory
339 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/*"
340
341 if notes:
342 # Group notes by folder
343 folders = {}
344 for b in sorted(notes, key=lambda x: x.label):
345 parts = b.label.rsplit("/", 1)
346 if len(parts) == 2:
347 folder, name = parts[0] + "/", parts[1]
348 else:
349 folder, name = "/", b.label[1:] # Root level
350 if folder not in folders:
351 folders[folder] = []
352 first_line = (b.value or "").split("\n")[0][:80]
353 if len((b.value or "").split("\n")[0]) > 80:
354 first_line += "..."
355 folders[folder].append((name, first_line))
356
357 # Build tree view
358 lines = []
359 for folder in sorted(folders.keys()):
360 lines.append(folder)
361 items = folders[folder]
362 max_name_len = max(len(name) for name, _ in items)
363 for name, summary in items:
364 lines.append(f" {name.ljust(max_name_len)} | {summary}")
365
366 dir_content = header + "\n\n" + "\n".join(lines)
367 else:
368 dir_content = header + "\n\n(no notes)"
369
370 # Find or create directory block
371 dir_blocks = list(client.blocks.list(label=dir_label, description_search=agent_id).items)
372 if dir_blocks:
373 client.blocks.update(block_id=dir_blocks[0].id, value=dir_content)
374 else:
375 # Create and attach directory block
376 dir_block = client.blocks.create(
377 label=dir_label,
378 value=dir_content,
379 description=f"owner:{agent_id}"
380 )
381 client.agents.blocks.attach(agent_id=agent_id, block_id=dir_block.id)
382
383 if result:
384 return result
385
386 except Exception as e:
387 return f"Error executing '{command}': {str(e)}"