X (Twitter) MCP server

Instructions

Implementation Reference

• src/x_twitter_mcp/server.py:127-141 (handler)

  The handler function for the 'get_user_followers' tool. It checks rate limits, initializes the Twitter client, fetches the followers using Tweepy's get_users_followers method, and returns a list of user dictionaries.
  @server.tool(name="get_user_followers", description="Retrieves a list of followers for a given user") async def get_user_followers(user_id: str, count: Optional[int] = 100, cursor: Optional[str] = None) -> List[Dict]: """Retrieves a list of followers for a given user. Args: user_id (str): The user ID whose followers are to be retrieved. count (Optional[int]): The number of followers to retrieve per page. Default is 100. Max is 100 for V2 API. cursor (Optional[str]): A pagination token for fetching the next set of results. """ if not check_rate_limit("follow_actions"): raise Exception("Follow action rate limit exceeded") client, _ = initialize_twitter_clients() followers = client.get_users_followers(id=user_id, max_results=count, pagination_token=cursor, user_fields=["id", "name", "username"]) return [user.data for user in followers.data]
• src/x_twitter_mcp/server.py:127-127 (registration)

  The @server.tool decorator registers the get_user_followers function as an MCP tool with the specified name and description.
  @server.tool(name="get_user_followers", description="Retrieves a list of followers for a given user")
• src/x_twitter_mcp/server.py:128-135 (schema)

  The function signature and docstring define the input schema (parameters: user_id (str), count (Optional[int]=100), cursor (Optional[str]=None)) and output (List[Dict]).
  async def get_user_followers(user_id: str, count: Optional[int] = 100, cursor: Optional[str] = None) -> List[Dict]: """Retrieves a list of followers for a given user. Args: user_id (str): The user ID whose followers are to be retrieved. count (Optional[int]): The number of followers to retrieve per page. Default is 100. Max is 100 for V2 API. cursor (Optional[str]): A pagination token for fetching the next set of results. """
• src/x_twitter_mcp/server.py:28-66 (helper)

  Helper function to lazily initialize the Tweepy Twitter v2 Client and v1.1 API.
  def initialize_twitter_clients() -> tuple[tweepy.Client, tweepy.API]: """Initialize Twitter API clients on-demand.""" global _twitter_client, _twitter_v1_api if _twitter_client is not None and _twitter_v1_api is not None: return _twitter_client, _twitter_v1_api # Verify required environment variables required_env_vars = [ "TWITTER_API_KEY", "TWITTER_API_SECRET", "TWITTER_ACCESS_TOKEN", "TWITTER_ACCESS_TOKEN_SECRET", "TWITTER_BEARER_TOKEN", ] for var in required_env_vars: if not os.getenv(var): raise EnvironmentError(f"Missing required environment variable: {var}") # Initialize v2 API client _twitter_client = tweepy.Client( consumer_key=os.getenv("TWITTER_API_KEY"), consumer_secret=os.getenv("TWITTER_API_SECRET"), access_token=os.getenv("TWITTER_ACCESS_TOKEN"), access_token_secret=os.getenv("TWITTER_ACCESS_TOKEN_SECRET"), bearer_token=os.getenv("TWITTER_BEARER_TOKEN") ) # Initialize v1.1 API for media uploads and other unsupported v2 endpoints auth = tweepy.OAuth1UserHandler( consumer_key=os.getenv("TWITTER_API_KEY"), consumer_secret=os.getenv("TWITTER_API_SECRET"), access_token=os.getenv("TWITTER_ACCESS_TOKEN"), access_token_secret=os.getenv("TWITTER_ACCESS_TOKEN_SECRET") ) _twitter_v1_api = tweepy.API(auth) return _twitter_client, _twitter_v1_api
• src/x_twitter_mcp/server.py:78-92 (helper)

  Helper function to check rate limits before performing actions like fetching followers.
  def check_rate_limit(action_type: str) -> bool: """Check if the action is within rate limits.""" config = RATE_LIMITS.get(action_type) if not config: return True # No limit defined counter = rate_limit_counters[action_type] now = datetime.now() if now >= counter["reset_time"]: counter["count"] = 0 counter["reset_time"] = now + config["window"] if counter["count"] >= config["limit"]: return False counter["count"] += 1 return True