""" VINCHY SDK — human-approval gate for AI agent actions. Wrap any risky action so a HUMAN must approve it before it runs, and every attempt is recorded on VINCHY's tamper-evident, Ed25519-signed audit trail. EU AI Act alignment: • Article 14 — human oversight (a person approves risky actions before execution) • Article 26 — deployer record-keeping (every action logged, retained, provable) Why an SDK (not just raw HTTP): a named `@requires_approval` decorator is a *statically visible* oversight control — code scanners and auditors can SEE the human-in-the-loop, which a bare runtime HTTP call does not expose. No third-party dependencies (stdlib only). Works with LangChain, CrewAI, AutoGen, n8n / Zapier code steps, or plain Python. Extras vs. a bare HTTP call (statically visible controls, parity with vinchy.js): • client-side RATE LIMIT (caps gated calls/min — guards resource exhaustion) • local AUDIT ECHO (logs the VINCHY approval id locally, to stderr) Usage: from vinchy import Vinchy vinchy = Vinchy() # reads VINCHY_API_KEY + VINCHY_AGENT_ID from env @vinchy.requires_approval("financial_transaction", detail=lambda iban, amt: {"recipient": iban, "amount": amt}) def pay_invoice(iban, amt): return bank.transfer(iban, amt) # runs ONLY after a human approves # or inline: if vinchy.gate("email_send", {"to": addr, "subject": s}): smtp.send(addr, s, body) """ from __future__ import annotations import os import sys import json import time import functools import urllib.request import urllib.error __all__ = ["Vinchy", "VinchyError", "ActionRejected", "ApprovalTimeout"] __version__ = "0.2.0" class VinchyError(Exception): """Base error for the VINCHY SDK.""" class ActionRejected(VinchyError): """A human rejected (or the gate expired) the requested action.""" class ApprovalTimeout(VinchyError): """No human decision arrived within the allowed wait window.""" class Vinchy: """Client for VINCHY's human-approval gate + tamper-evident audit trail.""" def __init__(self, api_key=None, agent_id=None, base_url="https://vinchy.ai", timeout=15, max_per_minute=30, audit_log=True): self.api_key = api_key or os.environ.get("VINCHY_API_KEY") self.agent_id = agent_id or os.environ.get("VINCHY_AGENT_ID") self.base_url = base_url.rstrip("/") self.timeout = timeout self.max_per_minute = max_per_minute # client-side rate limit self.audit_log = audit_log # local audit-id echo self._calls = [] if not self.api_key: raise VinchyError("Set VINCHY_API_KEY (project API key) or pass api_key=.") # ── client-side rate limit (sliding 60s window, parity with vinchy.js) ──── def _check_rate_limit(self): now = time.time() self._calls = [t for t in self._calls if now - t < 60.0] if len(self._calls) >= self.max_per_minute: raise VinchyError( f"Rate limit exceeded: > {self.max_per_minute} gated actions/min") self._calls.append(now) # ── local audit echo (to stderr, so stdout stays clean for pipelines) ───── def _audit_echo(self, action_type, status, approval_id=None): if self.audit_log: suffix = f" · {approval_id}" if approval_id else "" sys.stderr.write(f"[vinchy audit] {action_type} → {status}{suffix}\n") # ── HTTP helpers (stdlib only) ──────────────────────────────────────────── def _request(self, method, path, body=None, extra_headers=None): data = json.dumps(body).encode() if body is not None else None headers = dict(extra_headers or {}) headers["Authorization"] = f"Bearer {self.api_key}" if data is not None: headers["Content-Type"] = "application/json" req = urllib.request.Request(self.base_url + path, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: return json.loads(resp.read().decode() or "{}") except urllib.error.HTTPError as e: try: return json.loads(e.read().decode() or "{}") except Exception: raise VinchyError(f"HTTP {e.code} from {path}") # ── Core API ────────────────────────────────────────────────────────────── def request_approval(self, action_type, action_detail=None, agent_id=None, idempotency_key=None): """Submit a risky action to VINCHY's human-approval gate. Returns the approval record, e.g. {"status": "pending", "approval_id": "..."}. Pass idempotency_key to make a retry reuse the first approval instead of opening a duplicate (sent as the Idempotency-Key header). """ self._check_rate_limit() agent = agent_id or self.agent_id if not agent: raise VinchyError("Set VINCHY_AGENT_ID or pass agent_id=.") extra = {"Idempotency-Key": idempotency_key} if idempotency_key else None rec = self._request("POST", "/api/agent-action", { "agent_id": agent, "action_type": action_type, "action_detail": action_detail or {}, }, extra) self._audit_echo(action_type, rec.get("status", "unknown"), rec.get("approval_id")) return rec def get_status(self, approval_id): """Poll the current decision status for an approval id.""" return self._request("GET", f"/api/decide/{approval_id}").get("status") def wait_for_decision(self, approval_id, poll_every=5, max_wait=900): """Block until a human approves/rejects (or the gate expires).""" waited = 0 while waited <= max_wait: status = self.get_status(approval_id) if status in ("approved", "rejected", "expired"): return status time.sleep(poll_every) waited += poll_every raise ApprovalTimeout(f"No human decision within {max_wait}s.") def gate(self, action_type, action_detail=None, agent_id=None, wait=True, poll_every=5, max_wait=900, idempotency_key=None): """Hold a risky action for HUMAN APPROVAL before it runs. Returns True only if a human approved (or policy auto-approved) — i.e. it is safe to execute. Returns False if held and wait=False. Raises ActionRejected if a human rejected it. This is the EU AI Act Art. 14 human-oversight checkpoint. """ approval = self.request_approval(action_type, action_detail, agent_id, idempotency_key) status = approval.get("status") if status == "approved": # policy auto-approved (low risk) return True if status in ("rejected", "blocked"): raise ActionRejected(f"Action '{action_type}' was {status}.") approval_id = approval.get("approval_id") if not wait: return False # held in VINCHY Inbox; caller decides final = self.wait_for_decision(approval_id, poll_every, max_wait) self._audit_echo(action_type, final, approval_id) if final == "approved": return True raise ActionRejected(f"Action '{action_type}' was {final} by a human.") def requires_approval(self, action_type, detail=None, key=None): """Decorator: the wrapped function runs ONLY after human approval via VINCHY. `detail` may be a dict or a callable(*args, **kwargs) -> dict that describes the action for the human approver. `key` (optional idempotency, parity with vinchy.js `opts.key`) may be a value or a callable(*args, **kwargs) -> str; a retried call then reuses the first approval instead of opening a duplicate. EU AI Act Art. 14 (human oversight) + Art. 26 (record-keeping). """ def decorator(fn): @functools.wraps(fn) def wrapper(*args, **kwargs): payload = detail(*args, **kwargs) if callable(detail) else (detail or {}) idem = key(*args, **kwargs) if callable(key) else key # Human-in-the-loop checkpoint: blocks until a person approves. if self.gate(action_type, payload, idempotency_key=idem): return fn(*args, **kwargs) return None # held / not approved → the action does NOT run return wrapper return decorator