echo_server.py

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()