core.tools_consumer module

Tool-execution stream consumer for the dedicated tools service.

ToolExecConsumer mirrors core.stream_consumer.InboundStreamConsumer — blocking XREADGROUP on sg:stream:tools under the sg:tools group, an autoclaim sweep, and DLQ on repeated failure — but with two deliberate differences:

  • No per-channel distributed lock. Tool execution must run fully in parallel and load-balance across tools instances; user-visible message ordering is already enforced downstream by the gateway’s per-channel outbound lock.

  • The injected process_fn is expected to always reply (it pushes the result, or an error envelope, onto the caller’s reply stream) and to swallow tool-level exceptions. Only infrastructure failures propagate to the DLQ.

core.tools_consumer.autoclaim_attempts_key(stream_name, group_name)

Return an attempts hash colocated with stream_name in Cluster.

An untagged Redis key hashes over its complete text, so wrapping that same text in braces selects the identical slot for the sidecar. If the stream already carries a non-empty hash tag, reuse it. The group digest keeps independent consumer groups on one stream from sharing retry counters.

Return type:

str

Parameters:

  • stream_name (str)

  • group_name (str)

class core.tools_consumer.ToolExecConsumer(redis, consumer_name, process_fn, autoclaim_interval=30.0, autoclaim_min_idle=60000, *, stream_name='sg:stream:tools', group_name='sg:tools', require_durable_proof=False, delete_after_ack=False)

Bases: object

Consumes tool-execution requests and dispatches them to process_fn.

Parameters:

  • redis (Redis)

  • consumer_name (str)

  • process_fn (Callable[[dict[str, Any]], Awaitable[bool | None]])

  • autoclaim_interval (float)

  • autoclaim_min_idle (int)

  • stream_name (str)

  • group_name (str)

  • require_durable_proof (bool)

  • delete_after_ack (bool)

async start()
Return type:

None

async stop()
Return type:

None