Something went wrong. Try again.
forked from cameron.stream/void
Something went wrong. Try again.
Python
at contributing
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389"""Block management tools for user-specific memory blocks."""from pydantic import BaseModel, Fieldfrom typing import List, Dict, Anyimport logging
def get_letta_client(): """Get a Letta client using configuration.""" try: from config_loader import get_letta_config from letta_client import Letta config = get_letta_config() return Letta(token=config['api_key'], timeout=config['timeout']) except (ImportError, FileNotFoundError, KeyError): # Fallback to environment variable import os from letta_client import Letta return Letta(token=os.environ["LETTA_API_KEY"])
class AttachUserBlocksArgs(BaseModel): handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])")
class DetachUserBlocksArgs(BaseModel): handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])")
class UserNoteAppendArgs(BaseModel): handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')") note: str = Field(..., description="Note to append to the user's memory block (e.g., '\\n- Cameron is a person')")
class UserNoteReplaceArgs(BaseModel): handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')") old_text: str = Field(..., description="Text to find and replace in the user's memory block") new_text: str = Field(..., description="Text to replace the old_text with")
class UserNoteSetArgs(BaseModel): handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')") content: str = Field(..., description="Complete content to set for the user's memory block")
class UserNoteViewArgs(BaseModel): handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')")
def attach_user_blocks(handles: list, agent_state: "AgentState") -> str: """ Attach user-specific memory blocks to the agent. Creates blocks if they don't exist. Args: handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social']) agent_state: The agent state object containing agent information Returns: String with attachment results for each handle """ logger = logging.getLogger(__name__)
handles = list(set(handles)) try: client = get_letta_client() results = []
# Get current blocks using the API current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) current_block_labels = set() for block in current_blocks: current_block_labels.add(block.label)
for handle in handles: # Sanitize handle for block label - completely self-contained clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') block_label = f"user_{clean_handle}"
# Skip if already attached if block_label in current_block_labels: results.append(f"✓ {handle}: Already attached") continue
# Check if block exists or create new one try: blocks = client.blocks.list(label=block_label) if blocks and len(blocks) > 0: block = blocks[0] logger.debug(f"Found existing block: {block_label}") else: block = client.blocks.create( label=block_label, value=f"# User: {handle}\n\nNo information about this user yet.", limit=5000 ) logger.info(f"Created new block: {block_label}")
# Attach block atomically client.agents.blocks.attach( agent_id=str(agent_state.id), block_id=str(block.id) ) results.append(f"✓ {handle}: Block attached") logger.debug(f"Successfully attached block {block_label} to agent")
except Exception as e: results.append(f"✗ {handle}: Error - {str(e)}") logger.error(f"Error processing block for {handle}: {e}")
return f"Attachment results:\n" + "\n".join(results) except Exception as e: logger.error(f"Error attaching user blocks: {e}") raise Exception(f"Error attaching user blocks: {str(e)}")
def detach_user_blocks(handles: list, agent_state: "AgentState") -> str: """ Detach user-specific memory blocks from the agent. Blocks are preserved for later use. Args: handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social']) agent_state: The agent state object containing agent information Returns: String with detachment results for each handle """ logger = logging.getLogger(__name__) try: client = get_letta_client() results = []
# Build mapping of block labels to IDs using the API current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) block_label_to_id = {}
for block in current_blocks: block_label_to_id[block.label] = str(block.id)
# Process each handle and detach atomically for handle in handles: # Sanitize handle for block label - completely self-contained clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') block_label = f"user_{clean_handle}"
if block_label in block_label_to_id: try: # Detach block atomically client.agents.blocks.detach( agent_id=str(agent_state.id), block_id=block_label_to_id[block_label] ) results.append(f"✓ {handle}: Detached") logger.debug(f"Successfully detached block {block_label} from agent") except Exception as e: results.append(f"✗ {handle}: Error during detachment - {str(e)}") logger.error(f"Error detaching block {block_label}: {e}") else: results.append(f"✗ {handle}: Not attached")
return f"Detachment results:\n" + "\n".join(results) except Exception as e: logger.error(f"Error detaching user blocks: {e}") raise Exception(f"Error detaching user blocks: {str(e)}")
def user_note_append(handle: str, note: str, agent_state: "AgentState") -> str: """ Append a note to a user's memory block. Creates the block if it doesn't exist. Args: handle: User Bluesky handle (e.g., 'cameron.pfiffer.org') note: Note to append to the user's memory block agent_state: The agent state object containing agent information Returns: String confirming the note was appended """ logger = logging.getLogger(__name__) try: client = get_letta_client() # Sanitize handle for block label clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') block_label = f"user_{clean_handle}" # Check if block exists blocks = client.blocks.list(label=block_label) if blocks and len(blocks) > 0: # Block exists, append to it block = blocks[0] current_value = block.value new_value = current_value + note # Update the block client.blocks.modify( block_id=str(block.id), value=new_value ) logger.info(f"Appended note to existing block: {block_label}") return f"✓ Appended note to {handle}'s memory block" else: # Block doesn't exist, create it with the note initial_value = f"# User: {handle}\n\n{note}" block = client.blocks.create( label=block_label, value=initial_value, limit=5000 ) logger.info(f"Created new block with note: {block_label}") # Check if block needs to be attached to agent current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) current_block_labels = {block.label for block in current_blocks} if block_label not in current_block_labels: # Attach the new block to the agent client.agents.blocks.attach( agent_id=str(agent_state.id), block_id=str(block.id) ) logger.info(f"Attached new block to agent: {block_label}") return f"✓ Created and attached {handle}'s memory block with note" else: return f"✓ Created {handle}'s memory block with note" except Exception as e: logger.error(f"Error appending note to user block: {e}") raise Exception(f"Error appending note to user block: {str(e)}")
def user_note_replace(handle: str, old_text: str, new_text: str, agent_state: "AgentState") -> str: """ Replace text in a user's memory block. Args: handle: User Bluesky handle (e.g., 'cameron.pfiffer.org') old_text: Text to find and replace new_text: Text to replace the old_text with agent_state: The agent state object containing agent information Returns: String confirming the text was replaced """ logger = logging.getLogger(__name__) try: client = get_letta_client() # Sanitize handle for block label clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') block_label = f"user_{clean_handle}" # Check if block exists blocks = client.blocks.list(label=block_label) if not blocks or len(blocks) == 0: raise Exception(f"No memory block found for user: {handle}") block = blocks[0] current_value = block.value # Check if old_text exists in the block if old_text not in current_value: raise Exception(f"Text '{old_text}' not found in {handle}'s memory block") # Replace the text new_value = current_value.replace(old_text, new_text) # Update the block client.blocks.modify( block_id=str(block.id), value=new_value ) logger.info(f"Replaced text in block: {block_label}") return f"✓ Replaced text in {handle}'s memory block" except Exception as e: logger.error(f"Error replacing text in user block: {e}") raise Exception(f"Error replacing text in user block: {str(e)}")
def user_note_set(handle: str, content: str, agent_state: "AgentState") -> str: """ Set the complete content of a user's memory block. Args: handle: User Bluesky handle (e.g., 'cameron.pfiffer.org') content: Complete content to set for the memory block agent_state: The agent state object containing agent information Returns: String confirming the content was set """ logger = logging.getLogger(__name__) try: client = get_letta_client() # Sanitize handle for block label clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') block_label = f"user_{clean_handle}" # Check if block exists blocks = client.blocks.list(label=block_label) if blocks and len(blocks) > 0: # Block exists, update it block = blocks[0] client.blocks.modify( block_id=str(block.id), value=content ) logger.info(f"Set content for existing block: {block_label}") return f"✓ Set content for {handle}'s memory block" else: # Block doesn't exist, create it block = client.blocks.create( label=block_label, value=content, limit=5000 ) logger.info(f"Created new block with content: {block_label}") # Check if block needs to be attached to agent current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) current_block_labels = {block.label for block in current_blocks} if block_label not in current_block_labels: # Attach the new block to the agent client.agents.blocks.attach( agent_id=str(agent_state.id), block_id=str(block.id) ) logger.info(f"Attached new block to agent: {block_label}") return f"✓ Created and attached {handle}'s memory block" else: return f"✓ Created {handle}'s memory block" except Exception as e: logger.error(f"Error setting user block content: {e}") raise Exception(f"Error setting user block content: {str(e)}")
def user_note_view(handle: str, agent_state: "AgentState") -> str: """ View the content of a user's memory block. Args: handle: User Bluesky handle (e.g., 'cameron.pfiffer.org') agent_state: The agent state object containing agent information Returns: String containing the user's memory block content """ logger = logging.getLogger(__name__) try: client = get_letta_client() # Sanitize handle for block label clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') block_label = f"user_{clean_handle}" # Check if block exists blocks = client.blocks.list(label=block_label) if not blocks or len(blocks) == 0: return f"No memory block found for user: {handle}" block = blocks[0] logger.info(f"Retrieved content for block: {block_label}") return f"Memory block for {handle}:\n\n{block.value}" except Exception as e: logger.error(f"Error viewing user block: {e}") raise Exception(f"Error viewing user block: {str(e)}")