loomcycle — async Python client

loomcycle is the async Python client for loomcycle's gRPC API (introduced in v0.5.5). As of v1.1.1 it covers all 42 gRPC RPCs — run streaming + continuation, the RFC AI interactive session (run_input + stream_run + an interactive=True flag), batch fan-out, run compaction, agent metadata + transcript, pause / resume / state, the snapshot lifecycle, the resolver probe, the full substrate-def family (incl. RFC AH volume_def), channel publish / subscribe / peek / ack / await / broadcast, and run-state streaming — through an ergonomic LoomcycleClient class with no need to import generated protobuf types in your application code.

Status

• Wraps loomcycle's gRPC server (LOOMCYCLE_GRPC_ADDR).
• Async-only (grpc.aio). Python 3.9+.
• Full parity with the gRPC service surface (42 RPCs).
• The TypeScript adapter (adapters/ts/) additionally exposes HTTP-only operations that have no gRPC RPC — memory-entry admin, interruptions, library enumeration, the LLM gateway, and whoami / list-users. Those are not reachable over gRPC and so are not in this client; use the HTTP+SSE surface for them.
• Production tag: 1.1.1 (42-RPC parity, version-aligned with the loomcycle v1.1.x line; ships on the python-v1.1.1 tag).

Install

pip install loomcycle

Quick start

import asyncio
from loomcycle import LoomcycleClient

async def main():
    async with LoomcycleClient(
        target="127.0.0.1:8788",
        auth_token="<LOOMCYCLE_AUTH_TOKEN>",
    ) as client:
        async for ev in client.run_streaming(
            agent="default",
            segments=[
                {
                    "role": "user",
                    "content": [
                        {"type": "trusted-text", "text": "Summarize loomcycle in one sentence."}
                    ],
                }
            ],
        ):
            if ev.type == "text":
                print(ev.text, end="", flush=True)
            elif ev.type == "tool_use":
                print(f"\n[tool_use: {ev.tool_use.name}]", flush=True)
            elif ev.type == "done":
                print(f"\n[done: {ev.stop_reason}]")

asyncio.run(main())

Capturing the run handle

The gRPC server's first two stream frames are synthetic registration frames — LoomcycleClient swallows them and exposes the captured IDs via an on_handle callback:

from loomcycle import LoomcycleClient, RunHandle

async def main():
    handle: RunHandle | None = None

    def capture(h: RunHandle):
        nonlocal handle
        handle = h
        print(f"agent_id={h.agent_id} session_id={h.session_id} run_id={h.run_id}")

    async with LoomcycleClient(target="127.0.0.1:8788") as client:
        async for ev in client.run_streaming(
            agent="default",
            segments=[...],
            on_handle=capture,
        ):
            ...

    # Use handle.session_id later to continue or read transcript.

API

All methods are coroutine methods on LoomcycleClient.

run_streaming / continue_session / each spawn_run_batch child also accept per-run sampling and compaction dict overrides (v0.8.0); an explicit temperature: 0.0 is preserved as deterministic.

Decision models

A decision model answers typed questions about a piece of text and returns each answer with its probabilities. It does not write: it reads the state you give it and answers a choice, a yes/no or a score, so there is no reply to parse and a call costs a few output tokens. Use it to route, gate, rank or grade text you already have. No run is involved; the tokens are charged to the caller and count against its token budget.

out = await client.decide(
    {"ticket": "My invoice for March was charged twice and I want my money back."},
    {
        "route": {
            "type": "choice",
            "instructions": "Which team should handle this ticket?",
            "criteria": {"billing": "invoices, refunds, charges", "support": "bugs, outages", "sales": None},
        },
        "urgent": {"type": "noul", "instructions": "Does this ticket need a reply within the hour?"},
        "detail": {
            "type": "score",
            "instructions": "How complete is the problem report?",
            "criteria": ["no detail", "some detail", "everything needed"],
        },
    },
)
out["answers"]["route"]   # {"type": "choice", "choice": "billing", "probabilities": {...}, "confidence": 0.91}
out["answers"]["urgent"]  # {"type": "noul", "noul": 0.316}
out["answers"]["detail"]  # {"type": "score", "score": 1.87, "legend": {...}, "probabilities": {...}, "confidence": 0.68}
out["usage"]              # {"input_tokens": 1059, "output_tokens": 4}

state and each question's criteria are plain dicts and lists, and each answer comes back as a dict: the JSON the wire carries is encoded and decoded for you.

• noul: 0 is an answer (a definite no), not a missing one — test it with "noul" in answer, never truthiness.
• The probabilities are not calibrated. Compare options within one answer; do not treat a fixed threshold as a guarantee.
• An answer is the model's own JSON, decoded and otherwise untouched, so a number is the number the model wrote and a field a later model adds is there to read.
• Leave model out for the deployment's default; name one only from list_decision_models().

A refusal raises with the decision's code on e.reason:

from loomcycle import LoomcycleError

try:
    out = await client.decide(state, questions)
except LoomcycleError as e:
    if e.reason == "prompt_too_large":
        ...  # shorten state and retry
    else:
        raise

Errors

Every method translates gRPC error codes to typed Python exceptions:

All exceptions inherit from LoomcycleError and preserve the original grpc.StatusCode on .code for log correlation.

One gRPC code covers several conditions (RESOURCE_EXHAUSTED is a full queue, a per-user quota or a token budget), so when the server says which one it is, that name is on .reason — token_limit_exceeded, backpressure, model_not_allowed, … — the same string the HTTP surface puts in an error body's code. Branch on .reason rather than on the message. It is None when the server attached none.

e.retryable: can the same call succeed later? Every exception has a read-only retryable, judged from its class and .reason. It is True only for a refusal that clears on its own:

• a BackpressureError whose .reason is backpressure, per_user_quota_exhausted, rate_limited or absent (a full queue, the caller's own run cap, starting runs too fast);
• an UnavailableError whose .reason is runtime_paused, tier_unavailable or pin_unavailable (the operator paused the runtime; no provider in the tier is up right now);
• a decide() refusal whose .reason is call_failed (UnavailableError: the decision-model call did not complete) or timeout (LoomcycleError from DEADLINE_EXCEEDED: it did not answer in time).

It is False for everything else — in particular a BackpressureError with .reason == "token_limit_exceeded", and the configured-run cap (a BackpressureError whose message ends "start or discard one first": the user already holds as many configured runs as allowed, and only starting or discarding one makes room; the server sends no reason for it over gRPC, so it is recognised by that message). Every RESOURCE_EXHAUSTED raises BackpressureError, so except BackpressureError: retry() alone retries a hard token budget for the rest of the month; test e.retryable first. A UnavailableError with no reason may be a dropped connection (worth a retry) or a feature the server does not have (not), which retryable cannot tell apart, so it is False: retry it on your own judgement. The default is False because a wrong True turns a refusal that never clears into an endless loop. The client never retries on its own, so bound your loop.

import asyncio

from loomcycle import LoomcycleError

for attempt in range(5):
    try:
        async for ev in client.run_streaming(...): ...
        break
    except LoomcycleError as e:
        if not e.retryable or attempt == 4:
            raise  # e.g. e.reason == "token_limit_exceeded": do not retry
        log.warning("loomcycle refused (reason=%s), retrying: %s", e.reason, e.message)
        await asyncio.sleep(2 ** attempt)

Allowed-hosts semantics

allowed_hosts mirrors the HTTP API's narrowing semantics:

This is enforced server-side in internal/tools/builtin/narrowing.go; allowed_hosts is a trust boundary — it must come from your application code, never from a model.

Verifying a signed hook delivery

A HookDef whose webhook body names a signing_secret signs every call with X-Loomcycle-Signature: t=<unix>,v1=<hex HMAC-SHA256 of t + "." + raw body>. Only the body is signed. It carries delivery_id (a retry keeps it), hook_name and owner; the X-Loomcycle-Delivery and X-Loomcycle-Hook headers are unsigned copies a replayer can change, so dedupe and identify the hook from the verified body, never from them. Verify the raw body, before parsing it:

from loomcycle import verify_loomcycle_delivery

raw = request.get_data()  # the exact bytes received
body = verify_loomcycle_delivery(raw, request.headers.get("X-Loomcycle-Signature", ""), SECRET)
if body is None:
    abort(401)
if already_seen(body["delivery_id"]):  # keep ids at least 300 s
    return previous_answer(body["delivery_id"])

verify_loomcycle_delivery returns the parsed body, or None when the signature fails or the body has no delivery_id/hook_name. verify_loomcycle_signature is the bare check: True only for a signature made within tolerance_sec (default 300) of now; a malformed header, a stale or future t, or a wrong MAC is False. Standard library only; the compare is hmac.compare_digest. Give each receiver its own signing secret, so a delivery captured at one cannot be replayed to another.

Development

# One-time setup:
python3 -m venv adapters/python/.venv
adapters/python/.venv/bin/pip install -e adapters/python[dev]

# Run tests (offline):
make python-test

# Regenerate stubs after editing proto/loomcycle.proto:
make python-proto

The package commits its generated loomcycle_pb2.py / loomcycle_pb2_grpc.py so end users don't need a working protoc to install. Re-run make python-proto whenever the proto changes.

Live integration test

To run the example end-to-end against a local loomcycle:

# In one shell — start loomcycle with gRPC enabled:
LOOMCYCLE_GRPC_ADDR=127.0.0.1:8788 \
LOOMCYCLE_AUTH_TOKEN=devtoken \
./bin/loomcycle --config loomcycle.yaml

# In another shell — run the example:
LOOMCYCLE_GRPC_ADDR=127.0.0.1:8788 \
LOOMCYCLE_AUTH_TOKEN=devtoken \
adapters/python/.venv/bin/python examples/python-cli/main.py

License

Apache-2.0. Same as loomcycle.