this repo has no description
1"""
2Bot detection tools for checking known_bots memory block.
3"""
4import os
5import random
6import logging
7from typing import List, Tuple, Optional
8from pydantic import BaseModel, Field
9from letta_client import Letta
10
11logger = logging.getLogger(__name__)
12
13
14class CheckKnownBotsArgs(BaseModel):
15 """Arguments for checking if users are in the known_bots list."""
16 handles: List[str] = Field(..., description="List of user handles to check against known_bots")
17
18
19def check_known_bots(handles: List[str], agent_state: "AgentState") -> str:
20 """
21 Check if any of the provided handles are in the known_bots memory block.
22
23 Args:
24 handles: List of user handles to check (e.g., ['horsedisc.bsky.social', 'user.bsky.social'])
25 agent_state: The agent state object containing agent information
26
27 Returns:
28 JSON string with bot detection results
29 """
30 import json
31
32 try:
33 # Create Letta client inline (for cloud execution)
34 client = Letta(token=os.environ["LETTA_API_KEY"])
35
36 # Get all blocks attached to the agent to check if known_bots is mounted
37 attached_blocks = client.agents.blocks.list(agent_id=str(agent_state.id))
38 attached_labels = {block.label for block in attached_blocks}
39
40 if "known_bots" not in attached_labels:
41 return json.dumps({
42 "error": "known_bots memory block is not mounted to this agent",
43 "bot_detected": False,
44 "detected_bots": []
45 })
46
47 # Retrieve known_bots block content using agent-specific retrieval
48 try:
49 known_bots_block = client.agents.blocks.retrieve(
50 agent_id=str(agent_state.id),
51 block_label="known_bots"
52 )
53 except Exception as e:
54 return json.dumps({
55 "error": f"Error retrieving known_bots block: {str(e)}",
56 "bot_detected": False,
57 "detected_bots": []
58 })
59 known_bots_content = known_bots_block.value
60
61 # Parse known bots from content
62 known_bot_handles = []
63 for line in known_bots_content.split('\n'):
64 line = line.strip()
65 if line and not line.startswith('#'):
66 # Extract handle from lines like "- @handle.bsky.social" or "- @handle.bsky.social: description"
67 if line.startswith('- @'):
68 handle = line[3:].split(':')[0].strip()
69 known_bot_handles.append(handle)
70 elif line.startswith('-'):
71 handle = line[1:].split(':')[0].strip().lstrip('@')
72 known_bot_handles.append(handle)
73
74 # Normalize handles for comparison
75 normalized_input_handles = [h.lstrip('@').strip() for h in handles]
76 normalized_bot_handles = [h.strip() for h in known_bot_handles]
77
78 # Check for matches
79 detected_bots = []
80 for handle in normalized_input_handles:
81 if handle in normalized_bot_handles:
82 detected_bots.append(handle)
83
84 bot_detected = len(detected_bots) > 0
85
86 return json.dumps({
87 "bot_detected": bot_detected,
88 "detected_bots": detected_bots,
89 "total_known_bots": len(normalized_bot_handles),
90 "checked_handles": normalized_input_handles
91 })
92
93 except Exception as e:
94 return json.dumps({
95 "error": f"Error checking known_bots: {str(e)}",
96 "bot_detected": False,
97 "detected_bots": []
98 })
99
100
101def should_respond_to_bot_thread() -> bool:
102 """
103 Determine if we should respond to a bot thread (10% chance).
104
105 Returns:
106 True if we should respond, False if we should skip
107 """
108 return random.random() < 0.1
109
110
111def extract_handles_from_thread(thread_data: dict) -> List[str]:
112 """
113 Extract all unique handles from a thread structure.
114
115 Args:
116 thread_data: Thread data dictionary from Bluesky API
117
118 Returns:
119 List of unique handles found in the thread
120 """
121 handles = set()
122
123 def extract_from_post(post):
124 """Recursively extract handles from a post and its replies."""
125 if isinstance(post, dict):
126 # Get author handle
127 if 'post' in post and 'author' in post['post']:
128 handle = post['post']['author'].get('handle')
129 if handle:
130 handles.add(handle)
131 elif 'author' in post:
132 handle = post['author'].get('handle')
133 if handle:
134 handles.add(handle)
135
136 # Check replies
137 if 'replies' in post:
138 for reply in post['replies']:
139 extract_from_post(reply)
140
141 # Check parent
142 if 'parent' in post:
143 extract_from_post(post['parent'])
144
145 # Start extraction from thread root
146 if 'thread' in thread_data:
147 extract_from_post(thread_data['thread'])
148 else:
149 extract_from_post(thread_data)
150
151 return list(handles)
152
153
154# Tool configuration for registration
155TOOL_CONFIG = {
156 "type": "function",
157 "function": {
158 "name": "check_known_bots",
159 "description": "Check if any of the provided handles are in the known_bots memory block",
160 "parameters": CheckKnownBotsArgs.model_json_schema(),
161 },
162}