E-Commerce Compliance

Your AI agent is handling EU customers, processing returns with payment info, and storing conversation logs. Is that GDPR compliant? PCI-DSS compliant?

Non-compliance isn't just a fine risk - it's an existential business risk. Let's build compliant AI agents.

The Compliance Landscape

GDPR Compliance for AI Agents

Key Requirements

Implementation

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Dict
from enum import Enum

class ConsentType(Enum):
    ESSENTIAL = "essential"  # Required for service
    ANALYTICS = "analytics"  # Usage analytics
    PERSONALIZATION = "personalization"  # AI personalization
    MARKETING = "marketing"  # Marketing communications

@dataclass
class ConsentRecord:
    user_id: str
    consent_type: ConsentType
    granted: bool
    timestamp: datetime
    source: str  # Where consent was collected
    version: str  # Privacy policy version

class GDPRCompliantAgent:
    """AI agent with GDPR compliance built-in"""

    def __init__(self, agent, consent_store, data_store):
        self.agent = agent
        self.consent_store = consent_store
        self.data_store = data_store

    async def handle(
        self,
        query: str,
        context: dict,
        user_id: str
    ) -> str:
        # Check for data subject requests
        if self._is_data_request(query):
            return await self._handle_data_request(query, user_id)

        # Check consent before processing
        consent = await self.consent_store.get_consent(user_id)

        # Apply data minimization based on consent
        filtered_context = self._filter_context_by_consent(context, consent)

        # Process with limited context
        response = await self.agent.handle(query, filtered_context)

        # Log interaction with retention policy
        await self._log_with_retention(user_id, query, response, consent)

        return response

    def _is_data_request(self, query: str) -> bool:
        """Detect GDPR data subject requests"""
        data_request_patterns = [
            r"what (data|information) do you have (about|on) me",
            r"delete (my|all) (data|information|account)",
            r"(exercise|invoke) my (gdpr|privacy) rights",
            r"(right to be forgotten|erasure)",
            r"download my data",
            r"stop processing my data",
        ]
        query_lower = query.lower()
        return any(re.search(p, query_lower) for p in data_request_patterns)

    async def _handle_data_request(
        self,
        query: str,
        user_id: str
    ) -> str:
        """Handle GDPR data subject requests"""

        if "delete" in query.lower() or "erasure" in query.lower():
            # Right to erasure (Article 17)
            return await self._initiate_erasure_request(user_id)

        elif "download" in query.lower() or "what data" in query.lower():
            # Right of access (Article 15)
            return await self._initiate_access_request(user_id)

        elif "stop processing" in query.lower():
            # Right to object (Article 21)
            return await self._initiate_objection(user_id)

        return (
            "I understand you want to exercise your privacy rights. "
            "I've escalated this to our privacy team who will respond within 30 days. "
            "You can also email privacy@example.com directly."
        )

    async def _initiate_access_request(self, user_id: str) -> str:
        """Initiate Subject Access Request"""
        request_id = await self.data_store.create_sar_request(user_id)

        return (
            f"I've initiated a data access request (Reference: {request_id}). "
            f"You'll receive a copy of all personal data we hold about you within 30 days. "
            f"This includes: conversation history, order information, preferences, and any analytics data. "
            f"The data will be sent to your registered email address in a machine-readable format."
        )

    async def _initiate_erasure_request(self, user_id: str) -> str:
        """Initiate Right to Erasure request"""
        request_id = await self.data_store.create_erasure_request(user_id)

        return (
            f"I've initiated a data deletion request (Reference: {request_id}). "
            f"Please note: We'll delete all personal data except what we're legally required to keep "
            f"(e.g., financial records for tax purposes). "
            f"You'll receive confirmation within 30 days. "
            f"This action cannot be undone - are you sure you want to proceed?"
        )

    def _filter_context_by_consent(
        self,
        context: dict,
        consent: Dict[ConsentType, bool]
    ) -> dict:
        """Apply data minimization based on consent"""
        filtered = {"essential": context.get("essential", {})}

        # Only include personalization data if consented
        if consent.get(ConsentType.PERSONALIZATION, False):
            filtered["preferences"] = context.get("preferences", {})
            filtered["browsing_history"] = context.get("browsing_history", [])

        # Order history is essential for order-related queries
        filtered["orders"] = context.get("orders", [])

        return filtered

    async def _log_with_retention(
        self,
        user_id: str,
        query: str,
        response: str,
        consent: Dict
    ):
        """Log with GDPR-compliant retention"""
        await self.data_store.log_interaction(
            user_id=user_id,
            query=query,
            response=response,
            retention_days=self._get_retention_period(consent),
            legal_basis="legitimate_interest",  # or "consent"
            processing_purpose="customer_support"
        )

    def _get_retention_period(self, consent: Dict) -> int:
        """Determine retention period based on consent and purpose"""
        if consent.get(ConsentType.ANALYTICS, False):
            return 365  # 1 year for analytics
        return 90  # 90 days for essential support

PCI-DSS Compliance

Never let payment card data touch your AI agent without proper controls:

class PCICompliantAgent:
    """Agent that maintains PCI-DSS compliance"""

    # Card data should NEVER appear in these
    FORBIDDEN_CONTEXTS = ["llm_prompt", "log", "trace", "cache"]

    def __init__(self, agent, payment_tokenizer):
        self.agent = agent
        self.tokenizer = payment_tokenizer

    async def handle(
        self,
        query: str,
        context: dict
    ) -> str:
        # Step 1: Detect and tokenize any card data in query
        sanitized_query, tokens = await self._sanitize_card_data(query)

        # Step 2: Ensure context contains no card data
        safe_context = self._strip_card_data(context)

        # Step 3: Process with sanitized inputs
        response = await self.agent.handle(sanitized_query, safe_context)

        # Step 4: Ensure response contains no card data
        response = self._validate_no_card_data(response)

        return response

    async def _sanitize_card_data(self, text: str) -> tuple:
        """Replace card numbers with tokens"""
        card_pattern = r'\b(?:\d{4}[-\s]?){3}\d{4}\b'
        tokens = {}

        def tokenize(match):
            card_number = re.sub(r'[-\s]', '', match.group())
            token = f"CARD_TOKEN_{len(tokens)}"
            tokens[token] = {
                "last_four": card_number[-4:],
                "tokenized_at": datetime.now().isoformat()
            }
            # Never store full card number, even temporarily
            return f"[Card ending in {card_number[-4:]}]"

        sanitized = re.sub(card_pattern, tokenize, text)
        return sanitized, tokens

    def _strip_card_data(self, context: dict) -> dict:
        """Remove any card data from context"""
        safe = {}
        for key, value in context.items():
            if isinstance(value, str):
                # Check for card patterns
                if re.search(r'\b(?:\d{4}[-\s]?){3}\d{4}\b', value):
                    safe[key] = "[REDACTED - Card Data]"
                else:
                    safe[key] = value
            elif isinstance(value, dict):
                safe[key] = self._strip_card_data(value)
            elif isinstance(value, list):
                safe[key] = [self._strip_card_data(v) if isinstance(v, dict) else v for v in value]
            else:
                safe[key] = value
        return safe

    def _validate_no_card_data(self, response: str) -> str:
        """Ensure response doesn't leak card data"""
        card_pattern = r'\b(?:\d{4}[-\s]?){3}\d{4}\b'
        if re.search(card_pattern, response):
            # Log security event
            self.security_log.critical("Card data in agent response!")
            # Sanitize
            return re.sub(card_pattern, "[CARD NUMBER REDACTED]", response)
        return response


# Payment-related queries should use dedicated flow
class PaymentQueryHandler:
    """Handle payment queries without exposing card data to LLM"""

    async def handle_payment_query(
        self,
        query: str,
        user_id: str
    ) -> str:
        # Classify the payment query type
        query_type = self._classify_payment_query(query)

        if query_type == "update_card":
            # Never let AI handle card updates
            return (
                "To update your payment method, please visit your account settings "
                "or use our secure payment portal. For security, I cannot process "
                "card information directly."
            )

        elif query_type == "check_payment_status":
            # Fetch status from payment system (no card details)
            status = await self.payment_service.get_payment_status(user_id)
            return f"Your last payment of ${status.amount} was {status.status} on {status.date}."

        elif query_type == "refund":
            # Initiate refund flow (no card details needed)
            return await self._handle_refund_query(query, user_id)

Audit Trail Requirements

@dataclass
class AuditEvent:
    timestamp: datetime
    event_type: str
    user_id: str
    agent_id: str
    action: str
    data_accessed: List[str]
    legal_basis: str
    outcome: str
    ip_address: Optional[str]
    session_id: str

class ComplianceAuditLog:
    """Immutable audit log for compliance"""

    def __init__(self, storage):
        self.storage = storage

    async def log_data_access(
        self,
        user_id: str,
        agent_id: str,
        data_types: List[str],
        purpose: str,
        legal_basis: str
    ):
        """Log when personal data is accessed"""
        event = AuditEvent(
            timestamp=datetime.utcnow(),
            event_type="data_access",
            user_id=user_id,
            agent_id=agent_id,
            action="read",
            data_accessed=data_types,
            legal_basis=legal_basis,
            outcome="success",
            ip_address=self._get_client_ip(),
            session_id=self._get_session_id()
        )
        await self._store_immutable(event)

    async def log_data_processing(
        self,
        user_id: str,
        processing_type: str,
        data_involved: List[str],
        third_parties: List[str] = None
    ):
        """Log data processing activities"""
        event = AuditEvent(
            timestamp=datetime.utcnow(),
            event_type="data_processing",
            user_id=user_id,
            agent_id=self.agent_id,
            action=processing_type,
            data_accessed=data_involved,
            legal_basis="consent",  # or "legitimate_interest"
            outcome="success",
            ip_address=self._get_client_ip(),
            session_id=self._get_session_id()
        )

        # Log any third-party data sharing
        if third_parties:
            event.metadata = {"third_parties": third_parties}

        await self._store_immutable(event)

    async def _store_immutable(self, event: AuditEvent):
        """Store in immutable audit log"""
        # Use append-only storage
        await self.storage.append(
            collection="audit_log",
            document=asdict(event),
            options={"immutable": True}
        )

        # Also send to SIEM for monitoring
        await self.siem.send_event(event)

EU AI Act Considerations

The EU AI Act introduces specific requirements for AI systems:

class AIActCompliantAgent:
    """Agent compliant with EU AI Act requirements"""

    def __init__(self, agent):
        self.agent = agent
        self.disclosure_shown = set()

    async def handle(
        self,
        query: str,
        context: dict,
        session_id: str
    ) -> str:
        # Requirement: Transparency - disclose AI nature
        disclosure = ""
        if session_id not in self.disclosure_shown:
            disclosure = self._get_ai_disclosure()
            self.disclosure_shown.add(session_id)

        # Process query
        response = await self.agent.handle(query, context)

        # Add confidence indicator for transparency
        response_with_confidence = self._add_confidence_indicator(response)

        # Include human escalation option
        response_with_escalation = self._add_escalation_option(
            response_with_confidence,
            context
        )

        return disclosure + response_with_escalation

    def _get_ai_disclosure(self) -> str:
        """Mandatory AI disclosure"""
        return (
            "Hi! I'm an AI assistant here to help with your shopping questions. "
            "If you'd prefer to speak with a human agent, just let me know. "
            "\n\n"
        )

    def _add_confidence_indicator(self, response: str) -> str:
        """Add confidence context to response"""
        # This helps users understand AI limitations
        return response  # In practice, append confidence notes for uncertain responses

    def _add_escalation_option(self, response: str, context: dict) -> str:
        """Always provide human escalation option"""
        if context.get("high_value_transaction", False):
            return response + "\n\nWould you like me to connect you with a specialist for this?"
        return response

Compliance Checklist

Key Takeaways

Next up: Production AI Assessment - Test your knowledge with a comprehensive quiz!