32 lines
1.0 KiB
Python
32 lines
1.0 KiB
Python
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}]
|