ズンドコMCPサーバー・クライアント

         # Elicitationリクエスト送信
         elicit_result = await ctx.elicit(
             "It's time to say Ki-yo-shi!",
             response_type=str
         )

         # Elicitationレスポンスの処理
         match elicit_result:
             case AcceptedElicitation(data=response):
                 # ユーザーの入力値のチェック
                 if "Ki-yo-shi!" == response:
                     <snip>
                 else:
-                    return f"Pattern found! But you said '{response}' instead of 'Ki-yo-shi!'"
+                    prompt = (
+                        f"The user was supposed to say 'Ki-yo-shi!' but instead said '{response}'. "
+                        "Generate a humorous and creative warning message (one sentence) about their mistake."
+                    )
+                    # Samplingリクエスト送信
+                    sampling_result = await ctx.sample(
+                        messages=prompt,
+                        max_tokens=5120,
+                        temperature=1.0
+                    )
+                    return sampling_result.text
             case DeclinedElicitation():

from fastmcp.client.sampling import SamplingMessage, SamplingParams, RequestContext
from openai import AsyncOpenAI

# ハンドラ関数の実装
async def sampling_handler(
    messages: list[SamplingMessage],
    params: SamplingParams,
    context: RequestContext
) -> str:
    # Samplingリクエストからプロンプトを取得
    message_content = messages[0].content.text

    # OpenAIのモデルを呼んで推論を実行
    response = await AsyncOpenAI().responses.create(
        model="gpt-5-nano",
        input=message_content,
        max_output_tokens=params.maxTokens or 5120,
        temperature=params.temperature or 1.0,
    )

    # 推論結果をMCPサーバーに返す
    return response.output_text

 async def main():
     async with Client(
         "http://127.0.0.1:8080/mcp",
         log_handler=handle_log,
         elicitation_handler=elicitation_handler,
+        sampling_handler=sampling_handler,
     ) as client:
         while True:

<snip>
Zun
Pattern not found. Last 5 items: ['Zun', 'Doko', 'Zun', 'Zun', 'Zun']
Zun
Pattern not found. Last 5 items: ['Doko', 'Zun', 'Zun', 'Zun', 'Zun']
Doko
It's time to say Ki-yo-shi!:

<snip>
Zun
Pattern not found. Last 5 items: ['Zun', 'Doko', 'Zun', 'Zun', 'Zun']
Zun
Pattern not found. Last 5 items: ['Doko', 'Zun', 'Zun', 'Zun', 'Zun']
Doko
It's time to say Ki-yo-shi!: Ya-su-shi!
Warning: You mispronounced the sacred chant—'Ya-su-shi!'—and the dojo ninjas are filing a complaint with the Sushi Court; please retry with 'Ki-yo-shi!' before the rice turns rebellious.

-mcp = FastMCP("Zundoko Server")
+mcp = FastMCP(
+    "Zundoko Server",
+    sampling_handler=lambda *_: "Correct answer is 'Ki-yo-shi!'"
+)

 async def main():
     async with Client(
         "http://127.0.0.1:8080/mcp",
         log_handler=handle_log,
         elicitation_handler=elicitation_handler,
         sampling_handler=sampling_handler,
     ) as client:
+        print(f"Zundoko MCP Server is alive: {await client.ping()}")
+
         while True:

Zundoko MCP Server is alive: True