# Directory Structure
```
├── .gitignore
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── smithery.yaml
└── src
└── mcp_twikit
├── __init__.py
└── twitter.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
.venv/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# MCP-Twikit
[](https://smithery.ai/server/mcp-twikit)
A Model Context Protocol (MCP) server for interacting with Twitter.
<a href="https://glama.ai/mcp/servers/49i9dd08w8"><img width="380" height="200" src="https://glama.ai/mcp/servers/49i9dd08w8/badge" alt="mcp-twikit MCP server" /></a>
## Installation
### Installing via Smithery
To install Twikit Twitter Search for Claude Desktop automatically via [Smithery](https://smithery.ai/server/mcp-twikit):
```bash
npx -y @smithery/cli install mcp-twikit --client claude
```
### Manual Installation
```json
{
"mcpServer": {
"command": "uvx",
"args": ["--from", "git+https://github.com/adhikasp/mcp-twikit", "mcp-twikit"],
"env": {
"TWITTER_USERNAME": "@example",
"TWITTER_EMAIL": "[email protected]",
"TWITTER_PASSWORD": "secret",
}
}
}
```
## Example usage
Using [mcp-client-cli](https://github.com/adhikasp/mcp-client-cli)
### Comparing sentiments across different twitter accounts
```
$ llm compare 20 latest tweet directed @IndiHomeCare, @di_cbn, @BiznetHome, @ID_MyRepublic. What are people sentiment to the product? Do 1 search for each account
I'll help you analyze the latest tweets for these internet service providers in Indonesia to understand the public sentiment. I'll use the search_twitter function for each account.
Tool Calls:
search_twitter
Args:
query: to:IndiHomeCare
count: 20
sort_by: Latest
Tool Calls:
search_twitter
Args:
query: to:di_cbn
count: 20
sort_by: Latest
Tool Calls:
search_twitter
Args:
query: to:BiznetHome
count: 20
sort_by: Latest
Tool Calls:
search_twitter
Args:
query: to:ID_MyRepublic
count: 20
sort_by: Latest
Sentiment Analysis Summary:
1. IndiHomeCare:
- Predominantly Negative Sentiment
- Key Issues:
- Unstable internet connection
- Slow response times
- Intermittent service disruptions
- Long waiting times for customer support
- Specific Complaints:
- Network instability
- Lagging internet
- Unresolved technical issues
- Lack of timely communication
2. CBN (di_cbn):
- Mostly Negative Sentiment
- Key Issues:
- Prolonged internet outages (some reporting 1-2 days of no service)
- Slow customer support response
- Lack of clear communication about service disruptions
- Specific Complaints:
- LOS (Line of Sight) issues
- No timely resolution of technical problems
- Delayed email and support responses
3. BiznetHome:
- Strongly Negative Sentiment
- Key Issues:
- Frequent network interruptions
- Slow internet speeds
- Poor customer support responsiveness
- Specific Complaints:
- Packet loss
- Inconsistent connection
- Long download times
- Unresolved technical issues
4. MyRepublic (ID_MyRepublic):
- Extremely Negative Sentiment
- Key Issues:
- Severe and prolonged service disruptions
- Complete internet outages
- Inability to access online services and games
- Extremely poor customer support
- Specific Complaints:
- Multi-day service interruptions
- Inability to access websites and online games
- Missed technician appointments
- Lack of compensation or clear communication
Overall Sentiment Ranking (from Bad to Worst):
1. IndiHomeCare (Bad)
2. CBN (Worse)
3. BiznetHome (Very Bad)
4. MyRepublic (Extremely Bad)
Common Themes Across Providers:
- Poor network stability
- Slow customer support response
- Lack of transparent communication
- Prolonged service interruptions
- Unresolved technical issues
Recommendation: These Indonesian internet service providers are experiencing significant customer satisfaction issues. Customers are frustrated with unreliable services, poor technical support, and lack of timely problem resolution.
```
### Getting tweets from your home timeline
```
$ llm what is happening on my twitter timeline?
I'll help you check your Twitter timeline. I'll retrieve the latest tweets from your timeline:
Tool Calls:
get_timeline
Args:
I'll summarize the key highlights from your Twitter timeline:
1. Professional/Tech Highlights:
- @ibamarief is preparing to onboard their first intern, noting it's a significant responsibility
- @papanberjalan shared a critique about web developers not understanding computer performance
- @pathofexile is running a Twitch Drops promotion for their game
2. Notable Tweets:
- @elonmusk posted a cryptic "Hmm"
- @realDonaldTrump discussed eliminating Daylight Saving Time
- @blinkist shared an article about billionaires reading up to 50 books per year
3. Miscellaneous Interesting Tweets:
- @nealtaparia shared a thread about 11 companies built around a single product
- @Rixhabh__ posted about creative and iconic ads
- Several tweets in Indonesian covering various topics from personal stories to social issues
Would you like me to elaborate on any of these tweets or provide more context about any specific post?
```
```
--------------------------------------------------------------------------------
/src/mcp_twikit/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "mcp-twikit"
version = "0.1.0"
description = "Twitter search tool for FastMCP using Twikit"
readme = "README.md"
requires-python = ">=3.7"
license = "MIT"
authors = [
{ name = "Adhika Setya Pramudita", email = "[email protected]" }
]
dependencies = [
"fastmcp",
"twikit",
"requests",
]
[project.urls]
Homepage = "https://github.com/adhikasp/mcp-twikit"
[project.scripts]
mcp-twikit = "mcp_twikit.twitter:mcp.run"
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
# Use a Python image
FROM python:3.8-slim
# Set working directory
WORKDIR /app
# Copy the project files
COPY . /app
# Install the project's dependencies
RUN pip install --no-cache-dir hatchling
RUN pip install --no-cache-dir .
# Set environment variables for Twitter authentication
# These should be provided at runtime for security purposes
ENV TWITTER_USERNAME "@example"
ENV TWITTER_EMAIL "[email protected]"
ENV TWITTER_PASSWORD "secret"
# Set the entrypoint command to run the MCP server
ENTRYPOINT ["mcp-twikit"]
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
required:
- twitterUsername
- twitterEmail
- twitterPassword
properties:
twitterUsername:
type: string
description: Your Twitter username.
twitterEmail:
type: string
description: Your Twitter email.
twitterPassword:
type: string
description: Your Twitter password.
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({ command: 'mcp-twikit', env: { TWITTER_USERNAME: config.twitterUsername, TWITTER_EMAIL: config.twitterEmail, TWITTER_PASSWORD: config.twitterPassword } })
```
--------------------------------------------------------------------------------
/src/mcp_twikit/twitter.py:
--------------------------------------------------------------------------------
```python
from fastmcp import FastMCP, Context
import twikit
import os
from pathlib import Path
import logging
from typing import Optional, List
import time
# Create an MCP server
mcp = FastMCP("mcp-twikit")
logger = logging.getLogger(__name__)
httpx_logger = logging.getLogger("httpx")
httpx_logger.setLevel(logging.WARNING)
USERNAME = os.getenv('TWITTER_USERNAME')
EMAIL = os.getenv('TWITTER_EMAIL')
PASSWORD = os.getenv('TWITTER_PASSWORD')
USER_AGENT = os.getenv('USER_AGENT')
COOKIES_PATH = Path.home() / '.mcp-twikit' / 'cookies.json'
# Rate limit tracking
RATE_LIMITS = {}
RATE_LIMIT_WINDOW = 15 * 60 # 15 minutes in seconds
async def get_twitter_client() -> twikit.Client:
"""Initialize and return an authenticated Twitter client."""
client = twikit.Client('en-US', user_agent=USER_AGENT)
if COOKIES_PATH.exists():
client.load_cookies(COOKIES_PATH)
else:
try:
await client.login(
auth_info_1=USERNAME,
auth_info_2=EMAIL,
password=PASSWORD
)
except Exception as e:
logger.error(f"Failed to login: {e}")
raise
COOKIES_PATH.parent.mkdir(parents=True, exist_ok=True)
client.save_cookies(COOKIES_PATH)
return client
def check_rate_limit(endpoint: str) -> bool:
"""Check if we're within rate limits for a given endpoint."""
now = time.time()
if endpoint not in RATE_LIMITS:
RATE_LIMITS[endpoint] = []
# Remove old timestamps
RATE_LIMITS[endpoint] = [t for t in RATE_LIMITS[endpoint] if now - t < RATE_LIMIT_WINDOW]
# Check limits based on endpoint
if endpoint == 'tweet':
return len(RATE_LIMITS[endpoint]) < 300 # 300 tweets per 15 minutes
elif endpoint == 'dm':
return len(RATE_LIMITS[endpoint]) < 1000 # 1000 DMs per 15 minutes
return True
# Existing search and read tools
@mcp.tool()
async def search_twitter(query: str, sort_by: str = 'Top', count: int = 10, ctx: Context = None) -> str:
"""Search twitter with a query. Sort by 'Top' or 'Latest'"""
try:
client = await get_twitter_client()
tweets = await client.search_tweet(query, product=sort_by, count=count)
return convert_tweets_to_markdown(tweets)
except Exception as e:
logger.error(f"Failed to search tweets: {e}")
return f"Failed to search tweets: {e}"
@mcp.tool()
async def get_user_tweets(username: str, tweet_type: str = 'Tweets', count: int = 10, ctx: Context = None) -> str:
"""Get tweets from a specific user's timeline."""
try:
client = await get_twitter_client()
username = username.lstrip('@')
user = await client.get_user_by_screen_name(username)
if not user:
return f"Could not find user {username}"
tweets = await client.get_user_tweets(
user_id=user.id,
tweet_type=tweet_type,
count=count
)
return convert_tweets_to_markdown(tweets)
except Exception as e:
logger.error(f"Failed to get user tweets: {e}")
return f"Failed to get user tweets: {e}"
@mcp.tool()
async def get_timeline(count: int = 20) -> str:
"""Get tweets from your home timeline (For You)."""
try:
client = await get_twitter_client()
tweets = await client.get_timeline(count=count)
return convert_tweets_to_markdown(tweets)
except Exception as e:
logger.error(f"Failed to get timeline: {e}")
return f"Failed to get timeline: {e}"
@mcp.tool()
async def get_latest_timeline(count: int = 20) -> str:
"""Get tweets from your home timeline (Following)."""
try:
client = await get_twitter_client()
tweets = await client.get_latest_timeline(count=count)
return convert_tweets_to_markdown(tweets)
except Exception as e:
logger.error(f"Failed to get latest timeline: {e}")
return f"Failed to get latest timeline: {e}"
# New write tools
@mcp.tool()
async def post_tweet(
text: str,
media_paths: Optional[List[str]] = None,
reply_to: Optional[str] = None,
tags: Optional[List[str]] = None
) -> str:
"""Post a tweet with optional media, reply, and tags."""
try:
if not check_rate_limit('tweet'):
return "Rate limit exceeded for tweets. Please wait before posting again."
client = await get_twitter_client()
# Handle tags by converting to mentions
if tags:
mentions = ' '.join(f"@{tag.lstrip('@')}" for tag in tags)
text = f"""{text}
{mentions}"""
# Upload media if provided
media_ids = []
if media_paths:
for path in media_paths:
media_id = await client.upload_media(path, wait_for_completion=True)
media_ids.append(media_id)
# Create the tweet
tweet = await client.create_tweet(
text=text,
media_ids=media_ids if media_ids else None,
reply_to=reply_to
)
RATE_LIMITS.setdefault('tweet', []).append(time.time())
return f"Successfully posted tweet: {tweet.id}"
except Exception as e:
logger.error(f"Failed to post tweet: {e}")
return f"Failed to post tweet: {e}"
@mcp.tool()
async def delete_tweet(tweet_id: str) -> str:
"""Delete a tweet by its ID."""
try:
client = await get_twitter_client()
await client.delete_tweet(tweet_id)
return f"Successfully deleted tweet {tweet_id}"
except Exception as e:
logger.error(f"Failed to delete tweet: {e}")
return f"Failed to delete tweet: {e}"
@mcp.tool()
async def send_dm(user_id: str, message: str, media_path: Optional[str] = None) -> str:
"""Send a direct message to a user."""
try:
if not check_rate_limit('dm'):
return "Rate limit exceeded for DMs. Please wait before sending again."
client = await get_twitter_client()
media_id = None
if media_path:
media_id = await client.upload_media(media_path, wait_for_completion=True)
await client.send_dm(
user_id=user_id,
text=message,
media_id=media_id
)
RATE_LIMITS.setdefault('dm', []).append(time.time())
return f"Successfully sent DM to user {user_id}"
except Exception as e:
logger.error(f"Failed to send DM: {e}")
return f"Failed to send DM: {e}"
@mcp.tool()
async def delete_dm(message_id: str) -> str:
"""Delete a direct message by its ID."""
try:
client = await get_twitter_client()
await client.delete_dm(message_id)
return f"Successfully deleted DM {message_id}"
except Exception as e:
logger.error(f"Failed to delete DM: {e}")
return f"Failed to delete DM: {e}"
def convert_tweets_to_markdown(tweets) -> str:
"""Convert a list of tweets to markdown format."""
result = []
for tweet in tweets:
result.append(f"### @{tweet.user.screen_name}")
result.append(f"**{tweet.created_at}**")
result.append(tweet.text)
if tweet.media:
for media in tweet.media:
result.append(f"")
result.append("---")
return "\n".join(result)
```