cameron.stream
-44
register_x_tools.py
-44
register_x_tools.py
···
8
8
from config_loader import get_letta_config
9
9
10
10
# Import standalone functions and their schemas
11
-
from tools.blocks import (
12
-
attach_x_user_blocks, detach_x_user_blocks,
13
-
x_user_note_append, x_user_note_replace, x_user_note_set, x_user_note_view,
14
-
AttachXUserBlocksArgs, DetachXUserBlocksArgs,
15
-
XUserNoteAppendArgs, XUserNoteReplaceArgs, XUserNoteSetArgs, XUserNoteViewArgs
16
-
)
17
11
from tools.halt import halt_activity, HaltArgs
18
12
from tools.ignore import ignore_notification, IgnoreNotificationArgs
19
13
from tools.whitewind import create_whitewind_blog_post, WhitewindPostArgs
···
67
61
"args_schema": WebpageArgs,
68
62
"description": "Fetch a webpage and convert it to markdown/text format using Jina AI reader",
69
63
"tags": ["web", "fetch", "webpage", "markdown", "jina"]
70
-
},
71
-
72
-
# X user block management tools
73
-
{
74
-
"func": attach_x_user_blocks,
75
-
"args_schema": AttachXUserBlocksArgs,
76
-
"description": "Attach X user-specific memory blocks to the agent. Creates blocks if they don't exist.",
77
-
"tags": ["memory", "blocks", "user", "x", "twitter"]
78
-
},
79
-
{
80
-
"func": detach_x_user_blocks,
81
-
"args_schema": DetachXUserBlocksArgs,
82
-
"description": "Detach X user-specific memory blocks from the agent. Blocks are preserved for later use.",
83
-
"tags": ["memory", "blocks", "user", "x", "twitter"]
84
-
},
85
-
{
86
-
"func": x_user_note_append,
87
-
"args_schema": XUserNoteAppendArgs,
88
-
"description": "Append a note to an X user's memory block. Creates the block if it doesn't exist.",
89
-
"tags": ["memory", "blocks", "user", "append", "x", "twitter"]
90
-
},
91
-
{
92
-
"func": x_user_note_replace,
93
-
"args_schema": XUserNoteReplaceArgs,
94
-
"description": "Replace text in an X user's memory block.",
95
-
"tags": ["memory", "blocks", "user", "replace", "x", "twitter"]
96
-
},
97
-
{
98
-
"func": x_user_note_set,
99
-
"args_schema": XUserNoteSetArgs,
100
-
"description": "Set the complete content of an X user's memory block.",
101
-
"tags": ["memory", "blocks", "user", "set", "x", "twitter"]
102
-
},
103
-
{
104
-
"func": x_user_note_view,
105
-
"args_schema": XUserNoteViewArgs,
106
-
"description": "View the content of an X user's memory block.",
107
-
"tags": ["memory", "blocks", "user", "view", "x", "twitter"]
108
64
},
109
65
110
66
# X thread tool
+4
-10
tools/x_post.py
+4
-10
tools/x_post.py
···
1
1
"""Tool for creating standalone posts on X (Twitter)."""
2
2
import os
3
3
import json
4
-
import logging
5
4
import requests
6
5
from typing import Optional
7
6
from pydantic import BaseModel, Field, validator
8
-
from requests_oauthlib import OAuth1
9
-
10
-
logger = logging.getLogger(__name__)
11
-
12
7
13
8
class PostToXArgs(BaseModel):
14
9
text: str = Field(
···
39
34
Raises:
40
35
Exception: If text exceeds character limit or posting fails
41
36
"""
37
+
import requests
38
+
from requests_oauthlib import OAuth1
39
+
42
40
# Validate input
43
41
if len(text) > 280:
44
42
raise Exception(f"Text exceeds 280 character limit (current: {len(text)} characters)")
···
80
78
if 'data' in result:
81
79
tweet_id = result['data'].get('id', 'unknown')
82
80
tweet_url = f"https://x.com/i/status/{tweet_id}"
83
-
logger.info(f"Successfully posted to X: {tweet_url}")
84
81
return f"Successfully posted to X. Tweet ID: {tweet_id}. URL: {tweet_url}"
85
82
else:
86
83
raise Exception(f"Unexpected response format: {result}")
87
84
else:
88
85
error_msg = f"X API error: {response.status_code} - {response.text}"
89
-
logger.error(error_msg)
90
86
raise Exception(error_msg)
91
87
92
88
except requests.exceptions.RequestException as e:
93
89
error_msg = f"Network error posting to X: {str(e)}"
94
-
logger.error(error_msg)
95
90
raise Exception(error_msg)
96
91
except Exception as e:
97
92
if "Missing X API credentials" in str(e) or "X API error" in str(e):
98
93
raise
99
94
error_msg = f"Unexpected error posting to X: {str(e)}"
100
-
logger.error(error_msg)
101
-
raise Exception(error_msg)
95
+
raise Exception(error_msg)