Spaces:
Sleeping
Sleeping
File size: 5,285 Bytes
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 | #!/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())
|