Twitter MCP Server For Claude

🚀 構建用於Claude的Twitter趨勢分析MCP服務器

本教程將指導你創建一個Model Context Protocol (MCP)服務器，該服務器可連接Twitter即時熱門話題與Claude的分析功能。服務器會獲取即時Twitter趨勢，並利用Claude進行商業機會分析。

🚀 快速開始

先決條件

• Python 3.8或更高版本
• 安裝了Claude Desktop
• 具有API訪問權限的Twitter開發賬戶
• 具備基本的Python知識

📦 安裝指南

第一步：安裝必要的庫

確保你已經安裝了所需的Python庫：

pip install tweepy python-dotenv logging

第二步：創建項目結構

在你的工作目錄中創建以下文件和文件夾結構：

twitter-trends-mcp/
├── src/
│   ├── twitter_trends_mcp/
│   │   ├── __init__.py
│   │   └── server.py
│   └── twitter_config.json
└── twitter_server.log

第三步：配置 Twitter API 信息

在 twitter_config.json 文件中填寫你的Twitter開發者賬號的API密鑰：

{
    "twitter": {
        "api_key": "your_api_key",
        "api_secret": "your_api_secret"
    },
    "logging": {
        "level": "INFO",
        "file": "twitter_server.log"
    }
}

第四步：安裝項目依賴

在項目根目錄中安裝包：

pip install -e .

💻 使用示例

基礎用法

編寫服務器代碼

在 server.py 中實現 Twitter API 的認證和趨勢獲取功能：

import tweepy
from tweepy import OAuthHandler
import json
import logging

# 配置日誌
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('twitter_server.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

class TwitterTrendMonitor:
    def __init__(self, config_file):
        self.config = self._load_config(config_file)
        self.authenticated = False

    def _load_config(self, config_file):
        with open(config_file) as f:
            return json.load(f)

    def authenticate(self):
        try:
            auth = OAuthHandler(
                self.config['twitter']['api_key'],
                self.config['twitter']['api_secret']
            )
            self.authenticated = True
            logger.info("Successfully authenticated with Twitter API.")
        except Exception as e:
            logger.error(f"Authentication failed: {e}")
            raise

    def get_trends(self):
        if not self.authenticated:
            logger.error("Not authenticated. Cannot fetch trends.")
            return []

        try:
            # 假設你已經擁有API實例api
            api = self._get_api_instance()
            trends = api.trends_place(id=1)  # 獲取美國地區的趨勢
            return [trend.name for trend in trends[0]['trends']]
        except Exception as e:
            logger.error(f"Failed to fetch trends: {e}")
            return []

    def _get_api_instance(self):
        # 實現獲取API實例的方法，可能需要處理OAuth token等
        pass

def main():
    monitor = TwitterTrendMonitor('twitter_config.json')
    try:
        monitor.authenticate()
        trends = monitor.get_trends()
        logger.info(f"Current trending topics: {trends}")
    except Exception as e:
        logger.error(f"Main execution failed: {e}")

if __name__ == "__main__":
    main()

運行服務器

在終端或命令提示符中運行以下命令來啟動服務器：

python src/twitter_trends_mcp/server.py

監控日誌

查看生成的日誌文件 twitter_server.log 以確保程序正常運行：

tail -f twitter_server.log  # 在Linux/MacOS中使用
# 或者在Windows PowerShell中使用 Get-Content twitter_server.log -Wait

高級用法

配置Claude Desktop

1. 打開Claude Desktop。
2. 點擊插件圖標（通常顯示為一個插座符號）。
3. 選擇“添加MCP服務器”或相應選項以配置新的服務器實例。
4. 填寫以下信息：
   • 名稱：twitter-trends
   • 命令：C:\\Users\\YOUR_USERNAME\\twitter-trends-mcp\\.venv\\Scripts\\python.exe
   • 參數：C:\\Users\\YOUR_USERNAME\\twitter-trends-mcp\\server.py C:\\Users\\YOUR_USERNAME\\twitter-trends-mcp\\twitter_config.json
   • 環境變量：
     • PYTHONPATH: C:\\Users\\YOUR_USERNAME\\twitter-trends-mcp\\src
     • PYTHONUNBUFFERED: 1
   • 工作目錄: C:\\Users\\YOUR_USERNAME\\twitter-trends-mcp
5. 確保所有路徑正確無誤，然後保存配置。

在Claude Desktop中測試

• 點擊插件圖標。
• 查找“twitter-trends”實例。
• 嘗試輸入：Analyze current Twitter trends for SaaS opportunities

🔧 技術細節

項目結構

twitter-trends-mcp/
├── pyproject.toml
├── twitter_server_run.py
├── src/
│   └── twitter_trends_mcp/
│       ├── __init__.py
│       └── server.py

配置文件

pyproject.toml

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "twitter-trends-mcp"
version = "0.1.0"
description = "Twitter Trends MCP Server"
requires-python = ">=3.8"
dependencies = [
    "tweepy",
    "mcp",
    "python-dotenv",
    "hatchling"
]

[project.urls]
Homepage = "https://github.com/yourusername/twitter-trends-mcp"

[project.scripts]
console_scripts = [
    "twitter-trends-mcp=twitter_trends_mcp.server:main"
]

src/twitter_trends_mcp/server.py

import tweepy
from tweepy import OAuthHandler
import json
import logging

# 配置日誌
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='twitter_server.log'
)
logger = logging.getLogger(__name__)

class TwitterTrendServer:
    def __init__(self, config_path):
        self.config = self.load_config(config_path)
        self.authenticated_api = None

    def load_config(self, config_path):
        with open(config_path, 'r') as f:
            return json.load(f)

    def authenticate_twitter_api(self):
        try:
            auth = OAuthHandler(
                self.config['twitter']['api_key'],
                self.config['twitter']['api_secret']
            )
            api = tweepy.API(auth)
            logger.info("Twitter API 已成功認證")
            return api
        except Exception as e:
            logger.error(f"Twitter API 認證失敗: {str(e)}")
            raise

    def get_trending_topics(self, woe_id=1):
        try:
            if self.authenticated_api is None:
                self.authenticate_twitter_api()
            trends = self.authenticated_api.trends_place(woe_id)
            return [trend['name'] for trend in trends[0]['trends']]
        except Exception as e:
            logger.error(f"獲取趨勢失敗: {str(e)}")
            return []

    def run_server(self):
        while True:
            try:
                # 示例：即時獲取並打印趨勢
                trends = self.get_trending_topics()
                if trends:
                    logger.info("當前Twitter熱門話題:")
                    for trend in trends:
                        logger.info(f"- {trend}")
                else:
                    logger.warning("未能獲取到任何趨勢。請檢查日誌文件以獲取更多信息。")
            except KeyboardInterrupt:
                logger.info("服務器已停止。")
                break
            except Exception as e:
                logger.error(f"發生錯誤: {str(e)}")
                continue

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("使用方式：python server.py <配置文件路徑>")
        sys.exit(1)
    config_path = sys.argv[1]
    server = TwitterTrendServer(config_path)
    server.run_server()

src/twitter_trends_mcp/init.py

from .server import TwitterTrendServer

__version__ = "0.1.0"

twitter_config.json

{
    "twitter": {
        "api_key": "your_api_key",
        "api_secret": "your_api_secret"
    },
    "logging": {
        "level": "INFO",
        "file": "twitter_server.log"
    }
}

📚 詳細文檔

高級功能和擴展

1. 數據存儲

可以將獲取的趨勢數據存儲到數據庫中，以便後續分析。 步驟：

• 使用如SQLite、MySQL或PostgreSQL等數據庫。
• 修改 server.py 中的 get_trending_topics 方法以記錄數據。
• 添加定期備份機制以防數據丟失。

2. 數據可視化

集成可視化工具（如Plotly或Matplotlib）來即時顯示趨勢變化。 步驟：

• 在 server.py 中添加繪圖功能。
• 定期生成圖表並保存為圖片文件，或者通過Web界面展示。

3. 自動化分析

使用自然語言處理庫（如spaCy或nltk）對趨勢文本進行情感分析或其他類型分析。 步驟：

• 導入必要的NLP庫。
• 在獲取的趨勢數據上應用分析模型。
• 記錄並可視化分析結果。

4. 可擴展性和性能優化

如果需要處理大量數據，可以考慮以下優化措施：

• 異步請求：使用 asynchronous libraries like aiohttp 和 tweepy 來提高數據獲取速度。
• 分佈式計算：將任務分擔到多個節點上以提升處理能力。
• 隊列系統：使用如 RabbitMQ 或 Redis 來管理任務隊列，確保系統的可擴展性。

錯誤排查和調試

1. 認證問題

如果 Twitter API 認證失敗：

• 檢查 api_key 和 api_secret 是否正確。
• 確保你的開發者賬號和應用權限允許訪問即時趨勢數據。
• 查看Twitter Developer Console以獲取最新API密鑰。

2. 網絡問題

如果無法連接到 Twitter API：

• 檢查網絡連接。
• 使用代理服務器（如需要）。
• 查看是否有IP被封禁的情況。

3. 日誌分析

仔細查看 twitter_server.log 文件中的錯誤信息，獲取具體的報錯原因和位置。

4. 配置問題

確保所有路徑和配置參數正確無誤。特別是在 Windows 環境下，路徑分隔符可能需要調整為反斜槓或使用原始字符串。

安全注意事項

• API密鑰的安全性：不要在版本控制系統中明文存儲 API 密鑰。可以使用環境變量或加密配置文件來管理敏感信息。
• 日誌監控：及時審查日誌文件，防止潛在的安全事件如未授權訪問等。
• 定期更新依賴庫：保持項目的第三方庫為最新版本，以避免已知的安全漏洞。

📄 參考文獻和擴展閱讀

• Twitter API 官方文檔
• Tweepy 庫文檔
• Python 日誌庫文檔
• [MCP 插件開發指南（如果適用）]
• [即時數據分析的最佳實踐]

📚 項目總結

通過以上步驟，你已經成功搭建了一個基於 Twitter API 的即時趨勢監控服務器，並在 Claude Desktop 中進行了配置。你可以根據具體需求擴展功能，如數據存儲、可視化分析等，同時注意系統的安全性和性能優化。

如果有任何問題或需要進一步的幫助，請參考相關文檔或尋求社區支持。