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

@mcp.tool
async def check_kiyoshi(ctx: Context) -> str:
    """
    Checks if the zundoko history matches the winning pattern (Zun Zun Zun Zun Doko)
    and elicits "Ki-yo-shi!" from the user if it does.
    """
    history_count = len(zundoko_history)

    if history_count < 5:
        return f"History has only {history_count} item(s), need at least 5 to check for the pattern."

    if zundoko_history[-5:] == ["Zun", "Zun", "Zun", "Zun", "Doko"]:
        return "Pattern found!"
    else:
        return f"Pattern not found. Last 5 items: {zundoko_history[-5:]}"

 from fastmcp import FastMCP, Context
+from fastmcp.server.elicitation import (
+    AcceptedElicitation,
+    DeclinedElicitation,
+    CancelledElicitation,
+)

<snip>

     if zundoko_history[-5:] == ["Zun", "Zun", "Zun", "Zun", "Doko"]:
-        return "Pattern found!"
+        elicit_result = await ctx.elicit(
+            "It's time to say Ki-yo-shi!",
+            response_type=str
+        )
+
+        match elicit_result:
+            case AcceptedElicitation(data=response):
+                if "Ki-yo-shi!" == response:
+                    return "Perfect!'"
+                else:
+                    return f"Pattern found! But you said '{response}' instead of 'Ki-yo-shi!'"
+            case DeclinedElicitation():
+                return "Pattern found! You declined to say 'Ki-yo-shi!'"
+            case CancelledElicitation():
+                return "Pattern found! You cancelled Ki-yo-shi!"
     else:
         return f"Pattern not found. Last 5 items: {zundoko_history[-5:]}"

from fastmcp.client.elicitation import ElicitResult


kiyoshi_answered = False

# ハンドラ関数の実装
async def elicitation_handler(message: str, response_type: type, params, context):
    global kiyoshi_answered

    # Elicitationリクエストのメッセージをコンソールに表示して、ユーザーの入力を受け取る
    user_input = input(f"{message}: ")
    kiyoshi_answered = True

    # ズンドコMCPサーバーにユーザー入力を返す
    return ElicitResult(action="accept", content=response_type(value=user_input))

 async def main():
     async with Client(
         "http://127.0.0.1:8080/mcp",
         log_handler=handle_log,
+        elicitation_handler=elicitation_handler,
     ) as client:
         await client.call_tool("get_zundoko", {})

         elicitation_handler=elicitation_handler,
     ) as client:
-        await client.call_tool("get_zundoko", {})
+        while True:
+            # 「キ・ヨ・シ！」条件達成まで無限ループ
+            if kiyoshi_answered:
+                break
+
+            try:
+                # ズンドコを一つ取得
+                zundoko = await client.call_tool("get_zundoko", {})
+                print(zundoko.content[0].text)
+
+                # 「キ・ヨ・シ！」条件チェック
+                kiyoshi_check_result = await client.call_tool("check_kiyoshi", {})
+                print(kiyoshi_check_result.content[0].text)
+
+                await asyncio.sleep(1)
+            except Exception as e:
+                print(f"Error: {e}")
+                break

<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!: Ki-yo-shi!
Perfect!'