controlley 0.1.0

Gate any AI-agent action behind a human approval — propose, block until a person decides on their phone or the console, then run the side effect only if approved. Stdlib-only, typed.

controlley (Python)

A dependency-light, pip-installable Python client for the Controlley Tier-2 REST API. Gate any agent action behind a human approval: propose an action, block until a person decides on their phone or the console, then run your side effect only if approved.

• Stdlib only — urllib, hmac, hashlib, json, os, stat, time, pathlib. No third-party runtime dependencies.
• Typed — full type hints, ships a py.typed marker.
• Python 3.9+.
• Mirrors the TypeScript SDK (@controlley/sdk) surface: gate, request_approval, call_tool, submit_async, get_action, wait_for_decision.

Install

pip install controlley

From this subdirectory of the repo (editable or regular):

pip install ./clients/python
# or, from inside the directory:
pip install .

Straight from the repo, no clone:

pip install "git+https://github.com/andreidemianov8/controlley.git#subdirectory=clients/python"

Authenticate

This client uses API-key auth. Mint a key in the console at /api-keys (the ck_... token is shown exactly once) and put it in the environment:

export CONTROLLEY_API_KEY=ck_your_key_here
export CONTROLLEY_URL=http://localhost:3001   # your Controlley origin (no trailing slash)

PowerShell:

$env:CONTROLLEY_API_KEY = "ck_your_key_here"
$env:CONTROLLEY_URL = "http://localhost:3001"

Keep the key secret: it authenticates every proposal and derives the callback-signing secret (sha256(your ck_ token)).

Or pair keylessly — no key pasted onto the box

Instead of minting and copying a ck_ key, a headless bot can self-obtain a scoped token via device pairing (RFC 8628-adapted). Nothing is copied onto the machine: the bot prints a code, you approve once on your phone or the console, and the issued ctl_ agent token is cached locally.

from controlley import Controlley

# Prints a verification URL + user code, waits for you to approve, then caches
# the token at ~/.controlley/credentials.json (mode 0600) and returns a client.
ctl = Controlley.pair("alpaca-bot", scope_tools=["broker.*"])

# A cached, unexpired token short-circuits the flow on the next run — so on a VPS
# you pair once, then `systemctl start tradingbot` runs fully non-interactively.

Or from the shell (see CLI):

controlley pair --name alpaca-bot --scope-tools 'broker.*'

The paired token is propose-only, tool-scoped, short-lived, and revocable: it can propose actions but can never approve them or move funds. The plaintext token reaches only the bot and its local cache — never the console, never a log.

Quickstart — gate()

gate() runs your function only if the action is approved; otherwise it raises ControlleyRejected.

from controlley import Controlley, ControlleyRejected

ctl = Controlley()  # reads CONTROLLEY_API_KEY + CONTROLLEY_URL from the env

def place_order():
    # ... your real broker call. Reached only after a human approves.
    return {"status": "placed"}

try:
    result = ctl.gate(
        "BUY 100 AAPL @ market (~$18,500 notional)",
        place_order,
        details="Market order — fill price may differ from the reference.",
        risk="high",
    )
    print("approved and executed:", result)
except ControlleyRejected as r:
    print(f"not placed [{r.decision}]: {r.feedback}")

risk is an advisory hint — the classifier may raise its own verdict from it, never lower it.

Safe introduction — add Controlley to a live loop without breaking it

Adding Controlley to a running bot must not change its behavior on day one. The same gate() gives you a zero-risk rollout ladder, plus predictable failure and a one-line escape hatch.

The rollout ladder

1. Install — importing does nothing until you call gate().
2. Monitor mode — gate(..., mode="monitor"). Your bot runs every action immediately, exactly as before; Controlley does not block. The proposal is fired out-of-band (recorded / pushed to your phone) so you see what it would gate, with zero behavior change. Run it in production for a day.
3. Author policies from what you observed.
4. Enforce mode — mode="enforce" (the default). Now it actually gates. By this point you've tuned policies against real traffic, so the flip is boring.

# Day one: observe only. place_order() ALWAYS runs; nothing is ever blocked.
ctl = Controlley(mode="monitor")
ctl.gate("BUY 100 AAPL", place_order, risk="high")   # runs immediately

# Later: enforce (client default is already "enforce").
ctl = Controlley()
ctl.gate("BUY 100 AAPL", place_order, risk="high")   # gates for real

Failure is a configured choice, never a surprise

If Controlley is unreachable (network, outage, expired token, rate limit), on_unavailable decides what happens — there is no dangerous implicit default:

on_unavailable	behavior
"proceed"	fail-open — run the action, record a bypassed marker (the loop keeps working; you're told governance was skipped)
"block"	fail-closed — skip the action, raise ControlleyUnavailable (the bot decides how to handle a skipped action)
"queue"	defer — record a pending marker locally, skip for now, return None

Recommended pattern: proceed for low-risk, block for high-risk — set a client default and override per call.

ctl = Controlley(on_unavailable="block")            # safe default for risky work
ctl.gate("ping heartbeat", beat, on_unavailable="proceed")   # low-risk override

Never hang · never double-execute

• Bounded wait — gate_timeout_s (default 10s, well under the server sync wait) caps how long gate() blocks before applying the degradation policy.
• Circuit breaker — after breaker_threshold consecutive failures the client stops calling and degrades immediately (no hammering, no per-call hang) until a lightweight GET /healthz probe recovers, then it resumes gating.
• Auto-idempotency — a retry after a blip never places two orders. gate() derives a stable idempotency key from action_key (if given) or a hash of the summary (for call_tool, of tool_key + canonicalized params), so the server replays the original decision. Override with idempotency_key=....

ctl = Controlley(
    gate_timeout_s=10.0,
    breaker_threshold=3,
    breaker_cooldown_s=30.0,
    async_reporter=lambda decision: log(decision),  # monitor cards + markers
)
ctl.gate("BUY 100 AAPL", place_order, action_key="order-42")  # stable retry key

Kill switch — pull Controlley out in one line

ctl = Controlley(enabled=False)   # every gate() is a pass-through no-op

Or, without touching code, set CONTROLLEY_DISABLED=1 in the environment (checked live on every call) so an operator can disable Controlley during an incident and re-enable it by unsetting the variable.

Decorator and context-manager sugar

Same semantics as gate(), for teams that prefer them:

@ctl.gated(lambda qty, sym: f"BUY {qty} {sym}", risk="high")
def place_order(qty, sym):
    return broker.buy(qty, sym)      # reached only after approval

with ctl.gate_block("Deploy to prod", risk="high"):
    deploy()                          # the block runs only if approved

gate_block raises ControlleyRejected on a decline and ControlleyUnavailable on a fail-closed outage (a with body can't be silently un-run, so "queue" raises ControlleyUnavailable here rather than returning None).

Other calls

# request_approval(): get the Decision without auto-running anything.
decision = ctl.request_approval("Delete the staging database")
if decision.approved:
    ...

# Governed tool call by key — Controlley executes the tool on approval; the
# result lands in decision.result.
out = ctl.call_tool("slack.post_message", {"channel": "#general", "text": "shipped"})

# Async intake + poll (or receive a signed callback).
action_id = ctl.submit_async(
    summary="Nightly reconciliation run",
    callback_url="https://you.example/cb",
)
final = ctl.wait_for_decision(action_id, timeout=600)  # seconds

Every call returns a Decision:

field	meaning
status	executed / approved / rejected / revision_requested / blocked / pending / failed
approved	True only when approved and executed
action_id	the action id (or None for a block that created no action)
decision	the human's verbatim decision (approved / edited / rejected / …)
feedback	text the approver attached
result	the execution/adapter result when executed
edited_params	the corrected params when the approver edited before approving
policy	the matched policy on a block
message	a human-readable explanation on block / pending / failure

Verifying a callback signature

If you pass callback_url to submit_async, Controlley sends one signed JSON callback on the action's terminal transition. Verify it against the raw bytes before parsing:

import json
from controlley import verify_callback_signature

def handle_callback(raw_body: bytes, headers: dict[str, str], api_key: str) -> None:
    signature = headers.get("X-Controlley-Signature")
    if not verify_callback_signature(raw_body, signature, api_key):
        return  # drop — bad signature
    payload = json.loads(raw_body)
    # payload: { action_id, status, decision, feedback, execution_result, summary }
    ...

The signing secret is sha256(your ck_ token) — derived on both sides, never transmitted.

CLI

Installing the package puts a controlley command on your PATH:

# Keyless device pairing: prints a code, waits for phone/console approval, then
# caches the issued ctl_ token at ~/.controlley/credentials.json (mode 0600).
controlley pair --name alpaca-bot --scope-tools 'broker.*'

# Print the resolved origin + a redacted key fingerprint (never the key/token).
controlley whoami

controlley pair accepts --name (agent name shown on the approval card, 1–80 chars), --scope-tools (comma-separated tool-key allow-list, e.g. broker.*), and the top-level --base-url. A cached, unexpired credential short-circuits it.

Public API

from controlley import (
    Controlley,                 # the client
    Decision,                   # normalized outcome dataclass
    ControlleyError,            # typed error (carries the server's `code`)
    ControlleyRejected,         # raised by gate() on any non-approval
    ControlleyUnavailable,      # raised by gate() on a fail-closed outage
    verify_callback_signature,  # HMAC callback verification
    __version__,
)

Controlley(api_key=None, base_url=None, *, default_wait_ms=300_000, poll_interval_ms=2_000, timeout_s=30.0, enabled=True, mode="enforce", on_unavailable="block", gate_timeout_s=10.0, breaker_threshold=3, breaker_cooldown_s=30.0, async_reporter=None) — api_key/base_url fall back to CONTROLLEY_API_KEY / CONTROLLEY_URL; the safe-intro args set client-level defaults that per-call mode/on_unavailable override.

Methods: request_approval(summary, *, details=None, risk=None, idempotency_key=None, action_key=None, wait_ms=None), gate(summary, fn, *, details=None, risk=None, idempotency_key=None, action_key=None, wait_ms=None, mode=None, on_unavailable=None), gated(summary_fn, *, risk=None, details=None, mode=None, on_unavailable=None, action_key=None) (decorator), gate_block(summary, *, risk=None, details=None, mode=None, on_unavailable=None, action_key=None) (context manager), call_tool(tool_key, params, *, idempotency_key=None, action_key=None, wait_ms=None), submit_async(*, summary=None, details=None, risk=None, tool_key=None, params=None, callback_url=None, idempotency_key=None) -> str, get_action(action_id) -> Decision, wait_for_decision(action_id, *, timeout=None, interval=None) -> Decision.

Classmethod: Controlley.pair(client_name, *, scope_tools=None, base_url=None, credentials_path=None, reporter=None, timeout_s=30.0, enabled=True, mode="enforce", on_unavailable="block", gate_timeout_s=10.0, breaker_threshold=3, breaker_cooldown_s=30.0, async_reporter=None) -> Controlley — runs the device-pairing flow (or short-circuits on a cached unexpired token) and returns a ready client authenticated by the issued ctl_ token. The safe-intro args set the same client-level defaults as the constructor and are forwarded to the returned client, so a paired bot can set e.g. mode="monitor" once at connect time (parity with the TS connectControlley) instead of on every gate() call.

Development

python -m unittest discover -s tests   # from clients/python
python -m py_compile controlley/*.py tests/*.py