1"""
2Twitter/X Tools — BaseTool subclasses for agent use.
3 
413 read-only tools: search tweets, get tweets, user info, user tweets,
5user followers, user followings, tweet replies, tweet retweeters, search users,
6article, thread context, quote tweets, trends.
7"""
8 
9import asyncio
10import logging
11 
12from core.tool import BaseTool, ToolContext, ToolResult
13from .client import TwitterApiClient
14 
15logger = logging.getLogger(__name__)
16 
17# Module-level shared client instance
18_client: TwitterApiClient = None
19 
20 
21def _get_client() -> TwitterApiClient:
22    global _client
23    if _client is None:
24        _client = TwitterApiClient()
25    return _client
26 
27 
28# ── Tweet Tools ──────────────────────────────────────────────────────────────
29 
30 
31class TwitterSearchTweetsTool(BaseTool):
32    """Search tweets with advanced query syntax."""
33 
34    @property
35    def name(self) -> str:
36        return "twitter_search_tweets"
37 
38    @property
39    def description(self) -> str:
40        return """Search Twitter/X tweets using advanced query syntax.
41 
42Supports operators: keyword matching, from:user, to:user, #hashtag, $cashtag,
43lang:en, has:media, has:links, is:reply, min_faves:100, since:2024-01-01, until:2024-12-31.
44 
45Parameters:
46- query: Search query (required). Examples: "bitcoin", "from:elonmusk crypto", "$SOL min_faves:50"
47- cursor: Pagination cursor from previous response (optional)
48 
49Returns: tweets array with text, author, metrics, and next cursor for pagination"""
50 
51    @property
52    def parameters(self) -> dict:
53        return {
54            "type": "object",
55            "properties": {
56                "query": {
57                    "type": "string",
58                    "description": "Search query (supports advanced operators)",
59                },
60                "cursor": {
61                    "type": "string",
62                    "description": "Pagination cursor from previous response",
63                },
64            },
65            "required": ["query"],
66        }
67 
68    async def execute(self, ctx: ToolContext, query: str = "", cursor: str = None, **kwargs) -> ToolResult:
69        if not query:
70            return ToolResult(success=False, error="'query' is required")
71        try:
72            client = _get_client()
73            data = await asyncio.to_thread(client.search_tweets, query, cursor=cursor)
74            return ToolResult(success=True, output=data)
75        except Exception as e:
76            return ToolResult(success=False, error=str(e))
77 
78 
79class TwitterGetTweetsTool(BaseTool):
80    """Get tweets by their IDs."""
81 
82    @property
83    def name(self) -> str:
84        return "twitter_get_tweets"
85 
86    @property
87    def description(self) -> str:
88        return """Get one or more tweets by their tweet IDs.
89 
90Parameters:
91- tweet_ids: Array of tweet ID strings (required, e.g. ["1234567890", "9876543210"])
92 
93Returns: tweets array with full tweet data"""
94 
95    @property
96    def parameters(self) -> dict:
97        return {
98            "type": "object",
99            "properties": {
100                "tweet_ids": {
101                    "type": "array",
102                    "items": {"type": "string"},
103                    "description": "Array of tweet ID strings",
104                },
105            },
106            "required": ["tweet_ids"],
107        }
108 
109    async def execute(self, ctx: ToolContext, tweet_ids: list = None, **kwargs) -> ToolResult:
110        if not tweet_ids:
111            return ToolResult(success=False, error="'tweet_ids' is required")
112        try:
113            client = _get_client()
114            data = await asyncio.to_thread(client.get_tweets, tweet_ids)
115            return ToolResult(success=True, output=data)
116        except Exception as e:
117            return ToolResult(success=False, error=str(e))
118 
119 
120class TwitterTweetRepliesTool(BaseTool):
121    """Get replies to a specific tweet."""
122 
123    @property
124    def name(self) -> str:
125        return "twitter_tweet_replies"
126 
127    @property
128    def description(self) -> str:
129        return """Get replies to a specific tweet.
130 
131Parameters:
132- tweet_id: Tweet ID to get replies for (required)
133- cursor: Pagination cursor from previous response (optional)
134 
135Returns: replies array with tweet data and next cursor"""
136 
137    @property
138    def parameters(self) -> dict:
139        return {
140            "type": "object",
141            "properties": {
142                "tweet_id": {
143                    "type": "string",
144                    "description": "Tweet ID to get replies for",
145                },
146                "cursor": {
147                    "type": "string",
148                    "description": "Pagination cursor from previous response",
149                },
150            },
151            "required": ["tweet_id"],
152        }
153 
154    async def execute(self, ctx: ToolContext, tweet_id: str = "", cursor: str = None, **kwargs) -> ToolResult:
155        if not tweet_id:
156            return ToolResult(success=False, error="'tweet_id' is required")
157        try:
158            client = _get_client()
159            data = await asyncio.to_thread(client.get_tweet_replies, tweet_id, cursor=cursor)
160            return ToolResult(success=True, output=data)
161        except Exception as e:
162            return ToolResult(success=False, error=str(e))
163 
164 
165class TwitterTweetRetweetersTool(BaseTool):
166    """Get users who retweeted a tweet."""
167 
168    @property
169    def name(self) -> str:
170        return "twitter_tweet_retweeters"
171 
172    @property
173    def description(self) -> str:
174        return """Get users who retweeted a specific tweet.
175 
176Parameters:
177- tweet_id: Tweet ID to get retweeters for (required)
178- cursor: Pagination cursor from previous response (optional)
179 
180Returns: users array with profile data and next cursor"""
181 
182    @property
183    def parameters(self) -> dict:
184        return {
185            "type": "object",
186            "properties": {
187                "tweet_id": {
188                    "type": "string",
189                    "description": "Tweet ID to get retweeters for",
190                },
191                "cursor": {
192                    "type": "string",
193                    "description": "Pagination cursor from previous response",
194                },
195            },
196            "required": ["tweet_id"],
197        }
198 
199    async def execute(self, ctx: ToolContext, tweet_id: str = "", cursor: str = None, **kwargs) -> ToolResult:
200        if not tweet_id:
201            return ToolResult(success=False, error="'tweet_id' is required")
202        try:
203            client = _get_client()
204            data = await asyncio.to_thread(client.get_tweet_retweeters, tweet_id, cursor=cursor)
205            return ToolResult(success=True, output=data)
206        except Exception as e:
207            return ToolResult(success=False, error=str(e))
208 
209 
210# ── User Tools ───────────────────────────────────────────────────────────────
211 
212 
213class TwitterGetArticleTool(BaseTool):
214    """Get long-form article for a tweet."""
215 
216    @property
217    def name(self) -> str:
218        return "twitter_get_article"
219 
220    @property
221    def description(self) -> str:
222        return """Get X/Twitter long-form article by tweet ID.
223 
224Parameters:
225- tweet_id: Tweet ID of the article post (required)
226 
227Returns: article object with title, preview text, cover media URL, and content blocks"""
228 
229    @property
230    def parameters(self) -> dict:
231        return {
232            "type": "object",
233            "properties": {
234                "tweet_id": {
235                    "type": "string",
236                    "description": "Tweet ID of the article post",
237                },
238            },
239            "required": ["tweet_id"],
240        }
241 
242    async def execute(self, ctx: ToolContext, tweet_id: str = "", **kwargs) -> ToolResult:
243        if not tweet_id:
244            return ToolResult(success=False, error="'tweet_id' is required")
245        try:
246            client = _get_client()
247            data = await asyncio.to_thread(client.get_article, tweet_id)
248            return ToolResult(success=True, output=data)
249        except Exception as e:
250            return ToolResult(success=False, error=str(e))
251 
252 
253class TwitterTweetThreadContextTool(BaseTool):
254    """Get complete thread context for a tweet."""
255 
256    @property
257    def name(self) -> str:
258        return "twitter_tweet_thread_context"
259 
260    @property
261    def description(self) -> str:
262        return """Get complete thread context for a tweet.
263 
264Parameters:
265- tweet_id: Tweet ID to get thread context for (required)
266 
267Returns: parent tweets + direct replies in a single response"""
268 
269    @property
270    def parameters(self) -> dict:
271        return {
272            "type": "object",
273            "properties": {
274                "tweet_id": {
275                    "type": "string",
276                    "description": "Tweet ID to get thread context for",
277                },
278            },
279            "required": ["tweet_id"],
280        }
281 
282    async def execute(self, ctx: ToolContext, tweet_id: str = "", **kwargs) -> ToolResult:
283        if not tweet_id:
284            return ToolResult(success=False, error="'tweet_id' is required")
285        try:
286            client = _get_client()
287            data = await asyncio.to_thread(client.get_tweet_thread_context, tweet_id)
288            return ToolResult(success=True, output=data)
289        except Exception as e:
290            return ToolResult(success=False, error=str(e))
291 
292 
293class TwitterTweetQuoteTool(BaseTool):
294    """Get quote tweets for a tweet."""
295 
296    @property
297    def name(self) -> str:
298        return "twitter_tweet_quote"
299 
300    @property
301    def description(self) -> str:
302        return """Get quote tweets for a specific tweet.
303 
304Parameters:
305- tweet_id: Tweet ID to get quote tweets for (required)
306- cursor: Pagination cursor from previous response (optional)
307 
308Returns: quote tweets list and next cursor"""
309 
310    @property
311    def parameters(self) -> dict:
312        return {
313            "type": "object",
314            "properties": {
315                "tweet_id": {
316                    "type": "string",
317                    "description": "Tweet ID to get quote tweets for",
318                },
319                "cursor": {
320                    "type": "string",
321                    "description": "Pagination cursor from previous response",
322                },
323            },
324            "required": ["tweet_id"],
325        }
326 
327    async def execute(self, ctx: ToolContext, tweet_id: str = "", cursor: str = None, **kwargs) -> ToolResult:
328        if not tweet_id:
329            return ToolResult(success=False, error="'tweet_id' is required")
330        try:
331            client = _get_client()
332            data = await asyncio.to_thread(client.get_tweet_quote, tweet_id, cursor=cursor)
333            return ToolResult(success=True, output=data)
334        except Exception as e:
335            return ToolResult(success=False, error=str(e))
336 
337 
338class TwitterGetTrendsTool(BaseTool):
339    """Get Twitter/X trends."""
340 
341    @property
342    def name(self) -> str:
343        return "twitter_get_trends"
344 
345    @property
346    def description(self) -> str:
347        return """Get Twitter/X trends.
348 
349Parameters:
350- woeid: Where On Earth ID (optional)
351- country: Country code/name (optional)
352- category: Trend category (optional)
353- limit: Number of trends to return (optional)
354 
355Returns: trends list"""
356 
357    @property
358    def parameters(self) -> dict:
359        return {
360            "type": "object",
361            "properties": {
362                "woeid": {
363                    "type": "string",
364                    "description": "Where On Earth ID",
365                },
366                "country": {
367                    "type": "string",
368                    "description": "Country code/name",
369                },
370                "category": {
371                    "type": "string",
372                    "description": "Trend category",
373                },
374                "limit": {
375                    "type": "integer",
376                    "description": "Number of trends to return",
377                },
378            },
379            "required": [],
380        }
381 
382    async def execute(
383        self,
384        ctx: ToolContext,
385        woeid: str = None,
386        country: str = None,
387        category: str = None,
388        limit: int = None,
389        **kwargs,
390    ) -> ToolResult:
391        try:
392            client = _get_client()
393            data = await asyncio.to_thread(
394                client.get_trends,
395                woeid=woeid,
396                country=country,
397                category=category,
398                limit=limit,
399            )
400            return ToolResult(success=True, output=data)
401        except Exception as e:
402            return ToolResult(success=False, error=str(e))
403 
404 
405class TwitterUserInfoTool(BaseTool):
406    """Get a Twitter user's profile information."""
407 
408    @property
409    def name(self) -> str:
410        return "twitter_user_info"
411 
412    @property
413    def description(self) -> str:
414        return """Get a Twitter/X user's profile information: bio, follower count, following count, tweet count, verification status.
415 
416Parameters:
417- username: Twitter handle without @ (required, e.g. "elonmusk")
418 
419Returns: user profile with name, bio, followers_count, following_count, tweet_count, verified"""
420 
421    @property
422    def parameters(self) -> dict:
423        return {
424            "type": "object",
425            "properties": {
426                "username": {
427                    "type": "string",
428                    "description": "Twitter handle without @ (e.g. 'elonmusk')",
429                },
430            },
431            "required": ["username"],
432        }
433 
434    async def execute(self, ctx: ToolContext, username: str = "", **kwargs) -> ToolResult:
435        if not username:
436            return ToolResult(success=False, error="'username' is required")
437        try:
438            client = _get_client()
439            data = await asyncio.to_thread(client.get_user_info, username)
440            return ToolResult(success=True, output=data)
441        except Exception as e:
442            return ToolResult(success=False, error=str(e))
443 
444 
445class TwitterUserTweetsTool(BaseTool):
446    """Get a user's recent tweets."""
447 
448    @property
449    def name(self) -> str:
450        return "twitter_user_tweets"
451 
452    @property
453    def description(self) -> str:
454        return """Get a Twitter/X user's recent tweets.
455 
456Parameters:
457- username: Twitter handle without @ (required, e.g. "elonmusk")
458- cursor: Pagination cursor from previous response (optional)
459 
460Returns: tweets array with text, metrics, timestamps, and next cursor"""
461 
462    @property
463    def parameters(self) -> dict:
464        return {
465            "type": "object",
466            "properties": {
467                "username": {
468                    "type": "string",
469                    "description": "Twitter handle without @ (e.g. 'elonmusk')",
470                },
471                "cursor": {
472                    "type": "string",
473                    "description": "Pagination cursor from previous response",
474                },
475            },
476            "required": ["username"],
477        }
478 
479    async def execute(self, ctx: ToolContext, username: str = "", cursor: str = None, **kwargs) -> ToolResult:
480        if not username:
481            return ToolResult(success=False, error="'username' is required")
482        try:
483            client = _get_client()
484            data = await asyncio.to_thread(client.get_user_tweets, username, cursor=cursor)
485            return ToolResult(success=True, output=data)
486        except Exception as e:
487            return ToolResult(success=False, error=str(e))
488 
489 
490class TwitterUserFollowersTool(BaseTool):
491    """Get a user's followers."""
492 
493    @property
494    def name(self) -> str:
495        return "twitter_user_followers"
496 
497    @property
498    def description(self) -> str:
499        return """Get a Twitter/X user's followers.
500 
501Parameters:
502- username: Twitter handle without @ (required, e.g. "elonmusk")
503- cursor: Pagination cursor from previous response (optional)
504 
505Returns: users array with profile data and next cursor"""
506 
507    @property
508    def parameters(self) -> dict:
509        return {
510            "type": "object",
511            "properties": {
512                "username": {
513                    "type": "string",
514                    "description": "Twitter handle without @ (e.g. 'elonmusk')",
515                },
516                "cursor": {
517                    "type": "string",
518                    "description": "Pagination cursor from previous response",
519                },
520            },
521            "required": ["username"],
522        }
523 
524    async def execute(self, ctx: ToolContext, username: str = "", cursor: str = None, **kwargs) -> ToolResult:
525        if not username:
526            return ToolResult(success=False, error="'username' is required")
527        try:
528            client = _get_client()
529            data = await asyncio.to_thread(client.get_user_followers, username, cursor=cursor)
530            return ToolResult(success=True, output=data)
531        except Exception as e:
532            return ToolResult(success=False, error=str(e))
533 
534 
535class TwitterUserFollowingsTool(BaseTool):
536    """Get accounts a user follows."""
537 
538    @property
539    def name(self) -> str:
540        return "twitter_user_followings"
541 
542    @property
543    def description(self) -> str:
544        return """Get accounts that a Twitter/X user follows.
545 
546Parameters:
547- username: Twitter handle without @ (required, e.g. "elonmusk")
548- cursor: Pagination cursor from previous response (optional)
549 
550Returns: users array with profile data and next cursor"""
551 
552    @property
553    def parameters(self) -> dict:
554        return {
555            "type": "object",
556            "properties": {
557                "username": {
558                    "type": "string",
559                    "description": "Twitter handle without @ (e.g. 'elonmusk')",
560                },
561                "cursor": {
562                    "type": "string",
563                    "description": "Pagination cursor from previous response",
564                },
565            },
566            "required": ["username"],
567        }
568 
569    async def execute(self, ctx: ToolContext, username: str = "", cursor: str = None, **kwargs) -> ToolResult:
570        if not username:
571            return ToolResult(success=False, error="'username' is required")
572        try:
573            client = _get_client()
574            data = await asyncio.to_thread(client.get_user_followings, username, cursor=cursor)
575            return ToolResult(success=True, output=data)
576        except Exception as e:
577            return ToolResult(success=False, error=str(e))
578 
579 
580class TwitterSearchUsersTool(BaseTool):
581    """Search for Twitter users."""
582 
583    @property
584    def name(self) -> str:
585        return "twitter_search_users"
586 
587    @property
588    def description(self) -> str:
589        return """Search for Twitter/X users by name or keyword.
590 
591Parameters:
592- query: Search query (required, e.g. "crypto analyst", "bitcoin")
593- cursor: Pagination cursor from previous response (optional)
594 
595Returns: users array with profile data and next cursor"""
596 
597    @property
598    def parameters(self) -> dict:
599        return {
600            "type": "object",
601            "properties": {
602                "query": {
603                    "type": "string",
604                    "description": "Search query for users",
605                },
606                "cursor": {
607                    "type": "string",
608                    "description": "Pagination cursor from previous response",
609                },
610            },
611            "required": ["query"],
612        }
613 
614    async def execute(self, ctx: ToolContext, query: str = "", cursor: str = None, **kwargs) -> ToolResult:
615        if not query:
616            return ToolResult(success=False, error="'query' is required")
617        try:
618            client = _get_client()
619            data = await asyncio.to_thread(client.search_users, query, cursor=cursor)
620            return ToolResult(success=True, output=data)
621        except Exception as e:
622            return ToolResult(success=False, error=str(e))
623