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.

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:

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.