twitter_get_user_replies
Retrieve recent replies by a specific Twitter user using their ID or username. Access Twitter data without local credential setup through the MCP server, enabling structured API responses. Specify maximum results for tailored output.
Instructions
Get recent replies by a specific user
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| max_results | No | Maximum number of tweets to return (default: 10) | |
| twitter_access_token | Yes | Twitter OAuth2 access token | |
| user_id | No | Twitter user ID (optional if username is provided) | |
| username | No | Twitter username/handle (optional if user_id is provided) |
Implementation Reference
- Core handler function in TwitterClient that implements the tool logic: fetches user's recent replies using Twitter API v2 search/recent endpoint with query 'from:{user_id} is:reply'. Supports username resolution to user_id and handles errors.
def get_user_replies(self, user_id: str = None, username: str = None, max_results: int = 10) -> str: """Get recent replies by a specific user Args: user_id: Twitter user ID username: Twitter username/handle (without @ symbol) max_results: Maximum number of tweets to return (default: 10) Returns: JSON string with user replies """ try: if not self.access_token: return json.dumps({ "error": "No valid access token provided. Please refresh your token first.", "status": "error" }) # If username is provided but not user_id, look up the user_id if not user_id and username: # Remove @ symbol if it's included if username.startswith('@'): username = username[1:] logger.debug(f"Looking up user ID for username: {username}") user_lookup_result = self.get_user_id_by_username(username) user_lookup_data = json.loads(user_lookup_result) if user_lookup_data.get("status") == "success": user_id = user_lookup_data.get("user_id") logger.debug(f"Found user ID: {user_id}") else: return json.dumps({ "error": f"Could not find user ID for username: {username}", "details": user_lookup_data.get("error", "No details available"), "status": "error" }) if not user_id: return json.dumps({ "error": "Either user_id or username is required", "status": "error" }) logger.debug(f"Getting replies for user ID: {user_id}, max_results: {max_results}") # We'll use the search endpoint with a specific query to find replies url = f"{self.api_base_url}/tweets/search/recent" headers = { "Authorization": f"Bearer {self.access_token}" } # Query for tweets that are replies from the specified user query = f"from:{user_id} is:reply" params = { "query": query, "max_results": max_results, "tweet.fields": "id,text,created_at,in_reply_to_user_id,conversation_id", "expansions": "author_id,in_reply_to_user_id,referenced_tweets.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 get_user_replies: {str(e)}") return json.dumps({"error": str(e), "status": "error"}) - src/mcp_server_twitter_noauth/server.py:561-574 (registration)Tool registration in the list_tools handler, including name, description, and input schema definition.
types.Tool( name="twitter_get_user_replies", description="Get recent replies by a specific user", inputSchema={ "type": "object", "properties": { "twitter_access_token": {"type": "string", "description": "Twitter OAuth2 access token"}, "user_id": {"type": "string", "description": "Twitter user ID (optional if username is provided)"}, "username": {"type": "string", "description": "Twitter username/handle (optional if user_id is provided)"}, "max_results": {"type": "integer", "description": "Maximum number of tweets to return (default: 10)"} }, "required": ["twitter_access_token"] }, ), - Dispatch handler in the MCP server's call_tool function that extracts arguments, validates inputs, instantiates TwitterClient, calls the get_user_replies method, and returns the result.
elif name == "twitter_get_user_replies": user_id = arguments.get("user_id") username = arguments.get("username") max_results = int(arguments.get("max_results", 10)) if not user_id and not username: raise ValueError("Either user_id or username is required for twitter_get_user_replies") results = twitter.get_user_replies(user_id=user_id, username=username, max_results=max_results) return [types.TextContent(type="text", text=results)] - Helper method used by get_user_replies to resolve username to user_id via Twitter API users/by/username endpoint.
def get_user_id_by_username(self, username: str) -> str: """Lookup a user ID by username Args: username: Twitter username/handle (without the @ symbol) Returns: JSON string with user data including the user ID """ try: if not self.access_token: return json.dumps({ "error": "No valid access token provided. Please refresh your token first.", "status": "error" }) # Remove @ symbol if it's included if username.startswith('@'): username = username[1:] logger.debug(f"Looking up user ID for username: {username}") # Twitter API v2 user lookup by username endpoint url = f"{self.api_base_url}/users/by/username/{username}" headers = { "Authorization": f"Bearer {self.access_token}" } params = { "user.fields": "id,name,username" } response = requests.get(url, headers=headers, params=params) response.raise_for_status() result = response.json() # Extract the user ID from the response if "data" in result and "id" in result["data"]: user_id = result["data"]["id"] logger.debug(f"Found user ID: {user_id} for username: {username}") return json.dumps({ "user_id": user_id, "data": result["data"], "status": "success" }) else: return json.dumps({ "error": "User not found or ID not available", "status": "error" }) 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 get_user_id_by_username: {str(e)}") return json.dumps({"error": str(e), "status": "error"})