The tool runtime is the single enforcement point for all tool invocations. No code calls a tool directly — everything goes through tools/executor.py.
ToolRegistry maintains a dict of registered tools keyed by tool_name.
ToolRegistry
_tools: dict[str, RegisteredTool]
register(tool: BaseTool, config: ToolConfig) → None
get(tool_name: str) → RegisteredTool | None
list_healthy() → list[RegisteredTool]
update_health(tool_name, health: ToolHealth) → NoneRegisteredTool
tool_name: str
connector_id: str
capabilities: list[str]
scopes: list[str]
risk_default: RiskLevel
risk_map: dict[str, RiskLevel] — action-level overrides
supports_preview: bool
provider_type: ProviderType — native | mcp
health: ToolHealth
config_ref: str — DB ref; no secrets inline
ToolHealth
status: HealthStatus — healthy | degraded | unavailable
last_success_at: datetime | None
last_error: str | None
consecutive_errors: int
rate_limit_resets_at: datetime | NoneAll tool adapters implement BaseTool:
class BaseTool(ABC):
@property
def tool_name(self) -> str: ...
@property
def capabilities(self) -> list[str]: ...
@property
def scopes_required(self) -> list[str]: ...
def execute(
self,
action: str,
request: dict,
context: ToolCallContext, # trace_id, task_id, step_id,
# idempotency_key, autonomy_level,
# granted_scopes
) -> ToolResult: ...
def preview(
self,
action: str,
request: dict,
context: ToolCallContext,
) -> PreviewResult | None: ... # None = tool does not support previewToolCallContext carries all cross-cutting concerns; the tool implementation does not read system state directly.
tool_executor.call(tool_name, action, request, context) runs the following sequence. Every step that fails emits an AuditEvent before raising.
tool.scopes_required are present in context.granted_scopes. Failure raises ScopeViolation; not retried.risk = tool.risk_map.get(action, tool.risk_default), then adjusters applied. See safety/risk-and-gates.(autonomy_level, risk_level). If CONFIRM: create Approval record, emit gate.required, raise ApprovalRequired. If PREVIEW: run tool.preview(), raise PreviewRequired.idempotency_outcomes for context.idempotency_key. If found: emit tool_call.deduped, return cached result.tool_call.attempted, call tool.execute(). On success: write outcome, emit tool_call.succeeded. On retryable error: emit tool_call.failed. On transport failure: write unknown outcome, emit tool_call.unknown.The tool runtime enforces a strict secrets boundary. See ops/secrets for the full contract.
SecretsStore instance injected at construction.get_secret(connector_id, key) inside execute(), immediately before use.config_ref on RegisteredTool is a DB reference only — no credentials inline.