Automate policy violation handling and PII redaction using AISentinel webhooks, SDKs, and workflow orchestrators. This guide complements Webhook Integrations and LangChain examples with end-to-end remediation pipelines.
policy.decision.created events when an evaluation is denied or flagged.auditId references for compliance tracking.| Component | Responsibility |
|---|---|
| Webhook Receiver | Validates events and writes jobs to queue (SQS, RabbitMQ, Kafka). |
| Remediation Worker | Runs SDK scripts to redact or re-run agent tasks. |
| Ticketing Integration | Opens tickets when automation cannot resolve issues automatically. |
| Audit Store | Persists remediation outcome tied to AISentinel auditId. |
import os
from aisentinel import Client, ApiError
from redaction import redact_pii
client = Client(api_key=os.environ["AISENTINEL_API_KEY"], tenant_id=os.environ["AISENTINEL_TENANT_ID"])
def handle_job(job: dict) -> None:
decision = job["decision"]
if decision["decision"] != "deny":
return
content = decision["payload"]["content"]
redacted = redact_pii(content)
try:
reevaluation = client.policies.evaluate(
input=redacted,
rulepack=decision["rulepack"],
context={"sourceAuditId": decision["auditId"]},
)
except ApiError as exc:
if exc.status_code == 429:
raise RuntimeError("Rate limit hit during remediation") from exc
raise
if reevaluation.decision == "approve":
publish_fix(redacted, reevaluation.audit_id)
else:
escalate(decision, reevaluation)
redact_pii represents your custom sanitizer leveraging NLP or deterministic masking.publish_fix can push updates to data warehouses, vector stores, or CRM records.escalate sends context to incident management (PagerDuty, ServiceNow).decision.reasonCode == "pii_detected".auditId.5xx): Implement exponential backoff with jitter.403 policy_denied): Escalate to security analysts for manual review.mean_time_to_remediate (MTTR) and automation success rates.Automated remediation closes the loop between detection and resolution, ensuring AISentinel guardrails lead to rapid, auditable fixes.