Protecting Agentic Workflows

┌─────────────────────────────────────────────────────────────────────┐
│  PROTECTED AGENT WORKFLOW                                            │
│                                                                      │
│  ┌─────────┐     ┌──────────┐    ┌──────────┐    ┌─────────────┐   │
│  │  Input  │───▶│ Reasoning│───▶│Tool Call │───▶│   Output    │   │
│  │ Sanitize│     │  (Safe)  │    │(Validated)│   │ Re-protect  │   │
│  └─────────┘     └──────────┘    └──────────┘    └─────────────┘   │
│       ▲              ▲               ▲                ▲             │
│       │              │               │                │             │
│  ┌────┴──────────────┴───────────────┴────────────────┴────────┐   │
│  │            SEMANTIC GUARDRAILS (Continuous Monitoring)        │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Using AI Developer Edition, wrap my agent workflow with end-to-end protection. Scan every input for PII and secrets using Find and Protect, tokenize what is found, validate each tool call through Semantic Guardrails, block any request that looks like a data-extraction attempt, re-protect all outputs, and log every protection event for audit.

import protegrity_developer_python
import requests

protegrity_developer_python.configure(
    endpoint_url="http://localhost:8580/pty/data-discovery/v2/classify",
    classification_score_threshold=0.6,
    enable_logging=True,
    log_level="info"
)

GUARDRAIL_URL = "http://localhost:8581/pty/semantic-guardrail/v1.1/conversations/messages/scan"

class EndToEndProtectedAgent:
    """Agent with full AI Developer Edition protection pipeline."""
    
    def __init__(self, system_prompt: str):
        self.system_prompt = system_prompt
    
    def run(self, user_input: str) -> str:
        # Step 1: Scan and protect input
        safe_input = protegrity_developer_python.find_and_protect(user_input)
        
        # Step 2: Evaluate through guardrails
        payload = {
            "messages": [{"role": "user", "content": safe_input}],
            "processors": ["semantic", "pii"]
        }
        guard_check = requests.post(GUARDRAIL_URL, json=payload).json()
        if guard_check.get("action") == "BLOCK":
            return f"Request blocked: {guard_check.get('reason')}"
        
        # Step 3: Agent reasons on safe data
        response = self._reason(safe_input)
        
        # Step 4: Re-protect output
        safe_output = protegrity_developer_python.find_and_protect(response)
        
        return safe_output
    
    def _reason(self, safe_input: str) -> str:
        """Agent reasoning on protected data."""
        return llm.generate(self.system_prompt + "\n" + safe_input)

Using AI Developer Edition, review my repository for hardcoded API keys, connection strings with credentials, tokens in .env files, and PII in comments or test fixtures. Use Data Discovery to detect the secrets, replace each one with a token in TKN-XXX format before Cursor processes the code, and apply Semantic Guardrails so only read-only API calls are allowed. Block any write or delete operation unless explicitly approved, and log every detected secret for security review.

import protegrity_developer_python
import os

class SecureCodeAgent:
    """Agent that safely processes code repositories."""
    
    def analyze_repo(self, repo_path: str) -> dict:
        # Step 1: Discover and protect secrets in codebase
        secrets_found = 0
        for root, dirs, files_list in os.walk(repo_path):
            for file in files_list:
                if file.endswith((".py", ".js", ".java", ".yml", ".env")):
                    filepath = os.path.join(root, file)
                    with open(filepath, "r") as f:
                        content = f.read()
                    protected = protegrity_developer_python.find_and_protect(content)
                    if protected != content:
                        secrets_found += 1
                        with open(filepath + ".safe", "w") as f:
                            f.write(protected)
        
        # Step 2: Agent analyzes safe codebase
        analysis = agent.analyze(repo_path)
        
        return {
            "secrets_found": secrets_found,
            "secrets_protected": True,
            "analysis": analysis
        }

Using AI Developer Edition, orchestrate my multi-agent pipeline safely. Give each agent only the data it needs (least privilege), tokenize sensitive fields between agent handoffs, validate every agent action through Semantic Guardrails, and prevent any agent from accessing another agent's raw context. Log and audit all inter-agent communication.

import protegrity_developer_python
import requests

GUARDRAIL_URL = "http://localhost:8581/pty/semantic-guardrail/v1.1/conversations/messages/scan"

class SecureOrchestrator:
    """Orchestrates multiple agents with data protection between them."""
    
    def execute_pipeline(self, task: str, agents: list) -> str:
        context = task
        
        for i, agent in enumerate(agents):
            # Protect data before passing to next agent
            safe_context = protegrity_developer_python.find_and_protect(context)
            
            # Validate agent action through guardrails
            payload = {
                "messages": [{"role": "user", "content": f"Agent {agent.name}: {agent.planned_action}"}],
                "processors": ["semantic"]
            }
            action_check = requests.post(GUARDRAIL_URL, json=payload).json()
            
            if action_check.get("action") == "BLOCK":
                return f"Pipeline halted: Agent '{agent.name}' blocked - {action_check.get('reason')}"
            
            # Agent processes safe data
            result = agent.execute(safe_context)
            
            # Re-protect output before passing to next agent
            context = protegrity_developer_python.find_and_protect(result)
        
        return context

Using AI Developer Edition, help me debug production logs safely. Anonymize all user identifiers (emails, IDs, names) and mask tokens and session data, but preserve timestamps, error types, and stack traces so I still have enough context for root-cause analysis. Never expose real user data in the debug output.

import protegrity_developer_python

class SafeDebugger:
    """Debug production issues without exposing user data."""
    
    def debug(self, log_content: str) -> str:
        # Redact user data from logs
        safe_logs = protegrity_developer_python.find_and_redact(log_content)
        
        # Send to AI for analysis
        diagnosis = ai_model.analyze(safe_logs)
        
        return diagnosis

Using AI Developer Edition, protect my RAG pipeline. Anonymize sensitive fields in retrieved documents before they reach the LLM context, use Semantic Guardrails to prevent PII in the output, allow aggregate insights but block responses about individual records, and re-protect any response that references specific people.

import protegrity_developer_python
import requests

GUARDRAIL_URL = "http://localhost:8581/pty/semantic-guardrail/v1.1/conversations/messages/scan"

class ProtectedRAG:
    """RAG pipeline with full data protection."""
    
    def query(self, user_query: str) -> str:
        # Step 1: Evaluate query through guardrails
        payload = {
            "messages": [{"role": "user", "content": user_query}],
            "processors": ["semantic", "pii"]
        }
        query_check = requests.post(GUARDRAIL_URL, json=payload).json()
        if query_check.get("action") == "BLOCK":
            return "Query blocked: potential data extraction attempt"
        
        # Step 2: Retrieve documents
        documents = vector_db.retrieve(user_query, top_k=5)
        
        # Step 3: Redact PII from retrieved content
        safe_docs = [protegrity_developer_python.find_and_redact(doc) for doc in documents]
        
        # Step 4: LLM reasons on redacted context
        response = llm.generate(
            context=safe_docs,
            query=user_query,
            system="Provide insights without mentioning specific individuals."
        )
        
        # Step 5: Validate response for PII leakage
        resp_payload = {
            "messages": [{"role": "assistant", "content": response}],
            "processors": ["pii"]
        }
        response_check = requests.post(GUARDRAIL_URL, json=resp_payload).json()
        if response_check.get("action") == "BLOCK":
            response = protegrity_developer_python.find_and_redact(response)
        
        return response

"""
Complete Cursor Agent with AI Developer Edition Protection
-------------------------------------------------------
Demonstrates: protect → reason → validate → execute → re-protect
"""
import protegrity_developer_python
import requests

protegrity_developer_python.configure(
    endpoint_url="http://localhost:8580/pty/data-discovery/v2/classify",
    classification_score_threshold=0.6,
    enable_logging=True,
    log_level="info"
)

GUARDRAIL_URL = "http://localhost:8581/pty/semantic-guardrail/v1.1/conversations/messages/scan"

# ═══════════════════════════════════════════════════════════════
# PROMPT PROTECTION: Sanitize input before AI processing
# ═══════════════════════════════════════════════════════════════
def protect_input(user_input: str) -> str:
    """Scans input for PII/secrets and replaces with tokens."""
    return protegrity_developer_python.find_and_protect(user_input)

# ═══════════════════════════════════════════════════════════════
# REASONING: Agent processes safe data only
# ═══════════════════════════════════════════════════════════════
def reason(safe_prompt: str) -> str:
    """Agent reasons on safe data only."""
    response = agent.run(safe_prompt)
    return response

# ═══════════════════════════════════════════════════════════════
# TOOL CALL: Validate intent and restore data for execution
# ═══════════════════════════════════════════════════════════════
def execute_tool(action: str, data: dict, auth) -> any:
    """Restores real data only for approved execution."""
    payload = {
        "messages": [{"role": "user", "content": f"Execute action: {action}"}],
        "processors": ["semantic"]
    }
    guard_check = requests.post(GUARDRAIL_URL, json=payload).json()
    if guard_check.get("action") != "BLOCK":
        real_data = protegrity_developer_python.find_and_unprotect(str(data))
        result = call_tool(action, real_data)
        return result
    else:
        raise SecurityError(f"Action '{action}' blocked by guardrails")

# ═══════════════════════════════════════════════════════════════
# OUTPUT PROTECTION: Re-apply protection to response
# ═══════════════════════════════════════════════════════════════
def protect_output(response: str) -> str:
    """Re-applies protection to response to prevent data leakage."""
    return protegrity_developer_python.find_and_protect(response)

# ═══════════════════════════════════════════════════════════════
# CONTINUOUS GUARDRAILS: Monitor for unsafe behavior
# ═══════════════════════════════════════════════════════════════
def monitor_workflow(step: str, data: any) -> None:
    """Semantic Guardrails continuously score risk across the workflow."""
    payload = {
        "messages": [{"role": "user", "content": str(data)}],
        "processors": ["semantic", "pii"]
    }
    risk = requests.post(GUARDRAIL_URL, json=payload).json()
    if risk.get("action") == "BLOCK":
        block_or_escalate(risk)

# ═══════════════════════════════════════════════════════════════
# FULL WORKFLOW: End-to-end protected execution
# ═══════════════════════════════════════════════════════════════
def protected_workflow(user_input: str, auth) -> str:
    """
    Complete protected agent workflow:
    Input → Protect → Reason → Validate → Execute → Re-protect → Output
    """
    
    # 1. Protect input
    safe_input = protect_input(user_input)
    monitor_workflow("input", safe_input)
    
    # 2. Agent reasons on safe data
    response = reason(safe_input)
    monitor_workflow("reasoning", response)
    
    # 3. If tool call needed, validate and execute
    if requires_tool_call(response):
        action, data = parse_tool_call(response)
        monitor_workflow("tool_call", {"action": action, "data": data})
        result = execute_tool(action, data, auth)
        monitor_workflow("tool_result", result)
    else:
        result = response
    
    # 4. Protect output
    safe_output = protect_output(str(result))
    monitor_workflow("output", safe_output)
    
    return safe_output
# ═══════════════════════════════════════════════════════════════
# EXAMPLE: Cursor agent reads repo and calls API
# ═══════════════════════════════════════════════════════════════
# AI Developer Edition:
# - Masks secrets in prompt
# - Enforces safe tool access
# - Blocks data leakage across steps
user_request = "Analyze my repo and deploy the payment service"
result = protected_workflow(user_request, auth=developer_credentials)
# 1. Discovery finds API keys in code → tokenized
# 2. Agent analyzes code structure safely
# 3. Guardrails validate deployment action
# 4. Real credentials restored only for approved deploy
# 5. Output re-protected before returning to developer