X (Twitter) MCP server

Instructions

Implementation Reference

• src/x_twitter_mcp/server.py:275-286 (handler)

  The handler function for the 'favorite_tweet' tool. It performs rate limiting check, initializes the Twitter API client, calls the Tweepy client's like method to favorite the specified tweet, and returns the result.

  @server.tool(name="favorite_tweet", description="Favorites a tweet")
  async def favorite_tweet(tweet_id: str) -> Dict:
      """Favorites a tweet.
  
      Args:
          tweet_id (str): The ID of the tweet to favorite (like).
      """
      if not check_rate_limit("like_actions"):
          raise Exception("Like action rate limit exceeded")
      client, _ = initialize_twitter_clients()
      result = client.like(tweet_id=tweet_id)
      return {"tweet_id": tweet_id, "liked": result.data["liked"]}

• src/x_twitter_mcp/server.py:275-275 (registration)

  The @server.tool decorator registers the 'favorite_tweet' function as an MCP tool.

  @server.tool(name="favorite_tweet", description="Favorites a tweet")

• src/x_twitter_mcp/server.py:28-65 (helper)

  Helper function to lazily initialize the Tweepy Twitter API v2 Client and v1.1 API, used by favorite_tweet.

  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-91 (helper)

  Helper function for rate limiting checks, specifically used for 'like_actions' in favorite_tweet.

  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

• src/x_twitter_mcp/server.py:276-280 (schema)

  Input schema defined by function signature (tweet_id: str) and docstring, output Dict.

  async def favorite_tweet(tweet_id: str) -> Dict:
      """Favorites a tweet.
  
      Args:
          tweet_id (str): The ID of the tweet to favorite (like).