core.remote_tool_registry module

Inference-side stand-in for ToolRegistry that delegates tool execution.

RemoteToolRegistry duck-types the read surface of tools.ToolRegistry (OpenAI schemas, tool_names, repeat_allowed_tools, is_allowed …) from the Redis-published tool catalog, so the inference tier can present tools to the LLM and rank them without importing any handler module. Its call() routes each invocation through three tiers:

GATEWAY_PINNED_TOOLS -> delegate to the gateway (live discord.py client).
INFERENCE_PINNED_TOOLS (and nested calls, and tools_force_in_process) -> run in-process via the small local registry the inference service loads.
everything else -> delegate to the dedicated tools service over sg:stream:tools; the result + write-back (sent_files media bytes, injected_tools, sent_rich_messages) come back on a per-worker reply stream and are merged onto the live ctx so the LLM loop is unchanged.

In tools_service_mode == "in_process" (the safe default) every call runs on the local registry, exactly as before the split. The remote path is opt-in.

Drop-in: the executor and transport only touch the registry’s read surface plus call(), so swapping ToolRegistry for this requires no changes there.

exception core.remote_tool_registry.AmbiguousRemoteToolResult

Bases: RuntimeError

The remote request may have executed, so replay is not safe.

exception core.remote_tool_registry.DurableRemoteToolPending

Bases: RuntimeError

Durable transport is unresolved; preserve and recover the outer slot.

code = 'DURABLE_REMOTE_TOOL_PENDING'

preserve_outer_checkpoint = True

exception core.remote_tool_registry.DurableRemoteToolRejected

Bases: RuntimeError

A durable request was definitely rejected with no external effect.

Unlike DurableRemoteToolPending, this exception deliberately does not carry preserve_outer_checkpoint. The executor records it as a failed terminal tool result instead of leaving the outer slot EXECUTING.

code = 'DURABLE_REMOTE_TOOL_REJECTED'

class core.remote_tool_registry.RemoteToolRegistry(*, event_bus, redis, config, local_registry, worker_id)

Bases: object

Catalog-backed, delegation-capable replacement for ToolRegistry.

Parameters:

event_bus (RedisEventBus) – bus used to publish tool-exec requests.
redis (Any) – raw (decode_responses=False) async client — used to read the catalog and consume the per-worker reply stream.
config (Any) – live config.Config (routing mode, timeouts, overrides).
local_registry (Any) – a real tools.ToolRegistry for in-process tiers (gateway-pinned delegation needs ctx.adapter; inference-pinned + in_process mode run handlers here). During rollout it may hold every tool; in the lean end state only INFERENCE_PINNED_TOOLS.
worker_id (str) – stable id for this inference worker (reply-stream name).

async start()

Load the catalog and start the reply reader + catalog watcher.

In in_process mode this is a no-op: no catalog is read (the read surface stays delegated to the local registry) and no background tasks are spawned, so the inference worker behaves exactly as before the split.

Return type:: None

async stop()

Return type:: None

async reload_catalog()

(Re)load the catalog from Redis. Returns True if a catalog was found.

Return type:: bool

get_openai_tools()

Return type:: list[dict[str, Any]]

get_openai_tools_by_names(names)

Return type:: list[dict[str, Any]]
Parameters:: names (set[str])

tool_names()

Return type:: frozenset[str]

repeat_allowed_tools()

Return type:: frozenset[str]

list_tools()

Return type:: list[Any]

set_permissions(permissions)

Return type:: None
Parameters:: permissions (dict[str, list[str]])

is_allowed(tool_name, user_id)

Return type:

bool

Parameters:

tool_name (str)
user_id (str)

property has_tools: bool

property task_manager: Any

async call(name, arguments, user_id='', ctx=None, *, nested=False, operation_id=None, durable_execution=False)

Return type:

str

Parameters:

name (str)
arguments (dict[str, Any])
user_id (str)
ctx (Any)
nested (bool)
operation_id (str | None)
durable_execution (bool)

is_durable_replay_safe(name)

Whether name is fenced by the remote service result ledger.

Return type:: bool
Parameters:: name (str)

async refresh_durable_replay_safety(name)

Revalidate a previously durable route after process/catalog churn.

A turn checkpoint’s first writer owns the replay classification. When such a slot is recovered by a fresh inference process, the in-memory protocol confirmation starts false. Query the registered consumer fleet before allowing that already-replay-safe slot to execute again; a routing change to local/pinned execution remains fail-closed.

Return type:: bool
Parameters:: name (str)