Ask user tool

tools/ask_user.py

@@ -0,0 +1,31 @@
+import asyncio
+from typing import Any, Dict
+from tools.utils import ToolError
+
+logger = None  # logging not wired up in this module; keep it simple
+
+async def handle(args: Dict[str, Any]):
+    """
+    Asks the human operator a question and blocks until they type an answer.
+    Intended as a circuit-breaker when the model is missing critical context
+    or needs guidance. Prints the question to the MCP server's terminal and
+    reads the answer from stdin.
+    """
+    question = args.get("question")
+    if not question:
+        raise ToolError("Missing 'question' argument")
+
+    # Print the question to the server terminal so the human sees it.
+    print(f">>> {question}")
+
+    try:
+        answer = await asyncio.to_thread(input, ">>> ")
+    except (EOFError, KeyboardInterrupt):
+        print("\n>>> Input interrupted.")
+        raise ToolError("ask_user was interrupted")
+
+    answer = (answer or "").strip()
+    if not answer:
+        raise ToolError("ask_user received an empty answer")
+
+    return [{"type": "text", "text": answer}]