from fastmcp import FastMCP

mcp = FastMCP("Utility Server")

@mcp.tool
def calculate_area(length: float, width: float) -> float:
    """Calculate the area of a rectangle."""
    return length * width

@mcp.tool
def generate_password(length: int = 12, include_symbols: bool = True) -> str:
    """Generate a random password."""
    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:
    """Get current weather for a city."""
    await ctx.info(f"Fetching weather for {city}...")

    # Make HTTP request using context
    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:
    """Analyze text using the client's LLM."""
    prompt = f"""
    Please analyze the following text and provide insights about:
    1. Sentiment
    2. Key themes
    3. Writing style

    Text: {text}
    """

    # Use the client's LLM for analysis
    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:
    """Create a chart from data."""
    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"{chart_type.title()} Chart")
    plt.xlabel("Index")
    plt.ylabel("Value")

    # Save to bytes
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)

    # Convert to 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:
    """Process a list of numbers and return statistics."""
    import statistics

    if not numbers:
        return {"error": "No data provided"}

    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:
    """Database configuration settings."""
    return {
        "host": "localhost",
        "port": 5432,
        "database": "myapp",
        "ssl_enabled": True
    }

@mcp.resource("docs://api-guide")
def get_api_documentation() -> str:
    """API usage documentation."""
    return """
    # API Usage Guide

    ## Authentication
    All requests require an API key in the Authorization header.

    ## Rate Limits
    - 1000 requests per hour for basic tier
    - 10000 requests per hour for premium tier

    ## Available Endpoints
    - GET /users - List all users
    - POST /users - Create new user
    - GET /users/{id} - Get user details
    """

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """Get a user's profile information."""
    # In a real app, this would query a database
    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"User {user_id} not found")

    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:
    """Get statistics for a specific dataset."""
    # Simulate different datasets
    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"Dataset '{dataset}' not found")

    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:
    """Read content from a file in the project directory."""
    # Security: restrict to project directory
    project_root = Path(__file__).parent
    full_path = project_root / file_path

    # Ensure the path is within the project directory
    if not str(full_path.resolve()).startswith(str(project_root.resolve())):
        raise ValueError("Access denied: path outside project directory")

    if not full_path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")

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

@mcp.resource("logs://{date}")
def get_daily_logs(date: str) -> str:
    """Get application logs for a specific date."""
    import datetime

    try:
        # Validate date format
        datetime.datetime.strptime(date, "%Y-%m-%d")
    except ValueError:
        raise ValueError("Date must be in YYYY-MM-DD format")

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

    if not os.path.exists(log_file):
        return f"No logs found for {date}"

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

@mcp.prompt
def code_review_prompt(code: str, language: str = "python") -> str:
    """Generate a prompt for code review."""
    return f"""
    Please review the following {language} code and provide feedback on:

    1. Code quality and readability
    2. Performance considerations
    3. Security concerns
    4. Best practices adherence
    5. Suggestions for improvement

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

    Please provide constructive feedback with specific examples.
    """

@mcp.prompt
def data_analysis_prompt(data_description: str, questions: list[str]) -> str:
    """Generate a prompt for data analysis."""
    questions_text = "\\n".join(f"- {q}" for q in questions)

    return f"""
    I need help analyzing data with the following characteristics:
    {data_description}

    Specific questions I want to explore:
    {questions_text}

    Please provide a structured analysis approach and suggest 
    appropriate statistical methods or visualizations.
    """

@mcp.tool
def divide_numbers(a: float, b: float) -> float:
    """Divide two numbers with error handling."""
    if b == 0:
        raise ValueError("Cannot divide by zero")

    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise TypeError("Both arguments must be numbers")

    return a / b

@mcp.tool
def process_email(email: str) -> dict:
    """Validate and process an email address."""
    import re

    # Basic email validation
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'

    if not re.match(email_pattern, email):
        raise ValueError("Invalid email format")

    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:
        # Test tools
        result = await client.call_tool("calculate_area", {"length": 5, "width": 3})
        print(f"Area: {result.text}")

        # Test resources
        config = await client.read_resource("config://database")
        print(f"Config: {config.content}")

        # Test resource templates
        user = await client.read_resource("users://1/profile")
        print(f"User: {user.content}")

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