| 69 | |
| 70 | |
| 71 | def dispatch_tool(tool_name: str, code: str, session_id: str) -> list[ToolObservation]: |
| 72 | # Dispatch predefined tools |
| 73 | if tool_name in ALL_TOOLS: |
| 74 | return ALL_TOOLS[tool_name](code, session_id) |
| 75 | |
| 76 | code = code.strip().rstrip("<|observation|>").strip() |
| 77 | |
| 78 | # Dispatch custom tools |
| 79 | try: |
| 80 | tool_params = json.loads(code) |
| 81 | except json.JSONDecodeError as e: |
| 82 | err = f"Error decoding JSON: {e}" |
| 83 | return [ToolObservation("system_error", err)] |
| 84 | |
| 85 | if tool_name not in _TOOL_HOOKS: |
| 86 | err = f"Tool `{tool_name}` not found. Please use a provided tool." |
| 87 | return [ToolObservation("system_error", err)] |
| 88 | |
| 89 | tool_hook = _TOOL_HOOKS[tool_name] |
| 90 | try: |
| 91 | ret: str = tool_hook(**tool_params) |
| 92 | return [ToolObservation(tool_name, str(ret))] |
| 93 | except: |
| 94 | err = traceback.format_exc() |
| 95 | return [ToolObservation("system_error", err)] |
| 96 | |
| 97 | |
| 98 | def get_tools() -> list[dict]: |