| import logging |
| from typing import Callable |
|
|
| from ..core import BaseService |
|
|
| logger = logging.getLogger("qkernel.executors.tool") |
|
|
| ToolHandler = Callable[[dict], object] |
|
|
|
|
| class ToolExecutor(BaseService): |
| name: str = "tool" |
|
|
| def __init__(self) -> None: |
| self._tools: dict[str, ToolHandler] = {} |
|
|
| def register(self, name: str, handler: ToolHandler) -> None: |
| self._tools[name] = handler |
| logger.info("Tool registered: %s", name) |
|
|
| async def process(self, event: dict) -> dict: |
| tool_name = event.get("tool", "") |
| params = event.get("params", {}) or {} |
|
|
| handler = self._tools.get(tool_name) |
| if handler is None: |
| available = list(self._tools.keys()) |
| return { |
| "response": "", |
| "error": f"tool_not_found: {tool_name}", |
| "available": available, |
| } |
|
|
| try: |
| result = handler(params) |
| return {"response": result, "tool": tool_name} |
| except Exception as e: |
| logger.error("Tool %s failed: %s", tool_name, e) |
| return {"response": "", "tool": tool_name, "error": str(e)} |
