+1
.gitignore
+1
.gitignore
+31
CLAUDE.md
+31
CLAUDE.md
···
193
PDS_URI=https://bsky.social # Optional, defaults to bsky.social
194
```
195
196
+
### X Bot Configuration
197
+
198
+
The X bot uses a separate configuration file `x_config.yaml` with the following structure:
199
+
```yaml
200
+
x:
201
+
api_key: your_x_bearer_token
202
+
consumer_key: your_consumer_key
203
+
consumer_secret: your_consumer_secret
204
+
access_token: your_access_token
205
+
access_token_secret: your_access_token_secret
206
+
user_id: "your_user_id"
207
+
208
+
letta:
209
+
api_key: your_letta_api_key
210
+
project_id: your_project_id
211
+
timeout: 600
212
+
agent_id: your_agent_id
213
+
214
+
bot:
215
+
cleanup_interval: 10
216
+
max_thread_depth: 50
217
+
rate_limit_delay: 1
218
+
downrank_response_rate: 0.1
219
+
220
+
logging:
221
+
level: INFO
222
+
enable_debug_data: true
223
+
log_thread_context: true
224
+
log_agent_responses: true
225
+
```
226
+
227
## Key Development Patterns
228
229
1. **Tool System**: Tools are defined as standalone functions in `tools/functions.py` with Pydantic schemas for validation, registered via `register_tools.py`
+44
-35
x.py
+44
-35
x.py
···
495
logger.error("Failed to post tweet")
496
return None
497
498
-
def load_x_config(config_path: str = "config.yaml") -> Dict[str, str]:
499
-
"""Load X configuration from config file."""
500
try:
501
with open(config_path, 'r') as f:
502
config = yaml.safe_load(f)
503
-
504
x_config = config.get('x', {})
505
if not x_config.get('api_key') or not x_config.get('user_id'):
506
-
raise ValueError("X API key and user_id must be configured in config.yaml")
507
-
508
-
return x_config
509
except Exception as e:
510
logger.error(f"Failed to load X configuration: {e}")
511
raise
512
513
-
def create_x_client(config_path: str = "config.yaml") -> XClient:
514
"""Create and return an X client with configuration loaded from file."""
515
config = load_x_config(config_path)
516
return XClient(
517
-
api_key=config['api_key'],
518
-
user_id=config['user_id'],
519
-
access_token=config.get('access_token'),
520
-
consumer_key=config.get('consumer_key'),
521
-
consumer_secret=config.get('consumer_secret'),
522
-
access_token_secret=config.get('access_token_secret')
523
)
524
525
def mention_to_yaml_string(mention: Dict, users_data: Optional[Dict] = None) -> str:
···
605
606
try:
607
from tools.blocks import attach_x_user_blocks, x_user_note_set
608
-
from config_loader import get_letta_config
609
from letta_client import Letta
610
-
611
-
# Get Letta client and agent_id from config
612
-
config = get_letta_config()
613
client = Letta(token=config['api_key'], timeout=config['timeout'])
614
615
# Use provided agent_id or get from config
···
1011
print(f" Username: @{user_data.get('username')}")
1012
print(f" Name: {user_data.get('name')}")
1013
print(f" Description: {user_data.get('description', 'N/A')[:100]}...")
1014
-
print(f"\n🔧 Update your config.yaml with:")
1015
print(f" user_id: \"{user_data.get('id')}\"")
1016
return user_data
1017
else:
···
1143
import json
1144
import yaml
1145
1146
-
# Load full config to access letta section
1147
try:
1148
-
with open("config.yaml", 'r') as f:
1149
-
full_config = yaml.safe_load(f)
1150
-
1151
-
letta_config = full_config.get('letta', {})
1152
api_key = letta_config.get('api_key')
1153
config_agent_id = letta_config.get('agent_id')
1154
···
1158
agent_id = config_agent_id
1159
print(f"ℹ️ Using agent_id from config: {agent_id}")
1160
else:
1161
-
print("❌ No agent_id found in config.yaml")
1162
print("Expected config structure:")
1163
print(" letta:")
1164
print(" agent_id: your-agent-id")
···
1171
import os
1172
api_key = os.getenv('LETTA_API_KEY')
1173
if not api_key:
1174
-
print("❌ LETTA_API_KEY not found in config.yaml or environment")
1175
print("Expected config structure:")
1176
print(" letta:")
1177
print(" api_key: your-letta-api-key")
···
1179
else:
1180
print("ℹ️ Using LETTA_API_KEY from environment")
1181
else:
1182
-
print("ℹ️ Using LETTA_API_KEY from config.yaml")
1183
1184
except Exception as e:
1185
print(f"❌ Error loading config: {e}")
···
1548
print(f" {line}")
1549
1550
# Send to Letta agent
1551
-
from config_loader import get_letta_config
1552
from letta_client import Letta
1553
-
1554
-
config = get_letta_config()
1555
letta_client = Letta(token=config['api_key'], timeout=config['timeout'])
1556
1557
prompt_char_count = len(prompt)
···
2001
"""Initialize the void agent for X operations."""
2002
logger.info("Starting void agent initialization for X...")
2003
2004
-
from config_loader import get_letta_config
2005
from letta_client import Letta
2006
-
2007
# Get config
2008
-
config = get_letta_config()
2009
client = Letta(token=config['api_key'], timeout=config['timeout'])
2010
agent_id = config['agent_id']
2011
···
2046
"""
2047
import time
2048
from time import sleep
2049
-
from config_loader import get_letta_config
2050
from letta_client import Letta
2051
-
2052
logger.info("=== STARTING X VOID BOT ===")
2053
2054
# Initialize void agent
···
2060
logger.info("Connected to X API")
2061
2062
# Get Letta client for periodic cleanup
2063
-
config = get_letta_config()
2064
letta_client = Letta(token=config['api_key'], timeout=config['timeout'])
2065
2066
# Main loop
···
495
logger.error("Failed to post tweet")
496
return None
497
498
+
def load_x_config(config_path: str = "x_config.yaml") -> Dict[str, Any]:
499
+
"""Load complete X configuration from x_config.yaml."""
500
try:
501
with open(config_path, 'r') as f:
502
config = yaml.safe_load(f)
503
+
504
+
if not config:
505
+
raise ValueError(f"Empty or invalid configuration file: {config_path}")
506
+
507
+
# Validate required sections
508
x_config = config.get('x', {})
509
+
letta_config = config.get('letta', {})
510
+
511
if not x_config.get('api_key') or not x_config.get('user_id'):
512
+
raise ValueError("X API key and user_id must be configured in x_config.yaml")
513
+
514
+
if not letta_config.get('api_key') or not letta_config.get('agent_id'):
515
+
raise ValueError("Letta API key and agent_id must be configured in x_config.yaml")
516
+
517
+
return config
518
except Exception as e:
519
logger.error(f"Failed to load X configuration: {e}")
520
raise
521
522
+
def get_x_letta_config(config_path: str = "x_config.yaml") -> Dict[str, Any]:
523
+
"""Get Letta configuration from X config file."""
524
+
config = load_x_config(config_path)
525
+
return config['letta']
526
+
527
+
def create_x_client(config_path: str = "x_config.yaml") -> XClient:
528
"""Create and return an X client with configuration loaded from file."""
529
config = load_x_config(config_path)
530
+
x_config = config['x']
531
return XClient(
532
+
api_key=x_config['api_key'],
533
+
user_id=x_config['user_id'],
534
+
access_token=x_config.get('access_token'),
535
+
consumer_key=x_config.get('consumer_key'),
536
+
consumer_secret=x_config.get('consumer_secret'),
537
+
access_token_secret=x_config.get('access_token_secret')
538
)
539
540
def mention_to_yaml_string(mention: Dict, users_data: Optional[Dict] = None) -> str:
···
620
621
try:
622
from tools.blocks import attach_x_user_blocks, x_user_note_set
623
from letta_client import Letta
624
+
625
+
# Get Letta client and agent_id from X config
626
+
config = get_x_letta_config()
627
client = Letta(token=config['api_key'], timeout=config['timeout'])
628
629
# Use provided agent_id or get from config
···
1025
print(f" Username: @{user_data.get('username')}")
1026
print(f" Name: {user_data.get('name')}")
1027
print(f" Description: {user_data.get('description', 'N/A')[:100]}...")
1028
+
print(f"\n🔧 Update your x_config.yaml with:")
1029
print(f" user_id: \"{user_data.get('id')}\"")
1030
return user_data
1031
else:
···
1157
import json
1158
import yaml
1159
1160
+
# Load X config to access letta section
1161
try:
1162
+
x_config = load_x_config()
1163
+
letta_config = x_config.get('letta', {})
1164
api_key = letta_config.get('api_key')
1165
config_agent_id = letta_config.get('agent_id')
1166
···
1170
agent_id = config_agent_id
1171
print(f"ℹ️ Using agent_id from config: {agent_id}")
1172
else:
1173
+
print("❌ No agent_id found in x_config.yaml")
1174
print("Expected config structure:")
1175
print(" letta:")
1176
print(" agent_id: your-agent-id")
···
1183
import os
1184
api_key = os.getenv('LETTA_API_KEY')
1185
if not api_key:
1186
+
print("❌ LETTA_API_KEY not found in x_config.yaml or environment")
1187
print("Expected config structure:")
1188
print(" letta:")
1189
print(" api_key: your-letta-api-key")
···
1191
else:
1192
print("ℹ️ Using LETTA_API_KEY from environment")
1193
else:
1194
+
print("ℹ️ Using LETTA_API_KEY from x_config.yaml")
1195
1196
except Exception as e:
1197
print(f"❌ Error loading config: {e}")
···
1560
print(f" {line}")
1561
1562
# Send to Letta agent
1563
from letta_client import Letta
1564
+
1565
+
config = get_x_letta_config()
1566
letta_client = Letta(token=config['api_key'], timeout=config['timeout'])
1567
1568
prompt_char_count = len(prompt)
···
2012
"""Initialize the void agent for X operations."""
2013
logger.info("Starting void agent initialization for X...")
2014
2015
from letta_client import Letta
2016
+
2017
# Get config
2018
+
config = get_x_letta_config()
2019
client = Letta(token=config['api_key'], timeout=config['timeout'])
2020
agent_id = config['agent_id']
2021
···
2056
"""
2057
import time
2058
from time import sleep
2059
from letta_client import Letta
2060
+
2061
logger.info("=== STARTING X VOID BOT ===")
2062
2063
# Initialize void agent
···
2069
logger.info("Connected to X API")
2070
2071
# Get Letta client for periodic cleanup
2072
+
config = get_x_letta_config()
2073
letta_client = Letta(token=config['api_key'], timeout=config['timeout'])
2074
2075
# Main loop