Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760"""Block management tools for user-specific memory blocks."""from pydantic import BaseModel, Fieldfrom typing import List, Dict, Any
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')")
# X (Twitter) User Block Managementclass AttachXUserBlocksArgs(BaseModel): user_ids: List[str] = Field(..., description="List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])")
class DetachXUserBlocksArgs(BaseModel): user_ids: List[str] = Field(..., description="List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])")
class XUserNoteAppendArgs(BaseModel): user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") note: str = Field(..., description="Note to append to the user's memory block (e.g., '\\\\n- Cameron is a person')")
class XUserNoteReplaceArgs(BaseModel): user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") 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 XUserNoteSetArgs(BaseModel): user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") content: str = Field(..., description="Complete content to set for the user's memory block")
class XUserNoteViewArgs(BaseModel): user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')")
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 """
handles = list(set(handles)) try: # Try to get client the local way first, fall back to inline for cloud execution try: client = get_letta_client() except (ImportError, NameError): # Create Letta client inline for cloud execution import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) results = []
# Get current blocks using the API current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) current_block_labels = set() current_block_ids = set() for block in current_blocks: current_block_labels.add(block.label) current_block_ids.add(str(block.id))
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] # Double-check if this block is already attached by ID if str(block.id) in current_block_ids: results.append(f"✓ {handle}: Already attached (by ID)") continue else: block = client.blocks.create( label=block_label, value=f"# User: {handle}\n\nNo information about this user yet.", limit=5000 )
# Attach block atomically try: client.agents.blocks.attach( agent_id=str(agent_state.id), block_id=str(block.id) ) results.append(f"✓ {handle}: Block attached") except Exception as attach_error: # Check if it's a duplicate constraint error error_str = str(attach_error) if "duplicate key value violates unique constraint" in error_str and "unique_label_per_agent" in error_str: # Block is already attached, possibly with this exact label results.append(f"✓ {handle}: Already attached (verified)") else: # Re-raise other errors raise attach_error
except Exception as e: results.append(f"✗ {handle}: Error - {str(e)}")
return f"Attachment results:\n" + "\n".join(results) except Exception as 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 """ try: # Try to get client the local way first, fall back to inline for cloud execution try: client = get_letta_client() except (ImportError, NameError): # Create Letta client inline for cloud execution import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) 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") except Exception as e: results.append(f"✗ {handle}: Error during detachment - {str(e)}") else: results.append(f"✗ {handle}: Not attached")
return f"Detachment results:\n" + "\n".join(results) except Exception as 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) # 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 ) 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 ) # 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) ) 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: 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) # 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 ) return f"✓ Replaced text in {handle}'s memory block" except Exception as 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) # 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 ) 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 ) # 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) ) return f"✓ Created and attached {handle}'s memory block" else: return f"✓ Created {handle}'s memory block" except Exception as 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) # 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] return f"Memory block for {handle}:\n\n{block.value}" except Exception as e: raise Exception(f"Error viewing user block: {str(e)}")
# X (Twitter) User Block Management Functions
def attach_x_user_blocks(user_ids: list, agent_state: "AgentState") -> str: """ Attach X user-specific memory blocks to the agent. Creates blocks if they don't exist. Args: user_ids: List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592']) agent_state: The agent state object containing agent information Returns: String with attachment results for each user ID """
user_ids = list(set(user_ids)) try: # Try to get client the local way first, fall back to inline for cloud execution try: client = get_letta_client() except (ImportError, NameError): # Create Letta client inline for cloud execution import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) 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 user_id in user_ids: # Create block label with x_user_ prefix block_label = f"x_user_{user_id}"
# Skip if already attached if block_label in current_block_labels: results.append(f"✓ {user_id}: 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] else: block = client.blocks.create( label=block_label, value=f"# X User: {user_id}\n\nNo information about this user yet.", limit=5000 )
# Attach block atomically client.agents.blocks.attach( agent_id=str(agent_state.id), block_id=str(block.id) ) results.append(f"✓ {user_id}: Block attached")
except Exception as e: results.append(f"✗ {user_id}: Error - {str(e)}")
return f"X user attachment results:\n" + "\n".join(results) except Exception as e: raise Exception(f"Error attaching X user blocks: {str(e)}")
def detach_x_user_blocks(user_ids: list, agent_state: "AgentState") -> str: """ Detach X user-specific memory blocks from the agent. Blocks are preserved for later use. Args: user_ids: List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592']) agent_state: The agent state object containing agent information Returns: String with detachment results for each user ID """ try: # Try to get client the local way first, fall back to inline for cloud execution try: client = get_letta_client() except (ImportError, NameError): # Create Letta client inline for cloud execution import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) 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 user ID and detach atomically for user_id in user_ids: block_label = f"x_user_{user_id}"
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"✓ {user_id}: Detached") except Exception as e: results.append(f"✗ {user_id}: Error during detachment - {str(e)}") else: results.append(f"✗ {user_id}: Not attached")
return f"X user detachment results:\n" + "\n".join(results) except Exception as e: raise Exception(f"Error detaching X user blocks: {str(e)}")
def x_user_note_append(user_id: str, note: str, agent_state: "AgentState") -> str: """ Append a note to an X user's memory block. Creates the block if it doesn't exist. Args: user_id: X user ID (e.g., '1232326955652931584') 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) block_label = f"x_user_{user_id}" # 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 ) return f"✓ Appended note to X user {user_id}'s memory block" else: # Block doesn't exist, create it with the note initial_value = f"# X User: {user_id}\n\n{note}" block = client.blocks.create( label=block_label, value=initial_value, limit=5000 ) # 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) ) return f"✓ Created and attached X user {user_id}'s memory block with note" else: return f"✓ Created X user {user_id}'s memory block with note" except Exception as e: raise Exception(f"Error appending note to X user block: {str(e)}")
def x_user_note_replace(user_id: str, old_text: str, new_text: str, agent_state: "AgentState") -> str: """ Replace text in an X user's memory block. Args: user_id: X user ID (e.g., '1232326955652931584') 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) block_label = f"x_user_{user_id}" # 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 X user: {user_id}") 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 X user {user_id}'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 ) return f"✓ Replaced text in X user {user_id}'s memory block" except Exception as e: raise Exception(f"Error replacing text in X user block: {str(e)}")
def x_user_note_set(user_id: str, content: str, agent_state: "AgentState") -> str: """ Set the complete content of an X user's memory block. Args: user_id: X user ID (e.g., '1232326955652931584') 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 """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) block_label = f"x_user_{user_id}" # 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 ) return f"✓ Set content for X user {user_id}'s memory block" else: # Block doesn't exist, create it block = client.blocks.create( label=block_label, value=content, limit=5000 ) # 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) ) return f"✓ Created and attached X user {user_id}'s memory block" else: return f"✓ Created X user {user_id}'s memory block" except Exception as e: raise Exception(f"Error setting X user block content: {str(e)}")
def x_user_note_view(user_id: str, agent_state: "AgentState") -> str: """ View the content of an X user's memory block. Args: user_id: X user ID (e.g., '1232326955652931584') agent_state: The agent state object containing agent information Returns: String containing the user's memory block content """ try: # Create Letta client inline - cloud tools must be self-contained import os from letta_client import Letta client = Letta(token=os.environ["LETTA_API_KEY"]) block_label = f"x_user_{user_id}" # 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 X user: {user_id}" block = blocks[0] return f"Memory block for X user {user_id}:\n\n{block.value}" except Exception as e: raise Exception(f"Error viewing X user block: {str(e)}")