# Copyright (c) 2026 Dedalus Labs, Inc. and its contributors
# SPDX-License-Identifier: MIT
"""X/Twitter API read-only tools.
Uses OAuth 2.0 App-Only (Bearer Token) authentication.
Credentials provided by client via token exchange.
No server-side environment variables required.
API docs: https://docs.x.com/x-api/introduction
"""
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, SecretKeys
from dedalus_mcp.types import ToolAnnotations
from pydantic import Field
from pydantic.dataclasses import dataclass
# X API v2 uses Bearer token auth
# Base URL: https://api.x.com (or api.twitter.com)
# All paths include /2 prefix for v2 endpoints
x = Connection(
name="x",
secrets=SecretKeys(token="X_BEARER_TOKEN"),
auth_header_format="Bearer {api_key}",
)
# Common tweet fields to include in responses
DEFAULT_TWEET_FIELDS = "id,text,author_id,created_at,public_metrics,conversation_id"
DEFAULT_USER_FIELDS = "id,name,username,description,public_metrics,profile_image_url,verified"
@dataclass(frozen=True)
class XResult:
"""X API result."""
success: bool
data: Any = None
meta: dict[str, Any] = Field(default_factory=dict)
error: str | None = None
async def _req(method: HttpMethod, path: str, body: Any = None) -> XResult:
"""Execute X API request."""
ctx = get_context()
resp = await ctx.dispatch("x", HttpRequest(method=method, path=path, body=body))
if resp.success:
body = resp.response.body or {}
return XResult(success=True, data=body.get("data"), meta=body.get("meta", {}))
return XResult(success=False, error=resp.error.message if resp.error else "Request failed")
# --- Users ---
@tool(
description="Get the authenticated X user's profile (requires user access token, not app-only bearer)",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_whoami() -> XResult:
"""Get authenticated user profile.
NOTE: This endpoint requires a User Access Token (OAuth 2.0 with PKCE),
not an app-only Bearer token. If using app-only auth, use x_get_user_by_username instead.
Returns:
XResult with user data (id, name, username, etc.)
"""
return await _req(HttpMethod.GET, f"/2/users/me?user.fields={DEFAULT_USER_FIELDS}")
@tool(
description="Get an X user by their username",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_by_username(username: str) -> XResult:
"""Get user by username.
Args:
username: X username (without @)
Returns:
XResult with user data
"""
return await _req(HttpMethod.GET, f"/2/users/by/username/{username}?user.fields={DEFAULT_USER_FIELDS}")
@tool(
description="Get an X user by their ID",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user(user_id: str) -> XResult:
"""Get user by ID.
Args:
user_id: X user ID
Returns:
XResult with user data
"""
return await _req(HttpMethod.GET, f"/2/users/{user_id}?user.fields={DEFAULT_USER_FIELDS}")
@tool(
description="Get multiple X users by their IDs",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_users(user_ids: list[str]) -> XResult:
"""Get multiple users by ID.
Args:
user_ids: List of X user IDs (max 100)
Returns:
XResult with list of users
"""
ids = ",".join(user_ids[:100])
return await _req(HttpMethod.GET, f"/2/users?ids={ids}&user.fields={DEFAULT_USER_FIELDS}")
# --- Tweets ---
@tool(
description="Get a tweet by its ID",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_tweet(tweet_id: str) -> XResult:
"""Get tweet by ID.
Args:
tweet_id: Tweet ID
Returns:
XResult with tweet data
"""
return await _req(HttpMethod.GET, f"/2/tweets/{tweet_id}?tweet.fields={DEFAULT_TWEET_FIELDS}")
@tool(
description="Get multiple tweets by their IDs",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_tweets(tweet_ids: list[str]) -> XResult:
"""Get multiple tweets by ID.
Args:
tweet_ids: List of tweet IDs (max 100)
Returns:
XResult with list of tweets
"""
ids = ",".join(tweet_ids[:100])
return await _req(HttpMethod.GET, f"/2/tweets?ids={ids}&tweet.fields={DEFAULT_TWEET_FIELDS}")
@tool(
description="Get a user's recent tweets",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_tweets(
user_id: str,
max_results: int = 10,
pagination_token: str | None = None,
) -> XResult:
"""Get tweets authored by a user.
Args:
user_id: X user ID
max_results: Number of tweets to return (5-100, default 10)
pagination_token: Token for next page of results
Returns:
XResult with list of tweets and pagination meta
"""
max_results = max(5, min(100, max_results))
path = f"/2/users/{user_id}/tweets?tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
@tool(
description="Get tweets that mention a specific user",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_mentions(
user_id: str,
max_results: int = 10,
pagination_token: str | None = None,
) -> XResult:
"""Get tweets mentioning a user.
Args:
user_id: X user ID
max_results: Number of tweets to return (5-100, default 10)
pagination_token: Token for next page of results
Returns:
XResult with list of tweets and pagination meta
"""
max_results = max(5, min(100, max_results))
path = f"/2/users/{user_id}/mentions?tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
# --- Search ---
@tool(
description="Search recent tweets (last 7 days)",
tags=["search", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_search_recent(
query: str,
max_results: int = 10,
pagination_token: str | None = None,
) -> XResult:
"""Search recent tweets from the last 7 days.
Args:
query: Search query (see X API docs for operators)
max_results: Number of tweets to return (10-100, default 10)
pagination_token: Token for next page of results
Returns:
XResult with list of tweets and pagination meta
Example queries:
- "from:elonmusk" - tweets from a user
- "#AI" - tweets with hashtag
- "machine learning -is:retweet" - exclude retweets
- "python lang:en" - English tweets about python
"""
from urllib.parse import quote
max_results = max(10, min(100, max_results))
path = f"/2/tweets/search/recent?query={quote(query)}&tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&next_token={pagination_token}"
return await _req(HttpMethod.GET, path)
# --- Followers/Following ---
@tool(
description="Get users who follow a specific user",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_followers(
user_id: str,
max_results: int = 100,
pagination_token: str | None = None,
) -> XResult:
"""Get a user's followers.
Args:
user_id: X user ID
max_results: Number of users to return (1-1000, default 100)
pagination_token: Token for next page of results
Returns:
XResult with list of users and pagination meta
"""
max_results = max(1, min(1000, max_results))
path = f"/2/users/{user_id}/followers?user.fields={DEFAULT_USER_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
@tool(
description="Get users that a specific user follows",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_following(
user_id: str,
max_results: int = 100,
pagination_token: str | None = None,
) -> XResult:
"""Get users that a user follows.
Args:
user_id: X user ID
max_results: Number of users to return (1-1000, default 100)
pagination_token: Token for next page of results
Returns:
XResult with list of users and pagination meta
"""
max_results = max(1, min(1000, max_results))
path = f"/2/users/{user_id}/following?user.fields={DEFAULT_USER_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
# --- Likes ---
@tool(
description="Get tweets liked by a user",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_liked_tweets(
user_id: str,
max_results: int = 10,
pagination_token: str | None = None,
) -> XResult:
"""Get tweets liked by a user.
Args:
user_id: X user ID
max_results: Number of tweets to return (10-100, default 10)
pagination_token: Token for next page of results
Returns:
XResult with list of tweets and pagination meta
"""
max_results = max(10, min(100, max_results))
path = f"/2/users/{user_id}/liked_tweets?tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
@tool(
description="Get users who liked a specific tweet",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_liking_users(
tweet_id: str,
max_results: int = 100,
pagination_token: str | None = None,
) -> XResult:
"""Get users who liked a tweet.
Args:
tweet_id: Tweet ID
max_results: Number of users to return (1-100, default 100)
pagination_token: Token for next page of results
Returns:
XResult with list of users and pagination meta
"""
max_results = max(1, min(100, max_results))
path = f"/2/tweets/{tweet_id}/liking_users?user.fields={DEFAULT_USER_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
# --- Retweets ---
@tool(
description="Get users who retweeted a specific tweet",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_retweeted_by(
tweet_id: str,
max_results: int = 100,
pagination_token: str | None = None,
) -> XResult:
"""Get users who retweeted a tweet.
Args:
tweet_id: Tweet ID
max_results: Number of users to return (1-100, default 100)
pagination_token: Token for next page of results
Returns:
XResult with list of users and pagination meta
"""
max_results = max(1, min(100, max_results))
path = f"/2/tweets/{tweet_id}/retweeted_by?user.fields={DEFAULT_USER_FIELDS}&max_results={max_results}"
if pagination_token:
path += f"&pagination_token={pagination_token}"
return await _req(HttpMethod.GET, path)
# --- Export ---
x_tools = [
# Users
x_whoami,
x_get_user_by_username,
x_get_user,
x_get_users,
# Tweets
x_get_tweet,
x_get_tweets,
x_get_user_tweets,
x_get_user_mentions,
# Search
x_search_recent,
# Followers/Following
x_get_followers,
x_get_following,
# Likes
x_get_user_liked_tweets,
x_get_liking_users,
# Retweets
x_get_retweeted_by,
]
