Spaces:
Sleeping
Sleeping
File size: 4,762 Bytes
b3efdb3 524e1e0 b3efdb3 524e1e0 b3efdb3 524e1e0 b3efdb3 524e1e0 b3efdb3 524e1e0 b3efdb3 e7206f6 b3efdb3 80dc3a4 b3efdb3 | 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 | from fastapi import FastAPI, Request, HTTPException
from fastmcp import FastMCP
from authlib.integrations.starlette_client import OAuth
from starlette.middleware.sessions import SessionMiddleware
import os
from typing import Dict, Any, Optional
# Validation des variables d'environnement
required_env_vars = ["CLIENT_ID", "CLIENT_SECRET"]
missing_vars = [var for var in required_env_vars if not os.getenv(var)]
if missing_vars:
raise ValueError(f"Variables d'environnement manquantes: {', '.join(missing_vars)}")
app = FastAPI()
app.add_middleware(
SessionMiddleware,
secret_key="CHANGE_ME"
)
# --- Config OAuth Twitter ---
oauth = OAuth()
oauth.register(
name='twitter',
client_id=os.getenv("CLIENT_ID"),
client_secret=os.getenv("CLIENT_SECRET"),
authorize_url="https://twitter.com/i/oauth2/authorize",
access_token_url="https://api.twitter.com/2/oauth2/token",
client_kwargs={
"scope": "tweet.read users.read offline.access",
"token_endpoint_auth_method": "client_secret_post"
}
)
# --- Fonction d'authentification pour MCP ---
async def mcp_auth(request: Request) -> Optional[Dict[str, Any]]:
"""
Fonction d'authentification pour FastMCP
Retourne les données utilisateur si authentifié, None sinon
"""
user = request.session.get('user')
if not user:
return None
return user
# --- FastMCP Server avec auth intégrée ---
mcp = FastMCP(
name="TwitterMCP",
stateless_http=True,
auth=mcp_auth
)
@mcp.tool()
async def hello() -> str:
"""Fonction de test MCP"""
return "Hello depuis FastMCP 🎉"
@mcp.tool()
async def get_current_user(request: Request) -> Dict[str, Any]:
"""Retourne les informations de l'utilisateur authentifié"""
user = request.session.get('user', {})
return {
"username": user.get('username'),
"id": user.get('id'),
"name": user.get('name')
}
@mcp.tool()
async def tweet_something(request: Request, message: str) -> Dict[str, str]:
"""
Exemple d'outil qui pourrait tweeter (nécessiterait les bonnes permissions)
"""
user = request.session.get('user', {})
# Ici on pourrait utiliser le token stocké pour poster un tweet
return {
"status": "success",
"message": f"Tweet simulé de @{user.get('username', 'unknown')}: {message}"
}
# --- Routes OAuth ---
@app.get("/login")
async def login(request: Request):
"""Initie le processus d'authentification Twitter"""
try:
redirect_uri = request.url_for("auth_callback")
return await oauth.twitter.authorize_redirect(request, redirect_uri)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erreur lors de l'initialisation OAuth: {str(e)}")
@app.get("/callback")
async def auth_callback(request: Request) -> Dict[str, str]:
"""Gère le callback OAuth et stocke les infos utilisateur"""
try:
token = await oauth.twitter.authorize_access_token(request)
# Récupère les infos de l'utilisateur
user_req = await oauth.twitter.get(
"https://api.twitter.com/2/users/me",
token=token
)
if user_req.status_code != 200:
raise HTTPException(
status_code=400,
detail="Impossible de récupérer les informations utilisateur"
)
userinfo = user_req.json().get('data', {})
if not userinfo:
raise HTTPException(
status_code=400,
detail="Données utilisateur invalides"
)
# Stocke les infos utilisateur et le token en session
request.session['user'] = userinfo
request.session['token'] = token
return {"message": f"Connecté en tant que @{userinfo.get('username', 'utilisateur')}"}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erreur lors de l'authentification: {str(e)}")
@app.get("/logout")
async def logout(request: Request):
"""Déconnecte l'utilisateur"""
request.session.clear()
return {"message": "Déconnecté avec succès"}
@app.get("/profile")
async def get_profile(request: Request):
"""Retourne le profil de l'utilisateur connecté"""
user = request.session.get('user')
if not user:
raise HTTPException(status_code=401, detail="Non authentifié")
return {"user": user}
# --- Route de santé ---
@app.get("/health")
async def health_check():
"""Vérification de l'état de l'application"""
return {"status": "ok", "message": "Service opérationnel"}
# --- Intégration FastMCP dans FastAPI ---
app.mount("/mcp", mcp.app)
|