FastMCP 入門 (5) - プロンプト

from fastmcp import FastMCP
from fastmcp.prompts.prompt import UserMessage, AssistantMessage, Message

mcp = FastMCP(name="PromptServer")

# 文字列を返す基本プロンプト (UserMessage に変換される)
@mcp.prompt()
def ask_about_topic(topic: str) -> str:
    """トピックの説明を求めるユーザー メッセージを生成"""
    return f"'{topic}'の概念を説明していただけますか？"

# 特定のメッセージタイプを返すプロンプト
@mcp.prompt()
def generate_code_request(language: str, task_description: str) -> UserMessage:
    """コード生成を要求するユーザー メッセージを生成"""
    content = f"次のタスクを実行する {language} 関数を記述します: {task_description}"
    return UserMessage(content=content)

@mcp.prompt()
def roleplay_scenario(character: str, situation: str) -> list[Message]:
    """最初のメッセージでロールプレイング シナリオを設定"""
    return [
        UserMessage(f"ロールプレイをしましょう。あなたは{character}です。状況は{situation}です。"),
        AssistantMessage("分かりました。準備はできました。次は何をするんですか？")
    ]

@mcp.prompt()
def ask_for_feedback() -> dict:
    """フィードバックを求めるユーザー メッセージを生成"""
    return {"role": "user", "content": "私の前回の返答についてどう思いましたか?"}

from pydantic import Field
from typing import Literal, Optional

@mcp.prompt()
def generate_content_request(
    topic: str = Field(description="The main subject to cover"),
    format: Literal["blog", "email", "social"] = "blog",
    tone: str = "professional",
    word_count: Optional[int] = None
) -> str:
    """特定の形式でコンテンツを生成するためのリクエストを作成"""
    prompt = f"{topic} について、{tone} トーンで {format} の投稿を書いてください。"
    
    if word_count:
        prompt += f"長さは約 {word_count} 語です。"
        
    return prompt

@mcp.prompt()
def data_analysis_prompt(
    data_uri: str,                        # 必須 - デフォルト値なし
    analysis_type: str = "summary",       # オプション - デフォルト値あり
    include_charts: bool = False          # オプション - デフォルト値あり
) -> str:
    """特定のパラメータを使用してデータを分析するためのリクエストを作成"""
    prompt = f"{data_uri} にあるデータに対して '{analysis_type}' 分析を実行してください。"
    if include_charts:
        prompt += " Include relevant charts and visualizations."
    return prompt

@mcp.prompt(
    name="analyze_data_request",          # プロンプト名
    description="Creates a request to analyze data with specific parameters",  # 説明
    tags={"analysis", "data"}             # 分類タグ
)
def data_analysis_prompt(
    data_uri: str = Field(description="データを含むリソースのURI"),
    analysis_type: str = Field(default="summary", description="分析の種類")
) -> str:
    """説明が提供されている場合、このドキュメント文字列は無視される"""
    return f"{data_uri} にあるデータに対して '{analysis_type}' 分析を実行してください。"

# 同期プロンプト
@mcp.prompt()
def simple_question(question: str) -> str:
    """LLM に尋ねる簡単な質問を生成"""
    return f"Question: {question}"

# 非同期プロンプト
@mcp.prompt()
async def data_based_prompt(data_id: str) -> str:
    """取得する必要があるデータに基づいてプロンプトを生成"""
    # 実際のシナリオでは、データベースやAPIからデータを取得する場合がある
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/data/{data_id}") as response:
            data = await response.json()
            return f"Analyze this data: {data['content']}"

from fastmcp import FastMCP, Context

mcp = FastMCP(name="PromptServer")

@mcp.prompt()
async def generate_report_request(report_type: str, ctx: Context) -> str:
    """レポートのリクエストを生成"""
    return f"{report_type}レポートを作成してください。リクエストID: {ctx.request_id}"

from fastmcp import FastMCP

mcp = FastMCP(
    name="PromptServer",
    on_duplicate_prompts="error"  # Raise an error if a prompt name is duplicated
)

@mcp.prompt()
def greeting(): return "Hello, how can I help you today?"

# This registration attempt will raise a ValueError because
# "greeting" is already registered and the behavior is "error".
# @mcp.prompt()
# def greeting(): return "Hi there! What can I do for you?"