🚀 Claude用Twitterトレンド分析MCPサーバーの構築
このチュートリアルでは、TwitterのリアルタイムトレンドとClaudeの分析機能を接続するModel Context Protocol (MCP)サーバーの作成方法を説明します。サーバーはリアルタイムのTwitterトレンドを取得し、Claudeを使用してビジネスチャンスを分析します。
🚀 クイックスタート
このセクションでは、Claude用のTwitterトレンド分析MCPサーバーを構築するための手順を説明します。
✨ 主な機能
- Twitterのリアルタイムトレンドを取得します。
- Claudeを使用してビジネスチャンスを分析します。
- データの保存、可視化、自動分析などの拡張機能を提供します。
📦 インストール
前提条件
- Python 3.8以上
- Claude Desktopがインストールされていること
- APIアクセス権限を持つTwitter開発者アカウント
- Pythonの基本的な知識
環境のセットアップ
- 新しいプロジェクトディレクトリを作成します。
mkdir twitter-trends-mcp
cd twitter-trends-mcp
- 仮想環境を設定します。
python -m venv .venv
.venv\Scripts\activate
- 必要なパッケージをインストールします。
pip install tweepy mcp python-dotenv hatchling
プロジェクト構造の作成
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"
}
}
Claude Desktopでのサーバーの設定
-
Claude Desktopを開きます。
-
プラグインアイコン(通常はコンセントの記号)をクリックします。
-
「MCPサーバーを追加」またはそれに相当するオプションを選択し、新しいサーバーインスタンスを設定します。
-
以下の情報を入力します。
- 名前:
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
-
すべてのパスが正しいことを確認し、設定を保存します。
実行とテスト
- プロジェクトのルートディレクトリでパッケージをインストールします。
pip install -e .
- サーバーを起動します。
python server.py twitter_config.json
-
Claude Desktopで:
- プラグインアイコンをクリックします。
- 「twitter-trends」インスタンスを探します。
- 以下を入力してみます:
Analyze current Twitter trends for SaaS opportunities
-
ログファイルを監視します。
Get-Content twitter_server.log
💻 使用例
基本的な使用法
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("Twitter API の認証に成功しました。")
except Exception as e:
logger.error(f"認証に失敗しました: {e}")
raise
def get_trends(self):
if not self.authenticated:
logger.error("認証されていません。トレンドを取得できません。")
return []
try:
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"トレンドの取得に失敗しました: {e}")
return []
def _get_api_instance(self):
pass
def main():
monitor = TwitterTrendMonitor('twitter_config.json')
try:
monitor.authenticate()
trends = monitor.get_trends()
logger.info(f"現在のトレンドトピック: {trends}")
except Exception as e:
logger.error(f"メインの実行に失敗しました: {e}")
if __name__ == "__main__":
main()
高度な使用法
高度な機能を追加することで、サーバーの機能を拡張できます。以下にいくつかの例を示します。
データの保存
取得したトレンドデータをデータベースに保存して、後で分析できるようにします。
手順:
- SQLite、MySQL、PostgreSQLなどのデータベースを使用します。
server.py
の get_trending_topics
メソッドを変更してデータを記録します。
- データの損失を防ぐために定期的なバックアップ機構を追加します。
データの可視化
可視化ツール(PlotlyやMatplotlibなど)を統合して、トレンドの変化をリアルタイムで表示します。
手順:
server.py
にプロット機能を追加します。
- 定期的にグラフを生成して画像ファイルとして保存するか、Webインターフェイスで表示します。
自動分析
自然言語処理ライブラリ(spaCyやnltkなど)を使用して、トレンドテキストに感情分析やその他のタイプの分析を行います。
手順:
- 必要なNLPライブラリをインポートします。
- 取得したトレンドデータに分析モデルを適用します。
- 分析結果を記録して可視化します。
拡張性とパフォーマンスの最適化
大量のデータを処理する必要がある場合は、以下の最適化策を検討してください。
- 非同期リクエスト:
aiohttp
や tweepy
などの非同期ライブラリを使用して、データの取得速度を向上させます。
- 分散コンピューティング:複数のノードにタスクを分散して、処理能力を向上させます。
- キューシステム:RabbitMQやRedisなどを使用してタスクキューを管理し、システムの拡張性を確保します。
🔧 技術詳細
このサーバーは、Twitter APIを使用してリアルタイムのトレンドデータを取得し、Claudeを使用してビジネスチャンスを分析します。Pythonのライブラリであるtweepyを使用してTwitter APIと通信し、loggingライブラリを使用してログを記録します。また、MCPプロトコルを使用してClaude Desktopと通信します。
📚 ドキュメント
エラーのトラブルシューティングとデバッグ
認証問題
Twitter APIの認証に失敗した場合:
api_key
と api_secret
が正しいことを確認します。
- 開発者アカウントとアプリケーションの権限がリアルタイムトレンドデータにアクセスできることを確認します。
- Twitter Developer Consoleを確認して最新のAPIキーを取得します。
ネットワーク問題
Twitter APIに接続できない場合:
- ネットワーク接続を確認します。
- 必要に応じてプロキシサーバーを使用します。
- IPがブロックされていないか確認します。
ログ分析
twitter_server.log
ファイルのエラーメッセージを注意深く確認して、具体的なエラー原因と位置を特定します。
設定問題
すべてのパスと設定パラメータが正しいことを確認します。特にWindows環境では、パス区切り文字をバックスラッシュに調整するか、生文字列(raw string)を使用する必要があります。
セキュリティに関する注意事項
- APIキーのセキュリティ:APIキーをバージョン管理システムに平文で保存しないでください。環境変数または暗号化された設定ファイルを使用して機密情報を管理します。
- ログの監視:ログファイルを定期的にレビューして、未承認のアクセスなどの潜在的なセキュリティイベントを防止します。
- 依存ライブラリの定期的な更新:プロジェクトのサードパーティライブラリを最新バージョンに保ち、既知のセキュリティホールを回避します。
参考文献と拡張読書
📄 ライセンス
原ドキュメントにライセンス情報が記載されていないため、このセクションは省略されています。
プロジェクトのまとめ
以上の手順に従って、Twitter APIを使用したリアルタイムトレンド監視サーバーを構築し、Claude Desktopで設定することができました。必要に応じて、データの保存、可視化分析などの機能を拡張することができます。同時に、システムのセキュリティとパフォーマンスの最適化にも注意を払ってください。
何か問題がある場合やさらなる支援が必要な場合は、関連ドキュメントを参照するか、コミュニティのサポートを求めてください。