fix: oauth

TWITTER_OAUTH_README.md

@@ -0,0 +1,177 @@
+# Twitter OAuth Integration with FastMCP
+
+This project demonstrates how to integrate Twitter OAuth 2.1 authentication with FastMCP using the Model Context Protocol (MCP).
+
+## Features
+
+- **Twitter OAuth 2.1 Authentication**: Secure authentication using OAuth 2.1 with PKCE
+- **Twitter API v2 Integration**: Full access to Twitter's API v2 endpoints
+- **MCP Tools**: Ready-to-use tools for common Twitter operations
+- **Token Management**: Automatic token caching and refresh
+- **FastMCP Integration**: Seamless integration with FastMCP's OAuth helper
+
+## Available Tools
+
+The Echo server now includes the following Twitter OAuth tools:
+
+1. **`authenticate_twitter()`** - Authenticate with Twitter using OAuth 2.1
+2. **`post_tweet(text: str)`** - Post tweets to Twitter
+3. **`get_twitter_profile()`** - Get user's Twitter profile information
+4. **`search_tweets(query: str, max_results: int)`** - Search for tweets on Twitter
+
+## Setup Instructions
+
+### 1. Twitter Developer Account Setup
+
+1. Go to [Twitter Developer Portal](https://developer.twitter.com/)
+2. Create a new app or use an existing one
+3. Note down your **Client ID** and **Client Secret**
+4. Set up a redirect URI (e.g., `http://localhost:8080/callback`)
+
+### 2. Environment Configuration
+
+Create a `.env` file in the project root with the following variables:
+
+```bash
+# Twitter OAuth Configuration
+TWITTER_CLIENT_ID=your_twitter_client_id_here
+TWITTER_CLIENT_SECRET=your_twitter_client_secret_here
+TWITTER_REDIRECT_URI=http://localhost:8080/callback
+
+# Optional: Custom OAuth server URL
+TWITTER_OAUTH_SERVER_URL=https://api.twitter.com/2
+```
+
+### 3. Install Dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+### 4. Run the Server
+
+```bash
+python server.py
+```
+
+The server will be available at `http://localhost:10000` with the Echo MCP server (including Twitter OAuth) at `/echo/mcp`.
+
+## Usage Examples
+
+### Method 1: Simple OAuth (Default Settings)
+
+```python
+from fastmcp import Client
+
+# Uses default OAuth settings
+async with Client("https://api.twitter.com/2", auth="oauth") as client:
+    await client.ping()
+```
+
+### Method 2: Advanced OAuth (Custom Configuration)
+
+```python
+from fastmcp import Client
+from fastmcp.client.auth import OAuth
+
+# Configure OAuth with custom settings
+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()
+```
+
+### Method 3: Using MCP Tools
+
+```python
+from echo_server import authenticate_twitter, post_tweet, get_twitter_profile, search_tweets
+
+# Authenticate with Twitter
+auth_result = await authenticate_twitter()
+print(auth_result)
+
+# Get user profile
+profile = await get_twitter_profile()
+print(profile)
+
+# Search for tweets
+tweets = await search_tweets("FastMCP", max_results=10)
+print(tweets)
+
+# Post a tweet
+tweet_result = await post_tweet("Hello from FastMCP Twitter OAuth! 🐦")
+print(tweet_result)
+```
+
+## Running the Example Script
+
+A complete example script is provided in `twitter_oauth_example.py`:
+
+```bash
+python twitter_oauth_example.py
+```
+
+This script demonstrates all three methods of using Twitter OAuth with FastMCP.
+
+## OAuth Flow
+
+The OAuth flow follows these steps:
+
+1. **Token Check**: Check for existing valid tokens in the cache
+2. **OAuth Server Discovery**: Discover Twitter's OAuth endpoints
+3. **Dynamic Client Registration**: Register the client with Twitter (if needed)
+4. **Local Callback Server**: Start a temporary local server for OAuth callback
+5. **Browser Interaction**: Open user's browser for authentication
+6. **Authorization Code Exchange**: Exchange authorization code for access token
+7. **Token Caching**: Save tokens for future use
+8. **Authenticated Requests**: Use access token for API requests
+9. **Token Refresh**: Automatically refresh expired tokens
+
+## Token Management
+
+- Tokens are automatically cached in `~/.fastmcp/oauth-mcp-client-cache/`
+- Tokens persist between application runs
+- Automatic token refresh when expired
+- Clear tokens using the `FileTokenStorage` class if needed
+
+## Error Handling
+
+All tools return structured responses with:
+- `status`: "success" or "error"
+- `message`: Human-readable message
+- `data`: Response data (on success) or error details (on failure)
+
+## Security Notes
+
+- Never commit your `.env` file to version control
+- Use environment variables for sensitive configuration
+- The OAuth flow uses PKCE for enhanced security
+- Tokens are stored securely in the user's home directory
+
+## Troubleshooting
+
+### Common Issues
+
+1. **Missing Environment Variables**: Ensure all required environment variables are set
+2. **Invalid Client Credentials**: Verify your Twitter app credentials
+3. **Redirect URI Mismatch**: Ensure the redirect URI matches your Twitter app configuration
+4. **Network Issues**: Check your internet connection and firewall settings
+
+### Debug Mode
+
+Enable debug logging by setting the environment variable:
+```bash
+export FASTMCP_DEBUG=1
+```
+
+## API Reference
+
+For more details about FastMCP OAuth, see the [official documentation](https://gofastmcp.com/clients/auth/oauth).
+
+## License
+
+This project is part of the FastMCP ecosystem and follows the same licensing terms.

echo_server.py

@@ -1,7 +1,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)
+mcp = FastMCP(name='EchoServer', stateless_http=True)
 
-@mcp.tool(description="A simple echo tool")
+# 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}"
+    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()

requirements.txt

@@ -3,4 +3,5 @@ httpx>=0.27.0
 pydantic>=2.7.0
 selectolax>=0.3.15 
 fastapi
-jinja2
+jinja2
+python-dotenv>=1.0.0

templates/index.html

@@ -46,13 +46,13 @@
       <div class="card">
         <span class="badge">FastAPI • MCP</span>
         <h1>Host multiple MCP servers on a single app</h1>
-        <p class="subtitle">This template mounts multiple Model Context Protocol (MCP) servers under one FastAPI instance.</p>
+        <p class="subtitle">This template mounts multiple Model Context Protocol (MCP) servers under one FastAPI instance, including Twitter OAuth integration.</p>
 
         <div class="grid">
           <div class="tile">
             <h3>Available servers</h3>
             <ul class="muted">
-              <li><a href="/echo/mcp" target="_blank" rel="noopener noreferrer">{{ base_url }}/echo/mcp</a> — Echo MCP server</li>
+              <li><a href="/echo/mcp" target="_blank" rel="noopener noreferrer">{{ base_url }}/echo/mcp</a> — Echo MCP server with Twitter OAuth</li>
               <li><a href="/math/mcp" target="_blank" rel="noopener noreferrer">{{ base_url }}/math/mcp</a> — Math MCP server</li>
             </ul>
           </div>
@@ -62,9 +62,19 @@
           </div>
         </div>
 
+        <h3 style="margin:16px 0 8px">Twitter OAuth Integration</h3>
+        <p class="muted" style="margin:0 0 12px">The Echo server now includes Twitter OAuth 2.1 authentication with the following tools:</p>
+        <ul class="muted" style="margin:0 0 12px 18px">
+          <li><strong>authenticate_twitter</strong> — Authenticate with Twitter using OAuth 2.1</li>
+          <li><strong>post_tweet</strong> — Post tweets to Twitter</li>
+          <li><strong>get_twitter_profile</strong> — Get user's Twitter profile information</li>
+          <li><strong>search_tweets</strong> — Search for tweets on Twitter</li>
+        </ul>
+        <p class="muted" style="margin:0 0 12px">Configure environment variables: <code>TWITTER_CLIENT_ID</code>, <code>TWITTER_CLIENT_SECRET</code>, <code>TWITTER_REDIRECT_URI</code></p>
+
         <h3 style="margin:16px 0 8px">How to get your {base_URL}</h3>
         <ol class="muted" style="margin:0 0 12px 18px">
-          <li>Open your Space and click <strong>“Embed this Space”</strong>.</li>
+          <li>Open your Space and click <strong>"Embed this Space"</strong>.</li>
           <li>Copy the <strong>iframe</strong> code and take the value of the <strong>src</strong> attribute.</li>
           <li>That origin (e.g. <code>https://your-space.hf.space</code>) is your <code>{base_URL}</code>.</li>
         </ol>
@@ -73,8 +83,21 @@
 
 base_URL = {{ base_url }}
 
-Echo MCP  = {{ base_url }}/echo/mcp
-Math MCP  = {{ base_url }}/math/mcp
+Echo MCP (with Twitter OAuth) = {{ base_url }}/echo/mcp
+Math MCP                    = {{ base_url }}/math/mcp
+
+# Twitter OAuth Example
+from fastmcp import Client
+from fastmcp.client.auth import OAuth
+
+# Simple OAuth
+async with Client("https://api.twitter.com/2", auth="oauth") as client:
+    await client.ping()
+
+# Advanced OAuth
+oauth = OAuth(mcp_url="https://api.twitter.com/2")
+async with Client("https://api.twitter.com/2", auth=oauth) as client:
+    await client.ping()
 </code></pre>
 
         <p class="muted" style="margin:10px 0 8px">Illustration of the “Embed this Space” dialog:</p>

twitter_config.py

@@ -0,0 +1,51 @@
+"""
+Twitter OAuth Configuration
+
+To use this Twitter OAuth implementation, you need to:
+
+1. Create a Twitter Developer account at https://developer.twitter.com/
+2. Create a new app in the Twitter Developer Portal
+3. Get your Client ID and Client Secret
+4. Set up the following environment variables:
+
+Required Environment Variables:
+- TWITTER_CLIENT_ID: Your Twitter app's client ID
+- TWITTER_CLIENT_SECRET: Your Twitter app's client secret
+- TWITTER_REDIRECT_URI: The redirect URI for OAuth (default: http://localhost:8080/callback)
+
+Optional Environment Variables:
+- TWITTER_OAUTH_SERVER_URL: Custom OAuth server URL (default: https://api.twitter.com/2)
+
+Example .env file:
+TWITTER_CLIENT_ID=your_client_id_here
+TWITTER_CLIENT_SECRET=your_client_secret_here
+TWITTER_REDIRECT_URI=http://localhost:8080/callback
+"""
+
+import os
+from typing import Optional
+
+class TwitterConfig:
+    """Configuration class for Twitter OAuth"""
+    
+    def __init__(self):
+        self.client_id = os.getenv("TWITTER_CLIENT_ID")
+        self.client_secret = os.getenv("TWITTER_CLIENT_SECRET")
+        self.redirect_uri = os.getenv("TWITTER_REDIRECT_URI", "http://localhost:8080/callback")
+        self.oauth_server_url = os.getenv("TWITTER_OAUTH_SERVER_URL", "https://api.twitter.com/2")
+        
+    def validate(self) -> bool:
+        """Validate that required configuration is present"""
+        return bool(self.client_id and self.client_secret)
+    
+    def get_missing_config(self) -> list[str]:
+        """Get list of missing configuration variables"""
+        missing = []
+        if not self.client_id:
+            missing.append("TWITTER_CLIENT_ID")
+        if not self.client_secret:
+            missing.append("TWITTER_CLIENT_SECRET")
+        return missing
+
+# Global config instance
+twitter_config = TwitterConfig()

twitter_oauth_example.py

@@ -0,0 +1,138 @@
+#!/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())