cameron.stream
a digital person for bluesky
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
14
15def attach_user_blocks(handles: list, agent_state: "AgentState") -> str:
16 """
17 Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.
18
19 Args:
20 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
21 agent_state: The agent state object containing agent information
22
23 Returns:
24 String with attachment results for each handle
25 """
26 import os
27 import logging
28 from letta_client import Letta
29
30 logger = logging.getLogger(__name__)
31
32 try:
33 client = Letta(token=os.environ["LETTA_API_KEY"])
34 results = []
35
36 # Get current blocks using the API
37 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
38 current_block_labels = set()
39
40 for block in current_blocks:
41 current_block_labels.add(block.label)
42
43 for handle in handles:
44 # Sanitize handle for block label - completely self-contained
45 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
46 block_label = f"user_{clean_handle}"
47
48 # Skip if already attached
49 if block_label in current_block_labels:
50 results.append(f"✓ {handle}: Already attached")
51 continue
52
53 # Check if block exists or create new one
54 try:
55 blocks = client.blocks.list(label=block_label)
56 if blocks and len(blocks) > 0:
57 block = blocks[0]
58 logger.info(f"Found existing block: {block_label}")
59 else:
60 block = client.blocks.create(
61 label=block_label,
62 value=f"# User: {handle}\n\nNo information about this user yet.",
63 limit=5000
64 )
65 logger.info(f"Created new block: {block_label}")
66
67 # Attach block atomically
68 client.agents.blocks.attach(
69 agent_id=str(agent_state.id),
70 block_id=str(block.id)
71 )
72
73 # STOPGAP: Also update agent_state.memory to sync in-memory state
74 try:
75 agent_state.memory.set_block(block)
76 print(f"[SYNC] Successfully synced block {block_label} to agent_state.memory")
77 except Exception as sync_error:
78 print(f"[SYNC] Warning: Failed to sync block to agent_state.memory: {sync_error}")
79
80 results.append(f"✓ {handle}: Block attached")
81 logger.info(f"Successfully attached block {block_label} to agent")
82
83 except Exception as e:
84 results.append(f"✗ {handle}: Error - {str(e)}")
85 logger.error(f"Error processing block for {handle}: {e}")
86
87 return f"Attachment results:\n" + "\n".join(results)
88
89 except Exception as e:
90 logger.error(f"Error attaching user blocks: {e}")
91 raise Exception(f"Error attaching user blocks: {str(e)}")
92
93
94def detach_user_blocks(handles: list, agent_state: "AgentState") -> str:
95 """
96 Detach user-specific memory blocks from the agent. Blocks are preserved for later use.
97
98 Args:
99 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
100 agent_state: The agent state object containing agent information
101
102 Returns:
103 String with detachment results for each handle
104 """
105 import os
106 import logging
107 from letta_client import Letta
108
109 logger = logging.getLogger(__name__)
110
111 try:
112 client = Letta(token=os.environ["LETTA_API_KEY"])
113 results = []
114
115 # Build mapping of block labels to IDs using the API
116 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
117 block_label_to_id = {}
118
119 for block in current_blocks:
120 block_label_to_id[block.label] = str(block.id)
121
122 # Process each handle and detach atomically
123 for handle in handles:
124 # Sanitize handle for block label - completely self-contained
125 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
126 block_label = f"user_{clean_handle}"
127
128 if block_label in block_label_to_id:
129 try:
130 # Detach block atomically
131 client.agents.blocks.detach(
132 agent_id=str(agent_state.id),
133 block_id=block_label_to_id[block_label]
134 )
135 results.append(f"✓ {handle}: Detached")
136 logger.info(f"Successfully detached block {block_label} from agent")
137 except Exception as e:
138 results.append(f"✗ {handle}: Error during detachment - {str(e)}")
139 logger.error(f"Error detaching block {block_label}: {e}")
140 else:
141 results.append(f"✗ {handle}: Not attached")
142
143 return f"Detachment results:\n" + "\n".join(results)
144
145 except Exception as e:
146 logger.error(f"Error detaching user blocks: {e}")
147 raise Exception(f"Error detaching user blocks: {str(e)}")
148
149