from fastmcp import FastMCP

mcp = FastMCP("Utility Server")

@mcp.tool
def calculate_area(length: float, width: float) -> float:
    """長方形の面積を計算します。"""
    return length * width

@mcp.tool
def generate_password(length: int = 12, include_symbols: bool = True) -> str:
    """ランダムなパスワードを生成します。"""
    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:
    """指定された都市の現在の天気を取得します。"""
    await ctx.info(f"Fetching weather for {city}...")

    # コンテキストを使ってHTTPリクエストを行う
    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:
    """クライアントのLLMを使用してテキストを分析します。"""
    prompt = f"""
    以下のテキストを分析し、以下の点について洞察を提供してください:
    1. センチメント
    2. 主要なテーマ
    3. 執筆スタイル

    テキスト: {text}
    """

    # クライアントのLLMを使って分析する
    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:
    """データからチャートを作成します。"""
    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")

    # バイト形式で保存
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)

    # 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:
    """数値のリストを処理し、統計を返します。"""
    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:
    """データベース設定です。"""
    return {
        "host": "localhost",
        "port": 5432,
        "database": "myapp",
        "ssl_enabled": True
    }

@mcp.resource("docs://api-guide")
def get_api_documentation() -> str:
    """API利用ドキュメントです。"""
    return """
    # API利用ガイド

    ## 認証
    すべてのリクエストにはAuthorizationヘッダーにAPIキーが必要です。

    ## レート制限
    - ベーシックティア: 1時間あたり1000リクエスト
    - プレミアムティア: 1時間あたり10000リクエスト

    ## 利用可能なエンドポイント
    - GET /users - すべてのユーザーをリスト表示
    - POST /users - 新しいユーザーを作成
    - GET /users/{id} - ユーザーの詳細を取得
    """

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """ユーザーのプロフィール情報を取得します。"""
    # 実際のアプリケーションでは、これはデータベースをクエリします
    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:
    """特定のデータセットの統計情報を取得します。"""
    # 異なるデータセットをシミュレート
    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:
    """プロジェクトディレクトリ内のファイルからコンテンツを読み取ります。"""
    # セキュリティ: プロジェクトディレクトリ内に制限
    project_root = Path(__file__).parent
    full_path = project_root / file_path

    # パスがプロジェクトディレクトリ内であることを確認
    if not str(full_path.resolve()).startswith(str(project_root.resolve())):
        raise ValueError("アクセス拒否: パスがプロジェクトディレクトリ外です")

    if not full_path.exists():
        raise FileNotFoundError(f"ファイルが見つかりません: {file_path}")

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

@mcp.resource("logs://{date}")
def get_daily_logs(date: str) -> str:
    """特定の日付のアプリケーションログを取得します。"""
    import datetime

    try:
        # 日付形式を検証
        datetime.datetime.strptime(date, "%Y-%m-%d")
    except ValueError:
        raise ValueError("日付はYYYY-MM-DD形式である必要があります")

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

    if not os.path.exists(log_file):
        return f"日付 {date} のログは見つかりません"

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

@mcp.prompt
def code_review_prompt(code: str, language: str = "python") -> str:
    """コードレビューのためのプロンプトを生成します。"""
    return f"""
    以下の{language}コードをレビューし、以下の点についてフィードバックを提供してください:

    1. コードの品質と可読性
    2. パフォーマンスに関する考慮事項
    3. セキュリティ上の懸念
    4. ベストプラクティスの順守
    5. 改善のための提案

    コード:
    ```{language}
    {code}
    ```

    具体的な例を挙げて、建設的なフィードバックをしてください。
    """

@mcp.prompt
def data_analysis_prompt(data_description: str, questions: list[str]) -> str:
    """データ分析のためのプロンプトを生成します。"""
    questions_text = "\\n".join(f"- {q}" for q in questions)

    return f"""
    以下の特徴を持つデータの分析を手伝ってほしいです:
    {data_description}

    特に探求したい質問は以下の通りです:
    {questions_text}

    構造化された分析アプローチと、適切な統計的手法や視覚化を提案してください。
    """

@mcp.tool
def divide_numbers(a: float, b: float) -> float:
    """エラーハンドリング付きで2つの数値を割ります。"""
    if b == 0:
        raise ValueError("ゼロで割ることはできません")

    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise TypeError("両方の引数は数値である必要があります")

    return a / b

@mcp.tool
def process_email(email: str) -> dict:
    """メールアドレスを検証し処理します。"""
    import re

    # 基本的なメール検証
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'

    if not re.match(email_pattern, email):
        raise ValueError("無効なメール形式です")

    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:
        # ツールのテスト
        result = await client.call_tool("calculate_area", {"length": 5, "width": 3})
        print(f"面積: {result.text}")

        # リソースのテスト
        config = await client.read_resource("config://database")
        print(f"設定: {config.content}")

        # リソーステンプレートのテスト
        user = await client.read_resource("users://1/profile")
        print(f"ユーザー: {user.content}")

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