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.

from loomcycle import BackpressureError

try:
    async for ev in client.run_streaming(...): ...
except BackpressureError as e:
    log.warning("loomcycle backpressure (code=%s): %s", e.code, e.message)

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.

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.