| 1 | from __future__ import annotations |
| 2 | |
| 3 | from copy import deepcopy |
| 4 | from dataclasses import dataclass |
| 5 | from time import time |
| 6 | from typing import Any, Callable |
| 7 | |
| 8 | from .models import ToolCall, ToolResult, TurnControl |
| 9 | from .tools import ToolRegistry, ToolRuntimeContext |
| 10 | |
| 11 | |
| 12 | @dataclass(slots=True) |
| 13 | class ToolExecutionRecord: |
| 14 | requested_name: str |
| 15 | canonical_name: str |
| 16 | arguments_before: dict[str, Any] |
| 17 | arguments_after: dict[str, Any] |
| 18 | result: ToolResult |
| 19 | started_at: float |
| 20 | finished_at: float |
| 21 | telemetry: dict[str, Any] |
| 22 | |
| 23 | |
| 24 | class ToolExecutor: |
| 25 | def __init__(self, registry: ToolRegistry, session_index: Any) -> None: |
| 26 | self.registry = registry |
| 27 | self.session_index = session_index |
| 28 | |
| 29 | async def execute(self, call: ToolCall, control: TurnControl, progress_callback: Callable[[dict[str, Any]], None] | None = None) -> ToolExecutionRecord: |
| 30 | requested_name = call.name |
| 31 | canonical_name = self.registry.resolve_name(call.name) |
| 32 | before = deepcopy(call.arguments) |
| 33 | started_at = time() |
| 34 | validated, validation_error = self.registry.validate_arguments(canonical_name, call.arguments) |
| 35 | arguments = validated if validated is not None else call.arguments |
| 36 | runtime = ToolRuntimeContext(requested_name=requested_name, canonical_name=canonical_name, turn_id=control.turn_id, cancel_event=control.cancel_event, progress_callback=progress_callback, metadata={"cancel_reason": control.cancel_reason}) |
| 37 | if validation_error: |
| 38 | result = ToolResult(canonical_name, False, validation_error, {"validation_error": True}) |
| 39 | elif control.cancel_event.is_set(): |
| 40 | result = ToolResult(canonical_name, False, control.cancel_reason or "Tool execution cancelled", {"cancelled": True}) |
| 41 | else: |
| 42 | result = await self.registry.execute(canonical_name, arguments, runtime=runtime) |
| 43 | finished_at = time() |
| 44 | telemetry = {"duration_ms": int((finished_at - started_at) * 1000), "requested_name": requested_name, "canonical_name": canonical_name, "result_ok": result.ok} |
| 45 | self.session_index.append_log("tool_calls", {"turn_id": control.turn_id, "tool": canonical_name, "arguments_preview": str(before)[:500], "ok": result.ok, "content_preview": result.content[:500], **telemetry}) |
| 46 | return ToolExecutionRecord(requested_name, canonical_name, before, deepcopy(arguments), result, started_at, finished_at, telemetry) |
| 47 |