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