pliuz 0.1.1

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.dev, 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-dev.vercel.app	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

@gated automatically generates a deterministic idempotency_key per call — same args → same key. The backend dedupes within 24 h, so safe to retry from anywhere (job queues, retry loops, agent restarts).

@gated(policy="refund")
def refund(customer_id, cents):
    ...

# Calling twice in a 24h window → only 1 approval request created.
# Same human decision applies to both. No duplicate refunds.
refund("cus_123", 5000)
refund("cus_123", 5000)

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.dev/docs
• GitHub: https://github.com/mwhitex/pliuz
• Issues: https://github.com/mwhitex/pliuz/issues
• TypeScript SDK: @pliuz/sdk

---

License

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