a digital person for bluesky
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
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 results.append(f"✓ {handle}: Block attached")
73 logger.info(f"Successfully attached block {block_label} to agent")
74
75 except Exception as e:
76 results.append(f"✗ {handle}: Error - {str(e)}")
77 logger.error(f"Error processing block for {handle}: {e}")
78
79 return f"Attachment results:\n" + "\n".join(results)
80
81 except Exception as e:
82 logger.error(f"Error attaching user blocks: {e}")
83 raise Exception(f"Error attaching user blocks: {str(e)}")
84
85
86def detach_user_blocks(handles: list, agent_state: "AgentState") -> str:
87 """
88 Detach user-specific memory blocks from the agent. Blocks are preserved for later use.
89
90 Args:
91 handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])
92 agent_state: The agent state object containing agent information
93
94 Returns:
95 String with detachment results for each handle
96 """
97 import os
98 import logging
99 from letta_client import Letta
100
101 logger = logging.getLogger(__name__)
102
103 try:
104 client = Letta(token=os.environ["LETTA_API_KEY"])
105 results = []
106
107 # Build mapping of block labels to IDs using the API
108 current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
109 block_label_to_id = {}
110
111 for block in current_blocks:
112 block_label_to_id[block.label] = str(block.id)
113
114 # Process each handle and detach atomically
115 for handle in handles:
116 # Sanitize handle for block label - completely self-contained
117 clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_')
118 block_label = f"user_{clean_handle}"
119
120 if block_label in block_label_to_id:
121 try:
122 # Detach block atomically
123 client.agents.blocks.detach(
124 agent_id=str(agent_state.id),
125 block_id=block_label_to_id[block_label]
126 )
127 results.append(f"✓ {handle}: Detached")
128 logger.info(f"Successfully detached block {block_label} from agent")
129 except Exception as e:
130 results.append(f"✗ {handle}: Error during detachment - {str(e)}")
131 logger.error(f"Error detaching block {block_label}: {e}")
132 else:
133 results.append(f"✗ {handle}: Not attached")
134
135 return f"Detachment results:\n" + "\n".join(results)
136
137 except Exception as e:
138 logger.error(f"Error detaching user blocks: {e}")
139 raise Exception(f"Error detaching user blocks: {str(e)}")
140
141