FastMCP 入門 (6) - コンテキスト

from fastmcp import FastMCP, Context

mcp = FastMCP(name="ContextDemo")

@mcp.tool()
async def process_file(file_uri: str, ctx: Context) -> str:
    """ログ記録とリソース アクセスのコンテキストを使用して、ファイルを処理"""
    request_id = ctx.request_id
    await ctx.info(f"[{request_id}] Starting processing for {file_uri}")

    try:
        # コンテキストを使用してリソースを読み取る
        contents_list = await ctx.read_resource(file_uri)
        if not contents_list:
            await ctx.warning(f"Resource {file_uri} is empty.")
            return "Resource empty"

        data = contents_list[0].content # Assuming TextResourceContents
        await ctx.debug(f"Read {len(data)} bytes from {file_uri}")

        # 進捗状況を報告
        await ctx.report_progress(progress=50, total=100)
        
        # 作業をシミュレーション
        processed_data = data.upper() # Example processing

        await ctx.report_progress(progress=100, total=100)
        await ctx.info(f"Processing complete for {file_uri}")

        return f"Processed data length: {len(processed_data)}"

    except Exception as e:
        # コンテキストを使用してエラーを記録
        await ctx.error(f"Error processing {file_uri}: {str(e)}")
        raise # Re-raise to send error back to client

@mcp.tool()
async def analyze_data(data: list[float], ctx: Context) -> dict:
    """ログ記録を使用して数値データを分析"""
    await ctx.debug("Starting analysis of numerical data")
    await ctx.info(f"Analyzing {len(data)} data points")
    
    try:
        result = sum(data) / len(data)
        await ctx.info(f"Analysis complete, average: {result}")
        return {"average": result, "count": len(data)}
    except ZeroDivisionError:
        await ctx.warning("Empty data list provided")
        return {"error": "Empty data list"}
    except Exception as e:
        await ctx.error(f"Analysis failed: {str(e)}")
        raise

@mcp.tool()
async def process_items(items: list[str], ctx: Context) -> dict:
    """進捗状況の更新を含むアイテムのリストを処理"""
    total = len(items)
    results = []
    
    for i, item in enumerate(items):
        # Report progress as percentage
        await ctx.report_progress(progress=i, total=total)
        
        # Process the item (simulated with a sleep)
        await asyncio.sleep(0.1)
        results.append(item.upper())
    
    # 100%完了を報告
    await ctx.report_progress(progress=total, total=total)
    
    return {"processed": len(results), "results": results}

@mcp.tool()
async def summarize_document(document_uri: str, ctx: Context) -> str:
    """リソースURI別にドキュメントを要約"""
    # 文書の内容を読む
    content_list = await ctx.read_resource(document_uri)
    
    if not content_list:
        return "Document is empty"
    
    document_text = content_list[0].content
    
    # 例: 簡単な要約を生成 (長さベース)
    words = document_text.split()
    total_words = len(words)
    
    await ctx.info(f"Document has {total_words} words")
    
    # 簡単な要約を返す
    if total_words > 100:
        summary = " ".join(words[:100]) + "..."
        return f"Summary ({total_words} words total): {summary}"
    else:
        return f"Full document ({total_words} words): {document_text}"

@mcp.tool()
async def analyze_sentiment(text: str, ctx: Context) -> dict:
    """クライアントの LLM を使用してテキストの感情を分析"""
    # 感情分析を求めるサンプリングプロンプトを作成
    prompt = f"次のテキストの感情を、positive、negative、neutralの3つに分類して分析します。「positive」「negative」「neutral」のいずれか1つの単語を出力してください。分析対象テキスト：{text}"
    
    # サンプリングリクエストをクライアントのLLMに送信
    response = await ctx.sample(prompt)
    
    # LLMの応答を処理
    sentiment = response.text.strip().lower()
    
    # 標準的な感情値にマッピング
    if "positive" in sentiment:
        sentiment = "positive"
    elif "negative" in sentiment:
        sentiment = "negative"
    else:
        sentiment = "neutral"
    
    return {"text": text, "sentiment": sentiment}

@mcp.tool()
async def generate_example(concept: str, ctx: Context) -> str:
    """特定の概念の Python コード例を生成"""
    # システムプロンプトとユーザーメッセージの使用
    response = await ctx.sample(
        messages=f"'{concept}' を説明する簡単な Python コード例を記述します。",
        system_prompt="あなたはPythonのエキスパートプログラマーです。説明なしで簡潔かつ動作するコード例を提供してください。",
        temperature=0.7,
        max_tokens=300
    )
    
    code_example = response.text
    return f"```python\n{code_example}\n```"

@mcp.tool()
async def request_info(ctx: Context) -> dict:
    """現在のリクエストに関する情報を返す。"""
    return {
        "request_id": ctx.request_id,
        "client_id": ctx.client_id or "Unknown client"
    }

@mcp.tool()
async def advanced_tool(ctx: Context) -> str:
    """高度なコンテキスト アクセスをデモンストレーション"""
    # FastMCPサーバインスタンスにアクセス
    server_name = ctx.fastmcp.name
    
    # 低レベルのセッションアクセス (あまり使用しない)
    session = ctx.session
    request_context = ctx.request_context
    
    return f"Server: {server_name}"

@mcp.tool()
async def handle_web_request(ctx: Context) -> dict:
    """Starlette リクエストから HTTP リクエスト情報にアクセス"""
    request = ctx.get_http_request()
    
    # HTTP ヘッダ、クエリパラメータなどにアクセス
    user_agent = request.headers.get("user-agent", "Unknown")
    client_ip = request.client.host if request.client else "Unknown"
    
    return {
        "user_agent": user_agent,
        "client_ip": client_ip,
        "path": request.url.path,
    }

@mcp.resource("resource://user-data")
async def get_user_data(ctx: Context) -> dict:
    """リクエストのコンテキストに基づいてパーソナライズされたユーザー データを取得"""
    user_id = ctx.client_id or "anonymous"
    await ctx.info(f"Fetching data for user {user_id}")
    
    # 動的リソース生成にコンテキストを使用する例
    return {
        "user_id": user_id,
        "last_access": datetime.now().isoformat(),
        "request_id": ctx.request_id
    }

@mcp.resource("resource://users/{user_id}/profile")
async def get_user_profile(user_id: str, ctx: Context) -> dict:
    """コンテキスト認識ログを使用してデータベースからユーザー プロファイルを取得"""
    await ctx.info(f"Fetching profile for user {user_id}")
    
    # テンプレートリソースでコンテキストを使用する例
    # 実際の実装では、データベースにクエリを実行する場合がある
    return {
        "id": user_id,
        "name": f"User {user_id}",
        "request_id": ctx.request_id
    }

@mcp.prompt()
async def data_analysis_request(dataset: str, ctx: Context) -> str:
    """コンテキスト情報を使用してデータを分析するためのリクエストを生成"""
    await ctx.info(f"Generating data analysis prompt for {dataset}")
    
    # コンテキストを使用して構成を読み取ったり、プロンプトをカスタマイズしたりできる
    return f"""Please analyze the following dataset: {dataset}
    
Request initiated at: {datetime.now().isoformat()}
Request ID: {ctx.request_id}
"""