bunware.org
forked from
cameron.stream/void
this repo has no description
cameron.stream
1"""Block management tools for user-specific memory blocks."""
2from pydantic import BaseModel, Field
3from typing import List, Dict, Any
4
5
6class AttachUserBlocksArgs(BaseModel):
7 handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])")
8
9
10class DetachUserBlocksArgs(BaseModel):
11 handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])")
12
13
14class UserNoteAppendArgs(BaseModel):
15 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
16 note: str = Field(..., description="Note to append to the user's memory block (e.g., '\\n- Cameron is a person')")
17
18
19class UserNoteReplaceArgs(BaseModel):
20 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
21 old_text: str = Field(..., description="Text to find and replace in the user's memory block")
22 new_text: str = Field(..., description="Text to replace the old_text with")
23
24
25class UserNoteSetArgs(BaseModel):
26 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
27 content: str = Field(..., description="Complete content to set for the user's memory block")
28
29
30class UserNoteViewArgs(BaseModel):
31 handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
32
33
34
35def attach_user_blocks(handles: list, agent_state: "AgentState") -> str:
36 """
37 Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.
38
39 Args:
40 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
41 agent_state: The agent state object containing agent information
42
43 Returns:
44 String with attachment results for each handle
45 """
46 import os
47 import logging
48 from letta_client import Letta
49
50 logger = logging.getLogger(__name__)
51
52 try:
53 client = Letta(token=os.environ["LETTA_API_KEY"])
54 results = []
55
56 # Get current blocks using the API
57 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
58 current_block_labels = set()
59
60 for block in current_blocks:
61 current_block_labels.add(block.label)
62
63 for handle in handles:
64 # Sanitize handle for block label - completely self-contained
65 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
66 block_label = f"user_{clean_handle}"
67
68 # Skip if already attached
69 if block_label in current_block_labels:
70 results.append(f"✓ {handle}: Already attached")
71 continue
72
73 # Check if block exists or create new one
74 try:
75 blocks = client.blocks.list(label=block_label)
76 if blocks and len(blocks) > 0:
77 block = blocks[0]
78 logger.debug(f"Found existing block: {block_label}")
79 else:
80 block = client.blocks.create(
81 label=block_label,
82 value=f"# User: {handle}\n\nNo information about this user yet.",
83 limit=5000
84 )
85 logger.info(f"Created new block: {block_label}")
86
87 # Attach block atomically
88 client.agents.blocks.attach(
89 agent_id=str(agent_state.id),
90 block_id=str(block.id)
91 )
92
93 results.append(f"✓ {handle}: Block attached")
94 logger.debug(f"Successfully attached block {block_label} to agent")
95
96 except Exception as e:
97 results.append(f"✗ {handle}: Error - {str(e)}")
98 logger.error(f"Error processing block for {handle}: {e}")
99
100 return f"Attachment results:\n" + "\n".join(results)
101
102 except Exception as e:
103 logger.error(f"Error attaching user blocks: {e}")
104 raise Exception(f"Error attaching user blocks: {str(e)}")
105
106
107def detach_user_blocks(handles: list, agent_state: "AgentState") -> str:
108 """
109 Detach user-specific memory blocks from the agent. Blocks are preserved for later use.
110
111 Args:
112 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
113 agent_state: The agent state object containing agent information
114
115 Returns:
116 String with detachment results for each handle
117 """
118 import os
119 import logging
120 from letta_client import Letta
121
122 logger = logging.getLogger(__name__)
123
124 try:
125 client = Letta(token=os.environ["LETTA_API_KEY"])
126 results = []
127
128 # Build mapping of block labels to IDs using the API
129 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
130 block_label_to_id = {}
131
132 for block in current_blocks:
133 block_label_to_id[block.label] = str(block.id)
134
135 # Process each handle and detach atomically
136 for handle in handles:
137 # Sanitize handle for block label - completely self-contained
138 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
139 block_label = f"user_{clean_handle}"
140
141 if block_label in block_label_to_id:
142 try:
143 # Detach block atomically
144 client.agents.blocks.detach(
145 agent_id=str(agent_state.id),
146 block_id=block_label_to_id[block_label]
147 )
148 results.append(f"✓ {handle}: Detached")
149 logger.debug(f"Successfully detached block {block_label} from agent")
150 except Exception as e:
151 results.append(f"✗ {handle}: Error during detachment - {str(e)}")
152 logger.error(f"Error detaching block {block_label}: {e}")
153 else:
154 results.append(f"✗ {handle}: Not attached")
155
156 return f"Detachment results:\n" + "\n".join(results)
157
158 except Exception as e:
159 logger.error(f"Error detaching user blocks: {e}")
160 raise Exception(f"Error detaching user blocks: {str(e)}")
161
162
163def user_note_append(handle: str, note: str, agent_state: "AgentState") -> str:
164 """
165 Append a note to a user's memory block. Creates the block if it doesn't exist.
166
167 Args:
168 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
169 note: Note to append to the user's memory block
170 agent_state: The agent state object containing agent information
171
172 Returns:
173 String confirming the note was appended
174 """
175 import os
176 import logging
177 from letta_client import Letta
178
179 logger = logging.getLogger(__name__)
180
181 try:
182 client = Letta(token=os.environ["LETTA_API_KEY"])
183
184 # Sanitize handle for block label
185 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
186 block_label = f"user_{clean_handle}"
187
188 # Check if block exists
189 blocks = client.blocks.list(label=block_label)
190
191 if blocks and len(blocks) > 0:
192 # Block exists, append to it
193 block = blocks[0]
194 current_value = block.value
195 new_value = current_value + note
196
197 # Update the block
198 client.blocks.modify(
199 block_id=str(block.id),
200 value=new_value
201 )
202 logger.info(f"Appended note to existing block: {block_label}")
203 return f"✓ Appended note to {handle}'s memory block"
204
205 else:
206 # Block doesn't exist, create it with the note
207 initial_value = f"# User: {handle}\n\n{note}"
208 block = client.blocks.create(
209 label=block_label,
210 value=initial_value,
211 limit=5000
212 )
213 logger.info(f"Created new block with note: {block_label}")
214
215 # Check if block needs to be attached to agent
216 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
217 current_block_labels = {block.label for block in current_blocks}
218
219 if block_label not in current_block_labels:
220 # Attach the new block to the agent
221 client.agents.blocks.attach(
222 agent_id=str(agent_state.id),
223 block_id=str(block.id)
224 )
225 logger.info(f"Attached new block to agent: {block_label}")
226 return f"✓ Created and attached {handle}'s memory block with note"
227 else:
228 return f"✓ Created {handle}'s memory block with note"
229
230 except Exception as e:
231 logger.error(f"Error appending note to user block: {e}")
232 raise Exception(f"Error appending note to user block: {str(e)}")
233
234
235def user_note_replace(handle: str, old_text: str, new_text: str, agent_state: "AgentState") -> str:
236 """
237 Replace text in a user's memory block.
238
239 Args:
240 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
241 old_text: Text to find and replace
242 new_text: Text to replace the old_text with
243 agent_state: The agent state object containing agent information
244
245 Returns:
246 String confirming the text was replaced
247 """
248 import os
249 import logging
250 from letta_client import Letta
251
252 logger = logging.getLogger(__name__)
253
254 try:
255 client = Letta(token=os.environ["LETTA_API_KEY"])
256
257 # Sanitize handle for block label
258 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
259 block_label = f"user_{clean_handle}"
260
261 # Check if block exists
262 blocks = client.blocks.list(label=block_label)
263
264 if not blocks or len(blocks) == 0:
265 raise Exception(f"No memory block found for user: {handle}")
266
267 block = blocks[0]
268 current_value = block.value
269
270 # Check if old_text exists in the block
271 if old_text not in current_value:
272 raise Exception(f"Text '{old_text}' not found in {handle}'s memory block")
273
274 # Replace the text
275 new_value = current_value.replace(old_text, new_text)
276
277 # Update the block
278 client.blocks.modify(
279 block_id=str(block.id),
280 value=new_value
281 )
282 logger.info(f"Replaced text in block: {block_label}")
283 return f"✓ Replaced text in {handle}'s memory block"
284
285 except Exception as e:
286 logger.error(f"Error replacing text in user block: {e}")
287 raise Exception(f"Error replacing text in user block: {str(e)}")
288
289
290def user_note_set(handle: str, content: str, agent_state: "AgentState") -> str:
291 """
292 Set the complete content of a user's memory block.
293
294 Args:
295 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
296 content: Complete content to set for the memory block
297 agent_state: The agent state object containing agent information
298
299 Returns:
300 String confirming the content was set
301 """
302 import os
303 import logging
304 from letta_client import Letta
305
306 logger = logging.getLogger(__name__)
307
308 try:
309 client = Letta(token=os.environ["LETTA_API_KEY"])
310
311 # Sanitize handle for block label
312 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
313 block_label = f"user_{clean_handle}"
314
315 # Check if block exists
316 blocks = client.blocks.list(label=block_label)
317
318 if blocks and len(blocks) > 0:
319 # Block exists, update it
320 block = blocks[0]
321 client.blocks.modify(
322 block_id=str(block.id),
323 value=content
324 )
325 logger.info(f"Set content for existing block: {block_label}")
326 return f"✓ Set content for {handle}'s memory block"
327
328 else:
329 # Block doesn't exist, create it
330 block = client.blocks.create(
331 label=block_label,
332 value=content,
333 limit=5000
334 )
335 logger.info(f"Created new block with content: {block_label}")
336
337 # Check if block needs to be attached to agent
338 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
339 current_block_labels = {block.label for block in current_blocks}
340
341 if block_label not in current_block_labels:
342 # Attach the new block to the agent
343 client.agents.blocks.attach(
344 agent_id=str(agent_state.id),
345 block_id=str(block.id)
346 )
347 logger.info(f"Attached new block to agent: {block_label}")
348 return f"✓ Created and attached {handle}'s memory block"
349 else:
350 return f"✓ Created {handle}'s memory block"
351
352 except Exception as e:
353 logger.error(f"Error setting user block content: {e}")
354 raise Exception(f"Error setting user block content: {str(e)}")
355
356
357def user_note_view(handle: str, agent_state: "AgentState") -> str:
358 """
359 View the content of a user's memory block.
360
361 Args:
362 handle: User Bluesky handle (e.g., 'cameron.pfiffer.org')
363 agent_state: The agent state object containing agent information
364
365 Returns:
366 String containing the user's memory block content
367 """
368 import os
369 import logging
370 from letta_client import Letta
371
372 logger = logging.getLogger(__name__)
373
374 try:
375 client = Letta(token=os.environ["LETTA_API_KEY"])
376
377 # Sanitize handle for block label
378 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
379 block_label = f"user_{clean_handle}"
380
381 # Check if block exists
382 blocks = client.blocks.list(label=block_label)
383
384 if not blocks or len(blocks) == 0:
385 return f"No memory block found for user: {handle}"
386
387 block = blocks[0]
388 logger.info(f"Retrieved content for block: {block_label}")
389
390 return f"Memory block for {handle}:\n\n{block.value}"
391
392 except Exception as e:
393 logger.error(f"Error viewing user block: {e}")
394 raise Exception(f"Error viewing user block: {str(e)}")
395
396