from fastmcp import FastMCP

mcp = FastMCP("Utility Server")

@mcp.tool
def calculate_area(length: float, width: float) -> float:
    """Calcule la surface d'un rectangle."""
    return length * width

@mcp.tool
def generate_password(length: int = 12, include_symbols: bool = True) -> str:
    """Génère un mot de passe aléatoire."""
    import random
    import string

    chars = string.ascii_letters + string.digits
    if include_symbols:
        chars += "!@#$%^&*"

    return ''.join(random.choice(chars) for _ in range(length))

from fastmcp import FastMCP, Context
import aiohttp
import json

mcp = FastMCP("API Tools")

@mcp.tool
async def fetch_weather(city: str, ctx: Context) -> dict:
    """Obtient la météo actuelle pour une ville."""
    await ctx.info(f"Récupération de la météo pour {city}...")

    # Effectuer une requête HTTP en utilisant le contexte
    response = await ctx.http_request(
        "GET", 
        f"https://api.openweathermap.org/data/2.5/weather",
        params={"q": city, "appid": "your_api_key"}
    )

    weather_data = response.json()

    return {
        "city": city,
        "temperature": weather_data["main"]["temp"],
        "description": weather_data["weather"][0]["description"],
        "humidity": weather_data["main"]["humidity"]
    }

@mcp.tool
async def analyze_text(text: str, ctx: Context) -> str:
    """Analyse le texte en utilisant le LLM du client."""
    prompt = f"""
    Veuillez analyser le texte suivant et fournir des informations sur :
    1. Le sentiment
    2. Les thèmes clés
    3. Le style d'écriture

    Texte : {text}
    """

    # Utiliser le LLM du client pour l'analyse
    response = await ctx.sample(prompt)
    return response.text

from fastmcp import FastMCP
from fastmcp.types import Image, BlobResourceContents
import matplotlib.pyplot as plt
import io
import base64

mcp = FastMCP("Data Visualization")

@mcp.tool
def create_chart(data: list[float], chart_type: str = "line") -> Image:
    """Crée un graphique à partir de données."""
    plt.figure(figsize=(10, 6))

    if chart_type == "line":
        plt.plot(data)
    elif chart_type == "bar":
        plt.bar(range(len(data)), data)

    plt.title(f"Graphique {chart_type.title()}")
    plt.xlabel("Index")
    plt.ylabel("Valeur")

    # Sauvegarder en octets
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)

    # Convertir en base64
    image_base64 = base64.b64encode(buffer.read()).decode()

    return Image(
        data=image_base64,
        mimeType="image/png"
    )

@mcp.tool
def process_data(numbers: list[float]) -> dict:
    """Traite une liste de nombres et renvoie des statistiques."""
    import statistics

    if not numbers:
        return {"error": "Aucune donnée fournie"}

    return {
        "count": len(numbers),
        "sum": sum(numbers),
        "mean": statistics.mean(numbers),
        "median": statistics.median(numbers),
        "std_dev": statistics.stdev(numbers) if len(numbers) > 1 else 0,
        "min": min(numbers),
        "max": max(numbers)
    }

@mcp.resource("config://database")
def get_database_config() -> dict:
    """Paramètres de configuration de la base de données."""
    return {
        "host": "localhost",
        "port": 5432,
        "database": "myapp",
        "ssl_enabled": True
    }

@mcp.resource("docs://api-guide")
def get_api_documentation() -> str:
    """Documentation d'utilisation de l'API."""
    return """
    # Guide d'utilisation de l'API

    ## Authentification
    Toutes les requêtes nécessitent une clé API dans l'en-tête Authorization.

    ## Limites de débit
    - 1000 requêtes par heure pour le niveau de base
    - 10000 requêtes par heure pour le niveau premium

    ## Points d'accès disponibles
    - GET /users - Lister tous les utilisateurs
    - POST /users - Créer un nouvel utilisateur
    - GET /users/{id} - Obtenir les détails de l'utilisateur
    """

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """Obtient les informations de profil d'un utilisateur."""
    # Dans une application réelle, cela interrogerait une base de données
    users_db = {
        1: {"name": "Alice Johnson", "role": "Admin", "email": "[email protected]"},
        2: {"name": "Bob Smith", "role": "User", "email": "[email protected]"},
        3: {"name": "Carol Davis", "role": "Manager", "email": "[email protected]"}
    }

    user = users_db.get(user_id)
    if not user:
        raise ValueError(f"Utilisateur {user_id} introuvable")

    return {
        "id": user_id,
        **user,
        "last_login": "2024-01-15T10:30:00Z",
        "status": "active"
    }

@mcp.resource("data://{dataset}/stats")
def get_dataset_stats(dataset: str) -> dict:
    """Obtient les statistiques pour un jeu de données spécifique."""
    # Simuler différents jeux de données
    datasets = {
        "sales": {"records": 10000, "last_updated": "2024-01-15", "size_mb": 45.2},
        "users": {"records": 5000, "last_updated": "2024-01-14", "size_mb": 12.8},
        "products": {"records": 2500, "last_updated": "2024-01-13", "size_mb": 8.1}
    }

    if dataset not in datasets:
        raise ValueError(f"Jeu de données '{dataset}' introuvable")

    return {
        "dataset": dataset,
        **datasets[dataset],
        "access_level": "public"
    }

import os
from pathlib import Path

@mcp.resource("files://{file_path}")
def read_file_content(file_path: str) -> str:
    """Lit le contenu d'un fichier dans le répertoire du projet."""
    # Sécurité : restreindre au répertoire du projet
    project_root = Path(__file__).parent
    full_path = project_root / file_path

    # S'assurer que le chemin est dans le répertoire du projet
    if not str(full_path.resolve()).startswith(str(project_root.resolve())):
        raise ValueError("Accès refusé : chemin en dehors du répertoire du projet")

    if not full_path.exists():
        raise FileNotFoundError(f"Fichier introuvable : {file_path}")

    return full_path.read_text(encoding='utf-8')

@mcp.resource("logs://{date}")
def get_daily_logs(date: str) -> str:
    """Obtient les journaux d'application pour une date spécifique."""
    import datetime

    try:
        # Valider le format de la date
        datetime.datetime.strptime(date, "%Y-%m-%d")
    except ValueError:
        raise ValueError("La date doit être au format YYYY-MM-DD")

    log_file = f"logs/{date}.log"

    if not os.path.exists(log_file):
        return f"Aucun journal trouvé pour le {date}"

    with open(log_file, 'r') as f:
        return f.read()

@mcp.prompt
def code_review_prompt(code: str, language: str = "python") -> str:
    """Génère un prompt pour la révision de code."""
    return f"""
    Veuillez examiner le code {language} suivant et fournir des commentaires sur :

    1. La qualité et la lisibilité du code
    2. Les considérations de performance
    3. Les problèmes de sécurité
    4. L'adhésion aux meilleures pratiques
    5. Les suggestions d'amélioration

    Code :
    ```{language}
    {code}
    ```

    Veuillez fournir des commentaires constructifs avec des exemples spécifiques.
    """

@mcp.prompt
def data_analysis_prompt(data_description: str, questions: list[str]) -> str:
    """Génère un prompt pour l'analyse de données."""
    questions_text = "\\n".join(f"- {q}" for q in questions)

    return f"""
    J'ai besoin d'aide pour analyser des données avec les caractéristiques suivantes :
    {data_description}

    Questions spécifiques que je souhaite explorer :
    {questions_text}

    Veuillez fournir une approche d'analyse structurée et suggérer 
    des méthodes statistiques ou des visualisations appropriées.
    """

@mcp.tool
def divide_numbers(a: float, b: float) -> float:
    """Divise deux nombres avec gestion des erreurs."""
    if b == 0:
        raise ValueError("Impossible de diviser par zéro")

    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise TypeError("Les deux arguments doivent être des nombres")

    return a / b

@mcp.tool
def process_email(email: str) -> dict:
    """Valide et traite une adresse e-mail."""
    import re

    # Validation de base de l'e-mail
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'

    if not re.match(email_pattern, email):
        raise ValueError("Format d'e-mail invalide")

    username, domain = email.split('@')

    return {
        "email": email,
        "username": username,
        "domain": domain,
        "is_valid": True
    }

import asyncio
from fastmcp import Client

async def test_server():
    async with Client(mcp) as client:
        # Tester les outils
        result = await client.call_tool("calculate_area", {"length": 5, "width": 3})
        print(f"Surface : {result.text}")

        # Tester les ressources
        config = await client.read_resource("config://database")
        print(f"Configuration : {config.content}")

        # Tester les modèles de ressources
        user = await client.read_resource("users://1/profile")
        print(f"Utilisateur : {user.content}")

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