Spaces:
Sleeping
Sleeping
File size: 5,551 Bytes
700465c 532ff49 e7206f6 532ff49 80dc3a4 532ff49 700465c 532ff49 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | from mcp.server.fastmcp import FastMCP
from fastmcp import Client
from fastmcp.client.auth import OAuth
import os
from typing import Optional, Dict, Any
mcp = FastMCP(name='EchoServer', stateless_http=True)
# Twitter OAuth configuration
TWITTER_CLIENT_ID = os.getenv('TWITTER_CLIENT_ID')
TWITTER_CLIENT_SECRET = os.getenv('TWITTER_CLIENT_SECRET')
TWITTER_REDIRECT_URI = os.getenv(
'TWITTER_REDIRECT_URI', 'http://localhost:8080/callback'
)
# Global Twitter client instance
twitter_client: Optional[Client] = None
async def get_twitter_client() -> Client:
"""Get authenticated Twitter client using OAuth"""
global twitter_client
if twitter_client is None:
# Configure OAuth for Twitter
oauth = OAuth(
mcp_url='https://api.twitter.com/2', # Twitter API v2 endpoint
scopes=['tweet.read', 'tweet.write', 'users.read'],
client_name='FastMCP Twitter Client',
additional_client_metadata={
'client_id': TWITTER_CLIENT_ID,
'client_secret': TWITTER_CLIENT_SECRET,
'redirect_uri': TWITTER_REDIRECT_URI
}
)
twitter_client = Client('https://api.twitter.com/2', auth=oauth)
await twitter_client.__aenter__()
return twitter_client
@mcp.tool(description='A simple echo tool')
def echo(message: str) -> str:
return f'Echo: {message}'
@mcp.tool(description='Authenticate with Twitter using OAuth 2.1')
async def authenticate_twitter() -> Dict[str, Any]:
"""Authenticate with Twitter and return authentication status"""
try:
client = await get_twitter_client()
# Test the connection by making a simple API call
response = await client.request('GET', '/users/me')
return {
'status': 'success',
'message': 'Successfully authenticated with Twitter',
'user_data': (
response.json() if hasattr(response, 'json') else str(response)
)
}
except Exception as e:
return {
'status': 'error',
'message': f'Failed to authenticate with Twitter: {str(e)}',
'error': str(e)
}
@mcp.tool(description='Post a tweet using Twitter API v2')
async def post_tweet(text: str) -> Dict[str, Any]:
"""Post a tweet to Twitter"""
try:
client = await get_twitter_client()
# Twitter API v2 endpoint for posting tweets
tweet_data = {
'text': text
}
response = await client.request('POST', '/tweets', json=tweet_data)
return {
'status': 'success',
'message': 'Tweet posted successfully',
'tweet_id': (
response.json().get('data', {}).get('id')
if hasattr(response, 'json') else None
),
'response': (
response.json() if hasattr(response, 'json') else str(response)
)
}
except Exception as e:
return {
'status': 'error',
'message': f'Failed to post tweet: {str(e)}',
'error': str(e)
}
@mcp.tool(description="Get user's Twitter profile information")
async def get_twitter_profile() -> Dict[str, Any]:
"""Get the authenticated user's Twitter profile"""
try:
client = await get_twitter_client()
# Get user profile information
response = await client.request(
'GET', '/users/me?user.fields=id,name,username,public_metrics'
)
return {
'status': 'success',
'message': 'Profile retrieved successfully',
'profile': (
response.json() if hasattr(response, 'json') else str(response)
)
}
except Exception as e:
return {
'status': 'error',
'message': f'Failed to get profile: {str(e)}',
'error': str(e)
}
@mcp.tool(description='Search for tweets using Twitter API v2')
async def search_tweets(query: str, max_results: int = 10) -> Dict[str, Any]:
"""Search for tweets on Twitter"""
try:
client = await get_twitter_client()
# Twitter API v2 search endpoint
params = {
'query': query,
'max_results': min(max_results, 100), # Twitter API limit
'tweet.fields': 'created_at,author_id,public_metrics'
}
response = await client.request(
'GET', '/tweets/search/recent', params=params
)
return {
'status': 'success',
'message': f'Found tweets for query: {query}',
'tweets': (
response.json() if hasattr(response, 'json') else str(response)
)
}
except Exception as e:
return {
'status': 'error',
'message': f'Failed to search tweets: {str(e)}',
'error': str(e)
}
# Example usage functions
async def example_twitter_oauth_usage():
"""Example of how to use Twitter OAuth with FastMCP"""
# Method 1: Using default OAuth settings
async with Client('https://api.twitter.com/2', auth='oauth') as client:
await client.ping()
# Method 2: Using OAuth helper with custom configuration
oauth = OAuth(
mcp_url='https://api.twitter.com/2',
scopes=['tweet.read', 'tweet.write', 'users.read'],
client_name='FastMCP Twitter Client'
)
async with Client('https://api.twitter.com/2', auth=oauth) as client:
await client.ping() |