pliuz 0.2.2

Human-in-the-loop approval gates for AI agents — pause function calls for human review, then resume or abort.

pliuz — Python SDK

Human-in-the-loop approval gates for AI agents. Wrap any function in @gated to pause execution, route the call to a human approver via Pliuz, then resume — or abort — based on the decision. Every call is audited.

from pliuz import gated

@gated(policy="refund", redact=["customer.ssn"])
def issue_refund(customer_id: str, amount_cents: int) -> dict:
    return stripe.refunds.create(customer=customer_id, amount=amount_cents)

# Pauses. A human gets a Slack card. They click approve.
# Then your code runs — and Pliuz records the outcome.
result = issue_refund("cus_123", 5000)

---

Install

pip install pliuz
# or, with LangChain adapter
pip install 'pliuz[langchain]'

Requires Python ≥ 3.10.

Quickstart

1. Get an API key

Sign up at pliuz.com, create an agent in the dashboard, copy the key.

2. Set the env var

export PLIUZ_API_KEY=pli_live_...

3. Gate a function

from pliuz import gated

@gated(policy="risky_query")
def run_sql(query: str) -> list[dict]:
    return db.fetch(query)

# This call blocks until a human approves (or rejects, or it expires).
rows = run_sql("DELETE FROM users WHERE created_at < '2024-01-01'")

That's it. Pliuz handles the rest: idempotency, retries, audit logging, hash-chained event log.

What @gated does

your_fn(*args)                                    your_fn(*args)
     │                                                  │
     ▼                                                  ▼
 ┌──────────┐  POST /approvals                    ┌──────────┐
 │  pliuz   │ ───────────────►  ┌─────────────┐  │  pliuz   │
 │  gated   │                   │   Pliuz API  │  │  gated   │
 │          │  ◄─────────────── └─────────────┘  │          │
 └──────────┘     status=pending          ▲      └──────────┘
     │                                    │           │
     │              poll every 2s         │           ▼
     │ ────────────────────────────────► humans  ┌──────────┐
     ▼                                    │      │ original │
 ┌──────────┐  status=approved            │      │   fn()   │
 │ execute  │ ◄──────────────────────────┘      └──────────┘
 │ original │                                         │
 │   fn()   │  POST /:id/execution                    ▼
 └──────────┘ ─────────────────────────►       audit log
     │
     ▼
  result

Features

• Sync + async auto-detection via inspect.iscoroutinefunction
• Idempotency via deterministic key hashing — safe to retry from anywhere
• Client-side redaction — sensitive fields never leave your process plaintext
• Polling with backoff — uses time.sleep for sync, asyncio.sleep for async
• Single-shot execution reporting — closes the audit loop automatically
• Typed errors — PliuzRejectedError, PliuzApprovalExpiredError, PliuzApprovalTimeoutError, PliuzPolicyError, etc.
• LangChain adapter — @gated_tool stacks on @tool so existing LangChain agents gate transparently

---

API

@gated — the headline

@gated(
    policy="...",                    # Pliuz policy slug to bind (or None to match by tool_name)
    redact=["customer.ssn"],         # Dotted paths to strip BEFORE sending
    timeout_s=300,                   # Max polling duration (5 min default)
    poll_interval_s=2,               # Time between status checks
    tool_name="custom_name",         # Override fn.__name__
    client=PliuzClient(api_key="..."),  # Reuse a client (default: lazy from env)
    context_messages=["context for the human"],
    session_id="trace-abc",
    originator=Originator(type="user", id="user-42"),
)
def my_function(...): ...

Both forms work:

@gated  # uses defaults from env
def f(...): ...

@gated(policy="...")  # with options
def f(...): ...

@gated(...)
async def f(...): ...  # auto-detected

Errors

from pliuz import (
    PliuzError,                     # base
    PliuzApiError,                  # any 4xx/5xx from Pliuz API
        PliuzAuthError,             # 401 — bad/missing key
        PliuzForbiddenError,        # 403 — agent mismatch
        PliuzNotFoundError,         # 404 — approval not found
        PliuzConflictError,         # 409 — duplicate execution report
        PliuzValidationError,       # 400 — invalid body
        PliuzPolicyError,           # 422 — no policy matched
        PliuzRateLimitError,        # 429
        PliuzServerError,           # 5xx
    PliuzNetworkError,              # connection failed
    PliuzTimeoutError,              # request timed out
    PliuzRejectedError,             # human said no
    PliuzApprovalExpiredError,      # SLA expired before human decided
    PliuzApprovalTimeoutError,      # SDK polling gave up
)

Low-level client

from pliuz import PliuzClient, AsyncPliuzClient

with PliuzClient() as pliuz:
    resp = pliuz.create_approval(
        tool_name="refund",
        tool_args={"amount": 100},
        idempotency_key="abc-123",
    )

    full = pliuz.get_approval(resp.id)

    pliuz.report_execution(
        resp.id,
        status="success",
        latency_ms=42,
    )

# Async — same surface, every method awaitable
async with AsyncPliuzClient() as pliuz:
    resp = await pliuz.create_approval(...)

Redaction

from pliuz import apply_redaction

clean = apply_redaction(
    {"customer": {"ssn": "123-45-6789", "id": "cus_x"}},
    ["customer.ssn"],
)
# {"customer": {"ssn": "<redacted>", "id": "cus_x"}}

Supports dotted paths and [*] for array wildcards:

apply_redaction({"items": [{"card": "..."}]}, ["items[*].card"])

LangChain integration (optional)

pip install 'pliuz[langchain]'

from langchain_core.tools import tool
from pliuz.adapters.langchain import gated_tool

@gated_tool(policy="refund", redact=["customer.ssn"])
@tool
def issue_refund(customer_id: str, amount_cents: int) -> str:
    """Issue a refund to a customer."""
    return stripe.refunds.create(customer=customer_id, amount=amount_cents).id

# Pass to an agent like any other tool — the LLM sees it identically.
agent = create_react_agent(llm, tools=[issue_refund])

Or wrap existing tools compositionally:

from pliuz.adapters.langchain import wrap_tool

agent.tools = [
    wrap_tool(my_tool, policy="my_policy", redact=["secret"]),
    # ...
]

---

Configuration

Env var	Default	Purpose
PLIUZ_API_KEY	(required)	Per-agent API key. pli_live_... format.
PLIUZ_BASE_URL	https://pliuz.com	Override for self-hosted or staging environments.

All env vars can be passed as constructor kwargs:

PliuzClient(api_key="pli_live_...", base_url="https://pliuz.mycompany.com")

---

Production tips

Idempotency — read this before relying on it

@gated automatically generates a deterministic idempotency_key per call from (tool_name, args, session_id). The key is hashed over the pre-redaction args (only the truncated SHA-256 leaves your process), so two calls that differ only in a redacted field get different keys — each needs its own approval.

Important: the backend currently dedupes on this key with no time window — a (tool_name, args, session_id) combination that was approved once is replayed as approved on every identical future call (and a rejected one stays rejected). The dedupe exists for HTTP-retry safety, but with the default session_id=None it also applies across runs and days. If your agent legitimately repeats identical calls (crons, recurring jobs), set a per-run session_id so each run gets its own approval:

@gated(policy="refund", session_id=run_id)  # scope approvals to this run
def refund(customer_id, cents):
    ...

Within one run, retries are then safe: same args → one approval request, one human decision.

A bounded server-side dedupe window is planned; until then treat the replay semantics above as the contract.

When the approver edits the args

If a human decides with Edit & Approve, @gated executes the edited final_args (bound back onto your function's signature), never the original call. If the edited args don't bind to the signature, the SDK raises PliuzEditNotApplicableError — it refuses to run args the human did not approve. Fields the approver left as <redacted> are restored to the original values before execution.

Decision latency (long-poll)

@gated waits for the human decision via server long-poll: each wait call holds the connection open up to ~25s and returns the instant the approval is decided. Near-zero decision-delivery latency and ~12x fewer requests than fixed-interval polling, with no configuration. poll_interval_s now just caps the per-call wait window. The low-level client exposes it directly:

approval = pliuz.get_approval(approval_id, wait_seconds=25)  # long-poll

Custom timeouts per call

Don't set globally. Pick per use case:

• Live user-facing (chat agent waiting for a refund approval): timeout_s=30
• Background jobs (overnight batch): timeout_s=86400 (24 h)
• Critical security gates (deploy approvals): no @gated — use a synchronous Pliuz dashboard flow

When @gated raises mid-execution

If the human takes a long time and your polling times out, @gated raises PliuzApprovalTimeoutError. Your gated function never runs. The approval may still resolve later server-side — handle the retry policy yourself (PliuzClient.get_approval(id)).

What's NOT retried automatically

• POST /approvals without idempotency_key (would create duplicates)
• POST /approvals/:id/execution (single-shot — duplicate report = 409)

Everything else: GETs, 408, 429, 5xx → exponential backoff with full jitter, 3 retries default.

---

Versioning

This SDK follows SemVer. The current line is 0.x.y — minor versions may include breaking changes until 1.0.0.

Surface	Stability
gated, PliuzClient, AsyncPliuzClient	Stable across 0.x patches
pliuz.errors.*	Stable across 0.x patches
pliuz.adapters.*	Experimental — may shift before 1.0
Internal pliuz._*	Unstable — don't import

---

Examples

See examples/ for runnable scripts:

• basic.py — bare PliuzClient (no decorator)
• gated_basic.py — @gated happy path
• gated_async.py — async function with @gated
• langchain_agent.py — full LangChain agent with gated tools

---

Links

• Docs: https://pliuz.com/docs
• GitHub: https://github.com/pliuz/pliuz-py
• Issues: https://github.com/pliuz/pliuz-py/issues
• TypeScript SDK: @pliuz/sdk

---

License

Apache 2.0. The Pliuz platform itself is proprietary — this SDK is the public client only.