twitter_oauth_example.py

#!/usr/bin/env python3
"""
Twitter OAuth Example using FastMCP

This script demonstrates how to use Twitter OAuth authentication with FastMCP.
It shows both the simple "oauth" string method and the advanced OAuth helper method.

Before running this script:
1. Set up your Twitter Developer account and create an app
2. Set the required environment variables (see twitter_config.py)
3. Install dependencies: pip install -r requirements.txt
"""

import asyncio
import os
from dotenv import load_dotenv
from fastmcp import Client
from fastmcp.client.auth import OAuth
from twitter_config import twitter_config

# Load environment variables from .env file
load_dotenv()

async def example_simple_oauth():
    """Example using simple OAuth string configuration"""
    print("=== Simple OAuth Example ===")
    
    try:
        # Method 1: Using default OAuth settings
        async with Client("https://api.twitter.com/2", auth="oauth") as client:
            print("✓ Successfully connected to Twitter API with simple OAuth")
            
            # Test the connection
            response = await client.request("GET", "/users/me")
            print(f"✓ User data: {response.json() if hasattr(response, 'json') else str(response)}")
            
    except Exception as e:
        print(f"✗ Simple OAuth failed: {e}")

async def example_advanced_oauth():
    """Example using advanced OAuth helper configuration"""
    print("\n=== Advanced OAuth Example ===")
    
    # Validate configuration
    if not twitter_config.validate():
        missing = twitter_config.get_missing_config()
        print(f"✗ Missing configuration: {', '.join(missing)}")
        print("Please set the required environment variables (see twitter_config.py)")
        return
    
    try:
        # Method 2: Using OAuth helper with custom configuration
        oauth = OAuth(
            mcp_url=twitter_config.oauth_server_url,
            scopes=["tweet.read", "tweet.write", "users.read"],
            client_name="FastMCP Twitter Client",
            additional_client_metadata={
                "client_id": twitter_config.client_id,
                "client_secret": twitter_config.client_secret,
                "redirect_uri": twitter_config.redirect_uri
            }
        )
        
        async with Client(twitter_config.oauth_server_url, auth=oauth) as client:
            print("✓ Successfully connected to Twitter API with advanced OAuth")
            
            # Test the connection
            response = await client.request("GET", "/users/me")
            print(f"✓ User data: {response.json() if hasattr(response, 'json') else str(response)}")
            
            # Example: Post a tweet
            print("\n--- Posting a test tweet ---")
            tweet_data = {
                "text": "Hello from FastMCP Twitter OAuth! 🐦"
            }
            tweet_response = await client.request("POST", "/tweets", json=tweet_data)
            print(f"✓ Tweet posted: {tweet_response.json() if hasattr(tweet_response, 'json') else str(tweet_response)}")
            
    except Exception as e:
        print(f"✗ Advanced OAuth failed: {e}")

async def example_using_mcp_tools():
    """Example using the MCP tools from echo_server.py"""
    print("\n=== MCP Tools Example ===")
    
    try:
        # Import the MCP tools
        from echo_server import authenticate_twitter, post_tweet, get_twitter_profile, search_tweets
        
        # Authenticate
        auth_result = await authenticate_twitter()
        print(f"Authentication result: {auth_result}")
        
        if auth_result["status"] == "success":
            # Get profile
            profile_result = await get_twitter_profile()
            print(f"Profile result: {profile_result}")
            
            # Search tweets
            search_result = await search_tweets("FastMCP", max_results=5)
            print(f"Search result: {search_result}")
            
            # Post a tweet (uncomment to actually post)
            # tweet_result = await post_tweet("Testing FastMCP Twitter OAuth integration! 🚀")
            # print(f"Tweet result: {tweet_result}")
        
    except Exception as e:
        print(f"✗ MCP tools example failed: {e}")

async def main():
    """Main function to run all examples"""
    print("Twitter OAuth with FastMCP Examples")
    print("=" * 50)
    
    # Check if configuration is available
    if not twitter_config.validate():
        print("⚠️  Twitter configuration not found!")
        print("Please set the following environment variables:")
        for var in twitter_config.get_missing_config():
            print(f"  - {var}")
        print("\nSee twitter_config.py for more details.")
        return
    
    print("✓ Twitter configuration found")
    print(f"  Client ID: {twitter_config.client_id[:10]}..." if twitter_config.client_id else "Not set")
    print(f"  Redirect URI: {twitter_config.redirect_uri}")
    print(f"  OAuth Server: {twitter_config.oauth_server_url}")
    
    # Run examples
    await example_simple_oauth()
    await example_advanced_oauth()
    await example_using_mcp_tools()
    
    print("\n" + "=" * 50)
    print("Examples completed!")

if __name__ == "__main__":
    asyncio.run(main())