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 a framework adapter
pip install 'pliuz[langchain]'
pip install 'pliuz[claude-agent]'    # Anthropic Claude Agent SDK
pip install 'pliuz[openai-agents]'   # OpenAI Agents SDK

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.
• Framework adapters — @gated_tool stacks on the framework's tool decorator so existing agents gate transparently. Dedicated adapters for LangChain, the Claude Agent SDK, and the OpenAI Agents SDK (any other framework still works via @gated on your own functions)

---

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(...): ...

originator says who initiated the action: "user", "cron", "system" or "agent". With "agent" the id is required and non-empty — Originator(type="agent") raises. That id is what the four-eyes check compares against, so an originator that cannot name which agent asked is an attestation with nothing behind it. Reading an approval back gives you OriginatorRead, which is wider on purpose: older approvals carry origins this vocabulary never had, and a client that raises reading them is worse than one that shows them as-is — which is exactly what this SDK did before 0.4.0.

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
)

Validación local vs servidor. PliuzValidationError cubre solo los 400 del servidor. Los fallos de validación que se detectan en el cliente antes de salir a red (p. ej. tool_name vacío, o context_messages por encima del cap de bytes) los lanza pydantic.ValidationError, no PliuzValidationError. Si quieres capturar ambos casos del mismo modo, atrapa también pydantic.ValidationError.

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"]),
    # ...
]

Claude Agent SDK integration (optional)

pip install 'pliuz[claude-agent]'

@gated_tool stacks on the SDK's @tool. The model sees the tool identically (same name, description, input_schema); only the execution path is gated.

from claude_agent_sdk import tool, create_sdk_mcp_server
from pliuz.adapters.claude_agent import gated_tool

@gated_tool(policy="refund", redact=["customer_id"])
@tool("issue_refund", "Issue a refund", {"customer_id": str, "amount_cents": int})
async def issue_refund(args: dict) -> dict:
    refund_id = stripe.refunds.create(customer=args["customer_id"], amount=args["amount_cents"]).id
    return {"content": [{"type": "text", "text": refund_id}]}

server = create_sdk_mcp_server(name="billing", version="1.0.0", tools=[issue_refund])

Or wrap an existing SdkMcpTool with wrap_tool(my_tool, policy="...", redact=[...]).

OpenAI Agents SDK integration (optional)

pip install 'pliuz[openai-agents]'

@gated_tool stacks on @function_tool, preserving name, description, and params_json_schema. The args arrive as a JSON string with a live ToolContext alongside; the adapter parses them to a flat dict so top-level redact paths match and the ToolContext never leaks into the audit payload or the idempotency key.

from agents import Agent, function_tool
from pliuz.adapters.openai_agents import gated_tool

@gated_tool(policy="refund", redact=["customer_id"])
@function_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

agent = Agent(name="Billing", instructions="...", tools=[issue_refund])

Or wrap an existing FunctionTool with wrap_tool(my_tool, policy="...", redact=[...]).

The OpenAI Agents SDK also ships a built-in needs_approval flag, but it only pauses the local run loop — it has no external routing, audit trail, or approver UI. Use this adapter when approval must leave the process and be recorded in Pliuz's hash-chained audit log.

---

Management API client

ManagementClient is the synchronous, programmatic counterpart to the dashboard: manage policies, tools, agents, and users as config-as-code, and pull the audit trail into your SIEM. Use it from CI/CD pipelines and back-office automation — not from inside a gated agent.

It authenticates with a machine token (plm_live_...) sent as Authorization: Bearer ..., read from PLIUZ_MANAGEMENT_KEY. This is a separate credential from the per-agent PLIUZ_API_KEY used by @gated and PliuzClient: a machine token has tenant-wide management scope, an agent key does not.

from pliuz import ManagementClient

with ManagementClient() as mgmt:  # reads PLIUZ_MANAGEMENT_KEY
    # config-as-code: policies / tools / agents / users
    policy = mgmt.create_policy(...)
    mgmt.update_policy(policy.id, ...)
    tools = mgmt.list_tools()
    agents = mgmt.list_agents()
    users = mgmt.list_users()

    # audit trail → SIEM
    events = mgmt.list_events()
    mgmt.verify_chain()                  # validate the hash chain
    export: ExportResult = mgmt.export_events()

See examples/management_basic.py for a runnable script.

---

Configuration

Env var	Default	Purpose
PLIUZ_API_KEY	(required)	Per-agent API key. pli_live_... format.
PLIUZ_MANAGEMENT_KEY	(required for ManagementClient)	Machine token for the Management API (CI/CD, config-as-code, SIEM); separate from the per-agent key. plm_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 dedupes on this key within a 24-hour window, scoped to the tenant and originating agent. An identical (tool_name, args, session_id) combination replays the same approval during that window, which makes HTTP retries safe. Reusing the same idempotency_key within 24 hours with different tool_args is rejected as an idempotency conflict. After the window expires, the same key can create a new approval.

With the default session_id=None, repeated identical calls from the same agent can still replay for up to 24 hours. 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.

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 is only the safety-floor backoff used if the server ignores the wait hint (it does NOT shorten the long-poll 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
ManagementClient	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
• claude_agent.py — Claude Agent SDK tool gated with @gated_tool
• openai_agents.py — OpenAI Agents SDK tool gated with @gated_tool

---

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.