twitter_search_tweets

Instructions

Implementation Reference

• src/mcp_server_twitter_noauth/server.py:216-262 (handler)

  Core handler logic for the twitter_search_tweets tool: TwitterClient.search_tweets method performs the actual API request to Twitter's search endpoint.

  def search_tweets(self, query: str, max_results: int = 10) -> str:
      """Search for tweets using the Twitter API
      
      Args:
          query: The search query to execute
          max_results: Maximum number of tweets to return (default: 10)
          
      Returns:
          JSON string with search results
      """
      try:
          if not self.access_token:
              return json.dumps({
                  "error": "No valid access token provided. Please refresh your token first.",
                  "status": "error"
              })
          
          logger.debug(f"Searching tweets with query: {query}, max_results: {max_results}")
          
          # Twitter API v2 search recent endpoint
          url = f"{self.api_base_url}/tweets/search/recent"
          
          headers = {
              "Authorization": f"Bearer {self.access_token}"
          }
          
          params = {
              "query": query,
              "max_results": max_results,
              "tweet.fields": "id,text,created_at,author_id",
              "expansions": "author_id",
              "user.fields": "id,name,username"
          }
          
          response = requests.get(url, headers=headers, params=params)
          response.raise_for_status()
          
          # Return the raw JSON response
          return json.dumps(response.json())
          
      except requests.exceptions.RequestException as e:
          logger.error(f"API request error: {str(e)}")
          return json.dumps({"error": str(e), "status": "error"})
      except Exception as e:
          logger.error(f"Exception in search_tweets: {str(e)}")
          return json.dumps({"error": str(e), "status": "error"})

• src/mcp_server_twitter_noauth/server.py:534-546 (registration)

  Registration of the twitter_search_tweets tool in the MCP server's list_tools decorator, defining name, description, and input schema.

  types.Tool(
      name="twitter_search_tweets",
      description="Search for tweets using the Twitter API",
      inputSchema={
          "type": "object",
          "properties": {
              "twitter_access_token": {"type": "string", "description": "Twitter OAuth2 access token"},
              "query": {"type": "string", "description": "The search query to execute"},
              "max_results": {"type": "integer", "description": "Maximum number of tweets to return (default: 10)"}
          },
          "required": ["twitter_access_token", "query"]
      },
  ),

• src/mcp_server_twitter_noauth/server.py:646-655 (handler)

  Dispatch handler in @server.call_tool() that extracts arguments and calls TwitterClient.search_tweets for the twitter_search_tweets tool.

  if name == "twitter_search_tweets":
      query = arguments.get("query")
      max_results = int(arguments.get("max_results", 10))
      
      if not query:
          raise ValueError("query is required for twitter_search_tweets")
      
      results = twitter.search_tweets(query=query, max_results=max_results)
      return [types.TextContent(type="text", text=results)]

• src/mcp_server_twitter_noauth/server.py:538-544 (schema)

  Input schema definition for the twitter_search_tweets tool, specifying required parameters and types.

  "type": "object",
  "properties": {
      "twitter_access_token": {"type": "string", "description": "Twitter OAuth2 access token"},
      "query": {"type": "string", "description": "The search query to execute"},
      "max_results": {"type": "integer", "description": "Maximum number of tweets to return (default: 10)"}
  },
  "required": ["twitter_access_token", "query"]