| from __future__ import annotations | |
| import asyncio | |
| from typing import Any | |
| from app.schema import ToolResult | |
| from app.tool.base import BaseTool | |
| class AskHuman(BaseTool): | |
| name = "ask_human" | |
| description = "Ask the user a question and wait for their response via stdin." | |
| parameters = { | |
| "type": "object", | |
| "properties": { | |
| "question": {"type": "string", "description": "Question to ask the user."}, | |
| }, | |
| "required": ["question"], | |
| } | |
| async def execute(self, question: str, **_: Any) -> ToolResult: | |
| print(f"\n[ManusClaw asks]: {question}") | |
| try: | |
| loop = asyncio.get_event_loop() | |
| # Use run_in_executor to avoid blocking the event loop | |
| answer = await asyncio.wait_for( | |
| loop.run_in_executor(None, input, "Your answer: "), | |
| timeout=300, # 5-minute timeout to prevent hanging forever | |
| ) | |
| return ToolResult(output=f"User said: {answer}") | |
| except asyncio.TimeoutError: | |
| return ToolResult( | |
| error="No user response received within 5 minutes. " | |
| "In non-interactive mode (e.g., WebSocket server), " | |
| "the ask_human tool requires a connected client that can respond." | |
| ) |