this repo has no description
1"""Block management tools for user-specific memory blocks."""
2from pydantic import BaseModel, Field
3from typing import List, Dict, Any
4
5def get_letta_client():
6 """Get a Letta client using configuration."""
7 try:
8 from config_loader import get_letta_config
9 from letta_client import Letta
10 config = get_letta_config()
11 return Letta(token=config['api_key'], timeout=config['timeout'])
12 except (ImportError, FileNotFoundError, KeyError):
13 # Fallback to environment variable
14 import os
15 from letta_client import Letta
16 return Letta(token=os.environ["LETTA_API_KEY"])
17
18
19class AttachUserBlocksArgs(BaseModel):
20 handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])")
21
22
23class DetachUserBlocksArgs(BaseModel):
24 handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])")
25
26
27class UserNoteAppendArgs(BaseModel):
28 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
29 note: str = Field(..., description="Note to append to the user's memory block (e.g., '\\n- Cameron is a person')")
30
31
32class UserNoteReplaceArgs(BaseModel):
33 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
34 old_text: str = Field(..., description="Text to find and replace in the user's memory block")
35 new_text: str = Field(..., description="Text to replace the old_text with")
36
37
38class UserNoteSetArgs(BaseModel):
39 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
40 content: str = Field(..., description="Complete content to set for the user's memory block")
41
42
43class UserNoteViewArgs(BaseModel):
44 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
45
46
47# X (Twitter) User Block Management
48class AttachXUserBlocksArgs(BaseModel):
49 user_ids: List[str] = Field(..., description="List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])")
50
51
52class DetachXUserBlocksArgs(BaseModel):
53 user_ids: List[str] = Field(..., description="List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])")
54
55
56class XUserNoteAppendArgs(BaseModel):
57 user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')")
58 note: str = Field(..., description="Note to append to the user's memory block (e.g., '\\\\n- Cameron is a person')")
59
60
61class XUserNoteReplaceArgs(BaseModel):
62 user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')")
63 old_text: str = Field(..., description="Text to find and replace in the user's memory block")
64 new_text: str = Field(..., description="Text to replace the old_text with")
65
66
67class XUserNoteSetArgs(BaseModel):
68 user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')")
69 content: str = Field(..., description="Complete content to set for the user's memory block")
70
71
72class XUserNoteViewArgs(BaseModel):
73 user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')")
74
75
76
77def attach_user_blocks(handles: list, agent_state: "AgentState") -> str:
78 """
79 Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.
80
81 Args:
82 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
83 agent_state: The agent state object containing agent information
84
85 Returns:
86 String with attachment results for each handle
87 """
88
89 handles = list(set(handles))
90
91 try:
92 # Try to get client the local way first, fall back to inline for cloud execution
93 try:
94 client = get_letta_client()
95 except (ImportError, NameError):
96 # Create Letta client inline for cloud execution
97 import os
98 from letta_client import Letta
99 client = Letta(token=os.environ["LETTA_API_KEY"])
100 results = []
101
102 # Get current blocks using the API
103 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
104 current_block_labels = set()
105 current_block_ids = set()
106
107 for block in current_blocks:
108 current_block_labels.add(block.label)
109 current_block_ids.add(str(block.id))
110
111 for handle in handles:
112 # Sanitize handle for block label - completely self-contained
113 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
114 block_label = f"user_{clean_handle}"
115
116 # Skip if already attached
117 if block_label in current_block_labels:
118 results.append(f"✓ {handle}: Already attached")
119 continue
120
121 # Check if block exists or create new one
122 try:
123 blocks = client.blocks.list(label=block_label)
124 if blocks and len(blocks) > 0:
125 block = blocks[0]
126
127 # Double-check if this block is already attached by ID
128 if str(block.id) in current_block_ids:
129 results.append(f"✓ {handle}: Already attached (by ID)")
130 continue
131 else:
132 block = client.blocks.create(
133 label=block_label,
134 value=f"# User: {handle}\n\nNo information about this user yet.",
135 limit=5000
136 )
137
138 # Attach block atomically
139 try:
140 client.agents.blocks.attach(
141 agent_id=str(agent_state.id),
142 block_id=str(block.id)
143 )
144 results.append(f"✓ {handle}: Block attached")
145 except Exception as attach_error:
146 # Check if it's a duplicate constraint error
147 error_str = str(attach_error)
148 if "duplicate key value violates unique constraint" in error_str and "unique_label_per_agent" in error_str:
149 # Block is already attached, possibly with this exact label
150 results.append(f"✓ {handle}: Already attached (verified)")
151 else:
152 # Re-raise other errors
153 raise attach_error
154
155 except Exception as e:
156 results.append(f"✗ {handle}: Error - {str(e)}")
157
158 return f"Attachment results:\n" + "\n".join(results)
159
160 except Exception as e:
161 raise Exception(f"Error attaching user blocks: {str(e)}")
162
163
164def detach_user_blocks(handles: list, agent_state: "AgentState") -> str:
165 """
166 Detach user-specific memory blocks from the agent. Blocks are preserved for later use.
167
168 Args:
169 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
170 agent_state: The agent state object containing agent information
171
172 Returns:
173 String with detachment results for each handle
174 """
175
176 try:
177 # Try to get client the local way first, fall back to inline for cloud execution
178 try:
179 client = get_letta_client()
180 except (ImportError, NameError):
181 # Create Letta client inline for cloud execution
182 import os
183 from letta_client import Letta
184 client = Letta(token=os.environ["LETTA_API_KEY"])
185 results = []
186
187 # Build mapping of block labels to IDs using the API
188 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
189 block_label_to_id = {}
190
191 for block in current_blocks:
192 block_label_to_id[block.label] = str(block.id)
193
194 # Process each handle and detach atomically
195 for handle in handles:
196 # Sanitize handle for block label - completely self-contained
197 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
198 block_label = f"user_{clean_handle}"
199
200 if block_label in block_label_to_id:
201 try:
202 # Detach block atomically
203 client.agents.blocks.detach(
204 agent_id=str(agent_state.id),
205 block_id=block_label_to_id[block_label]
206 )
207 results.append(f"✓ {handle}: Detached")
208 except Exception as e:
209 results.append(f"✗ {handle}: Error during detachment - {str(e)}")
210 else:
211 results.append(f"✗ {handle}: Not attached")
212
213 return f"Detachment results:\n" + "\n".join(results)
214
215 except Exception as e:
216 raise Exception(f"Error detaching user blocks: {str(e)}")
217
218
219def user_note_append(handle: str, note: str, agent_state: "AgentState") -> str:
220 """
221 Append a note to a user's memory block. Creates the block if it doesn't exist.
222
223 Args:
224 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
225 note: Note to append to the user's memory block
226 agent_state: The agent state object containing agent information
227
228 Returns:
229 String confirming the note was appended
230 """
231
232 try:
233 # Create Letta client inline - cloud tools must be self-contained
234 import os
235 from letta_client import Letta
236 client = Letta(token=os.environ["LETTA_API_KEY"])
237
238 # Sanitize handle for block label
239 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
240 block_label = f"user_{clean_handle}"
241
242 # Check if block exists
243 blocks = client.blocks.list(label=block_label)
244
245 if blocks and len(blocks) > 0:
246 # Block exists, append to it
247 block = blocks[0]
248 current_value = block.value
249 new_value = current_value + note
250
251 # Update the block
252 client.blocks.modify(
253 block_id=str(block.id),
254 value=new_value
255 )
256 return f"✓ Appended note to {handle}'s memory block"
257
258 else:
259 # Block doesn't exist, create it with the note
260 initial_value = f"# User: {handle}\n\n{note}"
261 block = client.blocks.create(
262 label=block_label,
263 value=initial_value,
264 limit=5000
265 )
266
267 # Check if block needs to be attached to agent
268 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
269 current_block_labels = {block.label for block in current_blocks}
270
271 if block_label not in current_block_labels:
272 # Attach the new block to the agent
273 client.agents.blocks.attach(
274 agent_id=str(agent_state.id),
275 block_id=str(block.id)
276 )
277 return f"✓ Created and attached {handle}'s memory block with note"
278 else:
279 return f"✓ Created {handle}'s memory block with note"
280
281 except Exception as e:
282 raise Exception(f"Error appending note to user block: {str(e)}")
283
284
285def user_note_replace(handle: str, old_text: str, new_text: str, agent_state: "AgentState") -> str:
286 """
287 Replace text in a user's memory block.
288
289 Args:
290 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
291 old_text: Text to find and replace
292 new_text: Text to replace the old_text with
293 agent_state: The agent state object containing agent information
294
295 Returns:
296 String confirming the text was replaced
297 """
298
299 try:
300 # Create Letta client inline - cloud tools must be self-contained
301 import os
302 from letta_client import Letta
303 client = Letta(token=os.environ["LETTA_API_KEY"])
304
305 # Sanitize handle for block label
306 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
307 block_label = f"user_{clean_handle}"
308
309 # Check if block exists
310 blocks = client.blocks.list(label=block_label)
311
312 if not blocks or len(blocks) == 0:
313 raise Exception(f"No memory block found for user: {handle}")
314
315 block = blocks[0]
316 current_value = block.value
317
318 # Check if old_text exists in the block
319 if old_text not in current_value:
320 raise Exception(f"Text '{old_text}' not found in {handle}'s memory block")
321
322 # Replace the text
323 new_value = current_value.replace(old_text, new_text)
324
325 # Update the block
326 client.blocks.modify(
327 block_id=str(block.id),
328 value=new_value
329 )
330 return f"✓ Replaced text in {handle}'s memory block"
331
332 except Exception as e:
333 raise Exception(f"Error replacing text in user block: {str(e)}")
334
335
336def user_note_set(handle: str, content: str, agent_state: "AgentState") -> str:
337 """
338 Set the complete content of a user's memory block.
339
340 Args:
341 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
342 content: Complete content to set for the memory block
343 agent_state: The agent state object containing agent information
344
345 Returns:
346 String confirming the content was set
347 """
348
349 try:
350 # Create Letta client inline - cloud tools must be self-contained
351 import os
352 from letta_client import Letta
353 client = Letta(token=os.environ["LETTA_API_KEY"])
354
355 # Sanitize handle for block label
356 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
357 block_label = f"user_{clean_handle}"
358
359 # Check if block exists
360 blocks = client.blocks.list(label=block_label)
361
362 if blocks and len(blocks) > 0:
363 # Block exists, update it
364 block = blocks[0]
365 client.blocks.modify(
366 block_id=str(block.id),
367 value=content
368 )
369 return f"✓ Set content for {handle}'s memory block"
370
371 else:
372 # Block doesn't exist, create it
373 block = client.blocks.create(
374 label=block_label,
375 value=content,
376 limit=5000
377 )
378
379 # Check if block needs to be attached to agent
380 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
381 current_block_labels = {block.label for block in current_blocks}
382
383 if block_label not in current_block_labels:
384 # Attach the new block to the agent
385 client.agents.blocks.attach(
386 agent_id=str(agent_state.id),
387 block_id=str(block.id)
388 )
389 return f"✓ Created and attached {handle}'s memory block"
390 else:
391 return f"✓ Created {handle}'s memory block"
392
393 except Exception as e:
394 raise Exception(f"Error setting user block content: {str(e)}")
395
396
397def user_note_view(handle: str, agent_state: "AgentState") -> str:
398 """
399 View the content of a user's memory block.
400
401 Args:
402 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
403 agent_state: The agent state object containing agent information
404
405 Returns:
406 String containing the user's memory block content
407 """
408
409 try:
410 # Create Letta client inline - cloud tools must be self-contained
411 import os
412 from letta_client import Letta
413 client = Letta(token=os.environ["LETTA_API_KEY"])
414
415 # Sanitize handle for block label
416 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
417 block_label = f"user_{clean_handle}"
418
419 # Check if block exists
420 blocks = client.blocks.list(label=block_label)
421
422 if not blocks or len(blocks) == 0:
423 return f"No memory block found for user: {handle}"
424
425 block = blocks[0]
426
427 return f"Memory block for {handle}:\n\n{block.value}"
428
429 except Exception as e:
430 raise Exception(f"Error viewing user block: {str(e)}")
431
432
433# X (Twitter) User Block Management Functions
434
435def attach_x_user_blocks(user_ids: list, agent_state: "AgentState") -> str:
436 """
437 Attach X user-specific memory blocks to the agent. Creates blocks if they don't exist.
438
439 Args:
440 user_ids: List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])
441 agent_state: The agent state object containing agent information
442
443 Returns:
444 String with attachment results for each user ID
445 """
446
447 user_ids = list(set(user_ids))
448
449 try:
450 # Try to get client the local way first, fall back to inline for cloud execution
451 try:
452 client = get_letta_client()
453 except (ImportError, NameError):
454 # Create Letta client inline for cloud execution
455 import os
456 from letta_client import Letta
457 client = Letta(token=os.environ["LETTA_API_KEY"])
458 results = []
459
460 # Get current blocks using the API
461 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
462 current_block_labels = set()
463
464 for block in current_blocks:
465 current_block_labels.add(block.label)
466
467 for user_id in user_ids:
468 # Create block label with x_user_ prefix
469 block_label = f"x_user_{user_id}"
470
471 # Skip if already attached
472 if block_label in current_block_labels:
473 results.append(f"✓ {user_id}: Already attached")
474 continue
475
476 # Check if block exists or create new one
477 try:
478 blocks = client.blocks.list(label=block_label)
479 if blocks and len(blocks) > 0:
480 block = blocks[0]
481 else:
482 block = client.blocks.create(
483 label=block_label,
484 value=f"# X User: {user_id}\n\nNo information about this user yet.",
485 limit=5000
486 )
487
488 # Attach block atomically
489 client.agents.blocks.attach(
490 agent_id=str(agent_state.id),
491 block_id=str(block.id)
492 )
493
494 results.append(f"✓ {user_id}: Block attached")
495
496 except Exception as e:
497 results.append(f"✗ {user_id}: Error - {str(e)}")
498
499 return f"X user attachment results:\n" + "\n".join(results)
500
501 except Exception as e:
502 raise Exception(f"Error attaching X user blocks: {str(e)}")
503
504
505def detach_x_user_blocks(user_ids: list, agent_state: "AgentState") -> str:
506 """
507 Detach X user-specific memory blocks from the agent. Blocks are preserved for later use.
508
509 Args:
510 user_ids: List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])
511 agent_state: The agent state object containing agent information
512
513 Returns:
514 String with detachment results for each user ID
515 """
516
517 try:
518 # Try to get client the local way first, fall back to inline for cloud execution
519 try:
520 client = get_letta_client()
521 except (ImportError, NameError):
522 # Create Letta client inline for cloud execution
523 import os
524 from letta_client import Letta
525 client = Letta(token=os.environ["LETTA_API_KEY"])
526 results = []
527
528 # Build mapping of block labels to IDs using the API
529 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
530 block_label_to_id = {}
531
532 for block in current_blocks:
533 block_label_to_id[block.label] = str(block.id)
534
535 # Process each user ID and detach atomically
536 for user_id in user_ids:
537 block_label = f"x_user_{user_id}"
538
539 if block_label in block_label_to_id:
540 try:
541 # Detach block atomically
542 client.agents.blocks.detach(
543 agent_id=str(agent_state.id),
544 block_id=block_label_to_id[block_label]
545 )
546 results.append(f"✓ {user_id}: Detached")
547 except Exception as e:
548 results.append(f"✗ {user_id}: Error during detachment - {str(e)}")
549 else:
550 results.append(f"✗ {user_id}: Not attached")
551
552 return f"X user detachment results:\n" + "\n".join(results)
553
554 except Exception as e:
555 raise Exception(f"Error detaching X user blocks: {str(e)}")
556
557
558def x_user_note_append(user_id: str, note: str, agent_state: "AgentState") -> str:
559 """
560 Append a note to an X user's memory block. Creates the block if it doesn't exist.
561
562 Args:
563 user_id: X user ID (e.g., '1232326955652931584')
564 note: Note to append to the user's memory block
565 agent_state: The agent state object containing agent information
566
567 Returns:
568 String confirming the note was appended
569 """
570 try:
571 # Create Letta client inline - cloud tools must be self-contained
572 import os
573 from letta_client import Letta
574 client = Letta(token=os.environ["LETTA_API_KEY"])
575
576 block_label = f"x_user_{user_id}"
577
578 # Check if block exists
579 blocks = client.blocks.list(label=block_label)
580
581 if blocks and len(blocks) > 0:
582 # Block exists, append to it
583 block = blocks[0]
584 current_value = block.value
585 new_value = current_value + note
586
587 # Update the block
588 client.blocks.modify(
589 block_id=str(block.id),
590 value=new_value
591 )
592 return f"✓ Appended note to X user {user_id}'s memory block"
593
594 else:
595 # Block doesn't exist, create it with the note
596 initial_value = f"# X User: {user_id}\n\n{note}"
597 block = client.blocks.create(
598 label=block_label,
599 value=initial_value,
600 limit=5000
601 )
602
603 # Check if block needs to be attached to agent
604 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
605 current_block_labels = {block.label for block in current_blocks}
606
607 if block_label not in current_block_labels:
608 # Attach the new block to the agent
609 client.agents.blocks.attach(
610 agent_id=str(agent_state.id),
611 block_id=str(block.id)
612 )
613 return f"✓ Created and attached X user {user_id}'s memory block with note"
614 else:
615 return f"✓ Created X user {user_id}'s memory block with note"
616
617 except Exception as e:
618 raise Exception(f"Error appending note to X user block: {str(e)}")
619
620
621def x_user_note_replace(user_id: str, old_text: str, new_text: str, agent_state: "AgentState") -> str:
622 """
623 Replace text in an X user's memory block.
624
625 Args:
626 user_id: X user ID (e.g., '1232326955652931584')
627 old_text: Text to find and replace
628 new_text: Text to replace the old_text with
629 agent_state: The agent state object containing agent information
630
631 Returns:
632 String confirming the text was replaced
633 """
634 try:
635 # Create Letta client inline - cloud tools must be self-contained
636 import os
637 from letta_client import Letta
638 client = Letta(token=os.environ["LETTA_API_KEY"])
639
640 block_label = f"x_user_{user_id}"
641
642 # Check if block exists
643 blocks = client.blocks.list(label=block_label)
644
645 if not blocks or len(blocks) == 0:
646 raise Exception(f"No memory block found for X user: {user_id}")
647
648 block = blocks[0]
649 current_value = block.value
650
651 # Check if old_text exists in the block
652 if old_text not in current_value:
653 raise Exception(f"Text '{old_text}' not found in X user {user_id}'s memory block")
654
655 # Replace the text
656 new_value = current_value.replace(old_text, new_text)
657
658 # Update the block
659 client.blocks.modify(
660 block_id=str(block.id),
661 value=new_value
662 )
663 return f"✓ Replaced text in X user {user_id}'s memory block"
664
665 except Exception as e:
666 raise Exception(f"Error replacing text in X user block: {str(e)}")
667
668
669def x_user_note_set(user_id: str, content: str, agent_state: "AgentState") -> str:
670 """
671 Set the complete content of an X user's memory block.
672
673 Args:
674 user_id: X user ID (e.g., '1232326955652931584')
675 content: Complete content to set for the memory block
676 agent_state: The agent state object containing agent information
677
678 Returns:
679 String confirming the content was set
680 """
681 try:
682 # Create Letta client inline - cloud tools must be self-contained
683 import os
684 from letta_client import Letta
685 client = Letta(token=os.environ["LETTA_API_KEY"])
686
687 block_label = f"x_user_{user_id}"
688
689 # Check if block exists
690 blocks = client.blocks.list(label=block_label)
691
692 if blocks and len(blocks) > 0:
693 # Block exists, update it
694 block = blocks[0]
695 client.blocks.modify(
696 block_id=str(block.id),
697 value=content
698 )
699 return f"✓ Set content for X user {user_id}'s memory block"
700
701 else:
702 # Block doesn't exist, create it
703 block = client.blocks.create(
704 label=block_label,
705 value=content,
706 limit=5000
707 )
708
709 # Check if block needs to be attached to agent
710 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
711 current_block_labels = {block.label for block in current_blocks}
712
713 if block_label not in current_block_labels:
714 # Attach the new block to the agent
715 client.agents.blocks.attach(
716 agent_id=str(agent_state.id),
717 block_id=str(block.id)
718 )
719 return f"✓ Created and attached X user {user_id}'s memory block"
720 else:
721 return f"✓ Created X user {user_id}'s memory block"
722
723 except Exception as e:
724 raise Exception(f"Error setting X user block content: {str(e)}")
725
726
727def x_user_note_view(user_id: str, agent_state: "AgentState") -> str:
728 """
729 View the content of an X user's memory block.
730
731 Args:
732 user_id: X user ID (e.g., '1232326955652931584')
733 agent_state: The agent state object containing agent information
734
735 Returns:
736 String containing the user's memory block content
737 """
738 try:
739 # Create Letta client inline - cloud tools must be self-contained
740 import os
741 from letta_client import Letta
742 client = Letta(token=os.environ["LETTA_API_KEY"])
743
744 block_label = f"x_user_{user_id}"
745
746 # Check if block exists
747 blocks = client.blocks.list(label=block_label)
748
749 if not blocks or len(blocks) == 0:
750 return f"No memory block found for X user: {user_id}"
751
752 block = blocks[0]
753
754 return f"Memory block for X user {user_id}:\n\n{block.value}"
755
756 except Exception as e:
757 raise Exception(f"Error viewing X user block: {str(e)}")
758
759