HIPAA

1
class HIPAACompliantAgent(AgentBase):
2
    def __init__(self):
3
        super().__init__(name="hipaa-agent")
4
5
        # Use HIPAA-compliant voice provider
6
        self.add_language("English", "en-US", "rime.spore")
7
8
        # Use HIPAA-compliant LLM
9
        self.set_params({"ai_model": "gpt-oss-120b"})

1
from signalwire import AgentBase
2
from signalwire.core.function_result import FunctionResult
3
4
class HealthcareAgent(AgentBase):
5
    def __init__(self):
6
        super().__init__(name="healthcare-agent")
7
8
        # Mark PHI-accessing functions as secure
9
        # This adds cryptographic token validation
10
        self.define_tool(
11
            name="get_patient_info",
12
            description="Retrieve patient information by ID",
13
            parameters={
14
                "type": "object",
15
                "properties": {
16
                    "patient_id": {
17
                        "type": "string",
18
                        "description": "Patient identifier"
19
                    }
20
                },
21
                "required": ["patient_id"]
22
            },
23
            handler=self.get_patient_info,
24
            secure=True  # Enables token validation
25
        )
26
27
    def get_patient_info(self, args, raw_data):
28
        # Validate and sanitize input
29
        patient_id = args.get("patient_id", "").strip()
30
        if not self.validate_patient_id(patient_id):
31
            return FunctionResult("Invalid patient identifier.")
32
33
        # Fetch from secure backend
34
        # ... implementation
35
        return FunctionResult("Retrieved patient information.")

1
import logging
2
from datetime import datetime
3
from signalwire import AgentBase
4
from signalwire.core.function_result import FunctionResult
5
6
# Configure structured logging
7
logging.basicConfig(
8
    level=logging.INFO,
9
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
10
)
11
audit_logger = logging.getLogger('hipaa_audit')
12
13
class HIPAACompliantAgent(AgentBase):
14
    def __init__(self):
15
        super().__init__(name="hipaa-agent")
16
        self.add_language("English", "en-US", "rime.spore")
17
18
    def log_phi_access(self, action, user_id, resource_type, success, details=None):
19
        """Log PHI access for HIPAA audit trail."""
20
        audit_logger.info(
21
            "PHI_ACCESS",
22
            extra={
23
                "timestamp": datetime.utcnow().isoformat(),
24
                "action": action,
25
                "user_id": user_id,
26
                "resource_type": resource_type,
27
                "success": success,
28
                "details": self.sanitize_for_log(details)
29
            }
30
        )
31
32
    def sanitize_for_log(self, data):
33
        """Remove PHI from log entries."""
34
        if data is None:
35
            return None
36
37
        # Fields to redact
38
        sensitive_fields = [
39
            'ssn', 'social_security',
40
            'dob', 'date_of_birth',
41
            'address', 'phone',
42
            'medical_record', 'mrn',
43
            'diagnosis', 'prescription'
44
        ]
45
46
        sanitized = dict(data) if isinstance(data, dict) else {"value": "[REDACTED]"}
47
        for field in sensitive_fields:
48
            if field in sanitized:
49
                sanitized[field] = "[REDACTED]"
50
51
        return sanitized

1
class HealthcareAgent(AgentBase):
2
    def __init__(self):
3
        super().__init__(name="healthcare-agent")
4
        self.add_language("English", "en-US", "rime.spore")
5
6
        # Add recording disclosure to prompt
7
        self.prompt_add_section(
8
            "Recording Disclosure",
9
            body="""At the start of every call, you MUST inform the caller:
10
11
            "This call may be recorded for quality assurance and compliance purposes.
12
            Your continued participation indicates your consent to recording.
13
            If you do not wish to be recorded, please let me know now."
14
15
            Wait for acknowledgment before proceeding with any PHI discussion."""
16
        )

1
def collect_sensitive_info(self, args, raw_data):
2
    """Collect sensitive info with recording paused."""
3
    return (
4
        FunctionResult(
5
            "I'll need to collect some sensitive information. "
6
            "Recording will be paused during this process."
7
        )
8
        .stop_record_call(control_id="main_recording")
9
        # Recording is now paused
10
        # Collect sensitive data via subsequent function calls
11
    )
12
13
def resume_after_sensitive(self, args, raw_data):
14
    """Resume recording after sensitive collection."""
15
    return (
16
        FunctionResult("Thank you. Resuming our conversation.")
17
        .record_call(control_id="main_recording", stereo=True)
18
    )

1
class CompartmentalizedHealthcareAgent(AgentBase):
2
    def __init__(self):
3
        super().__init__(name="compartmentalized-healthcare")
4
        self.add_language("English", "en-US", "rime.spore")
5
6
    def verify_patient(self, args, raw_data):
7
        """Verify patient and store PHI in global_data, not in AI context."""
8
        dob = args.get("date_of_birth")
9
        ssn_last4 = args.get("ssn_last4")
10
11
        # Fetch full patient record from database
12
        patient = self.lookup_patient(dob, ssn_last4)
13
14
        if patient:
15
            # Store sensitive data in global_data - LLM cannot see this
16
            # Only your SWAIG functions can access it
17
            return (
18
                FunctionResult("I've verified your identity. How can I help?")
19
                .update_global_data({
20
                    "patient_id": patient["id"],
21
                    "mrn": patient["medical_record_number"],
22
                    "full_name": patient["full_name"],
23
                    "ssn": patient["ssn"],
24
                    "insurance_id": patient["insurance_id"],
25
                    "verified": True
26
                })
27
            )
28
29
        return FunctionResult("I couldn't verify your identity.")
30
31
    def schedule_appointment(self, args, raw_data):
32
        """Use stored PHI without exposing it to the LLM."""
33
        # Retrieve PHI from global_data - never sent to LLM
34
        global_data = raw_data.get("global_data", {})
35
36
        if not global_data.get("verified"):
37
            return FunctionResult("Please verify your identity first.")
38
39
        # Use the stored patient_id and MRN for scheduling
40
        patient_id = global_data.get("patient_id")
41
        mrn = global_data.get("mrn")
42
43
        # Schedule using real identifiers, but LLM only knows
44
        # "the verified patient" - not the actual PHI
45
        appointment = self.create_appointment(patient_id, args.get("date"))
46
47
        # Return confirmation without exposing PHI to AI
48
        return FunctionResult(
49
            f"Your appointment is confirmed for {args.get('date')}. "
50
            "You'll receive a confirmation at the contact information on file."
51
        )

1
from signalwire import AgentBase, FunctionResult
2
3
agent = AgentBase(name="healthcare-agent", route="/agent")
4
5
@agent.tool(name="collect_dob", description="Collect date of birth via DTMF, bypassing the LLM")
6
def collect_dob(args, raw_data=None):
7
    # Run the SWML `prompt` verb to collect digits. The input lands in ${prompt_value}
8
    # and never enters the model's context, the transcript, or (with recording paused)
9
    # the recording.
10
    return (
11
        FunctionResult()
12
        .execute_swml({
13
            "version": "1.0.0",
14
            "sections": {
15
                "main": [
16
                    {"stop_record_call": {"control_id": "main_recording"}},
17
                    {"prompt": {
18
                        "play": "say:Please enter your 8-digit date of birth. For example, 0 3 1 5 1 9 8 5 for March 15, 1985.",
19
                        "max_digits": 8,
20
                        "digit_timeout": 15,
21
                    }},
22
                    {"record_call": {"control_id": "main_recording", "stereo": True}},
23
                ]
24
            },
25
        })
26
        .set_response(
27
            "Thank the caller for providing their information and continue verifying "
28
            "their identity. Never repeat sensitive values back to the caller."
29
        )
30
    )
31
32
agent.run()

1
class SecureHealthcareAgent(AgentBase):
2
    def __init__(self):
3
        super().__init__(name="secure-healthcare")
4
        self.add_language("English", "en-US", "rime.spore")
5
6
        self.prompt_add_section(
7
            "PHI Protection Rules",
8
            body="""CRITICAL: Follow these rules for Protected Health Information:
9
10
            1. NEVER repeat back sensitive information like:
11
               - Social Security Numbers
12
               - Full date of birth
13
               - Medical record numbers
14
               - Detailed diagnosis information
15
16
            2. When confirming patient identity, use partial information:
17
               - "I see your date of birth ends in [last 2 digits of year]"
18
               - "Your record shows an address on [street name only]"
19
20
            3. NEVER include PHI in error messages or clarifications
21
22
            4. If asked to repeat sensitive info, say:
23
               "For security purposes, I cannot repeat that information."
24
25
            5. Verify caller identity before ANY PHI access using
26
               established verification procedures."""
27
        )
28
29
        self.prompt_add_section(
30
            "Identity Verification",
31
            body="""Before accessing ANY patient information:
32
33
            1. Collect patient identifier (MRN or account number)
34
            2. Verify with TWO of the following:
35
               - Date of birth
36
               - Last 4 of SSN
37
               - Address on file
38
               - Phone number on file
39
40
            Only proceed after successful verification."""
41
        )

1
def check_appointment(self, args, raw_data):
2
    """Check appointment - returns minimal PHI."""
3
    patient_id = args.get("patient_id")
4
5
    # Verify caller is authorized
6
    if not self.verify_caller_authorization(raw_data, patient_id):
7
        self.log_phi_access("appointment_check", patient_id, "appointment", False)
8
        return FunctionResult(
9
            "I was unable to verify your identity. "
10
            "Please contact us directly for assistance."
11
        )
12
13
    # Log the access
14
    self.log_phi_access("appointment_check", patient_id, "appointment", True)
15
16
    # Return minimal information
17
    appointment = self.get_next_appointment(patient_id)
18
    if appointment:
19
        # Don't include provider names or detailed location
20
        return FunctionResult(
21
            f"Your next appointment is on {appointment['date']} "
22
            f"at {appointment['time']}. "
23
            f"Please arrive 15 minutes early."
24
        )
25
26
    return FunctionResult("You have no upcoming appointments scheduled.")

1
import re
2
3
def validate_patient_id(self, patient_id):
4
    """Validate patient ID format."""
5
    # Only allow alphanumeric, specific length
6
    if not patient_id:
7
        return False
8
    if not re.match(r'^[A-Z0-9]{8,12}$', patient_id):
9
        return False
10
    return True
11
12
def validate_date_of_birth(self, dob):
13
    """Validate DOB format."""
14
    try:
15
        # Expected format: MM/DD/YYYY
16
        if not re.match(r'^\d{2}/\d{2}/\d{4}$', dob):
17
            return False
18
        # Additional validation...
19
        return True
20
    except Exception:
21
        return False

1
# Example: Role-based function access
2
class RoleBasedHealthcareAgent(AgentBase):
3
4
    # Define role permissions
5
    ROLE_PERMISSIONS = {
6
        "patient": ["check_appointment", "request_prescription_refill"],
7
        "staff": ["check_appointment", "schedule_appointment", "view_patient_summary"],
8
        "provider": ["check_appointment", "schedule_appointment", "view_patient_summary",
9
                     "view_medical_record", "add_clinical_note"]
10
    }
11
12
    def __init__(self):
13
        super().__init__(name="role-based-healthcare")
14
        self.caller_role = None
15
16
    def verify_permission(self, function_name):
17
        """Check if current caller role has permission."""
18
        if not self.caller_role:
19
            return False
20
        allowed = self.ROLE_PERMISSIONS.get(self.caller_role, [])
21
        return function_name in allowed

1
import os
2
3
class HealthcareAgent(AgentBase):
4
    def __init__(self):
5
        super().__init__(name="healthcare-agent")
6
        self.emergency_mode = os.environ.get("AGENT_EMERGENCY_MODE", "").lower() == "true"
7
8
    def access_phi(self, args, raw_data):
9
        """PHI access with emergency mode check."""
10
        if self.emergency_mode:
11
            audit_logger.warning("PHI_ACCESS_BLOCKED_EMERGENCY_MODE")
12
            return FunctionResult(
13
                "This service is temporarily unavailable. "
14
                "Please call our main line for assistance."
15
            )
16
17
        # Normal PHI access logic...

1
#!/usr/bin/env python3
2
"""HIPAA-compliant healthcare appointment agent."""
3
4
import os
5
import logging
6
from datetime import datetime
7
from signalwire import AgentBase
8
from signalwire.core.function_result import FunctionResult
9
10
# Audit logging setup
11
audit_logger = logging.getLogger('hipaa_audit')
12
audit_logger.setLevel(logging.INFO)
13
14
class HIPAAHealthcareAgent(AgentBase):
15
    """Healthcare agent with HIPAA compliance controls."""
16
17
    def __init__(self):
18
        super().__init__(name="hipaa-healthcare-agent")
19
20
        # Use HIPAA-compliant voice and LLM providers
21
        self.add_language("English", "en-US", "rime.spore")
22
        self.set_params({"ai_model": "gpt-oss-120b"})
23
24
        # Check required security configuration
25
        if not os.environ.get("SWML_BASIC_AUTH_PASSWORD"):
26
            raise ValueError("SWML_BASIC_AUTH_PASSWORD required for HIPAA compliance")
27
28
        self.emergency_mode = os.environ.get("AGENT_EMERGENCY_MODE", "").lower() == "true"
29
        self.verified_caller = False
30
31
        self._configure_prompts()
32
        self._setup_tools()
33
34
    def _configure_prompts(self):
35
        """Configure HIPAA-compliant prompts."""
36
        self.prompt_add_section(
37
            "Role",
38
            body="You are a healthcare appointment assistant for Example Medical Center."
39
        )
40
41
        self.prompt_add_section(
42
            "Recording Disclosure",
43
            body="""At the START of every call, say:
44
45
            "Thank you for calling Example Medical Center. This call may be recorded
46
            for quality and compliance purposes. How may I help you today?"
47
            """
48
        )
49
50
        self.prompt_add_section(
51
            "PHI Protection",
52
            body="""CRITICAL RULES:
53
            1. Verify caller identity before accessing ANY patient information
54
            2. Never repeat back full SSN, DOB, or medical record numbers
55
            3. Use partial confirmation only ("ending in...", "on file as...")
56
            4. Never include PHI in error messages
57
            5. If unsure about authorization, do not provide information
58
            """
59
        )
60
61
        self.prompt_add_section(
62
            "Verification Process",
63
            body="""Before accessing patient information:
64
            1. Ask for patient's date of birth
65
            2. Ask for last 4 digits of SSN OR phone number on file
66
            3. Use verify_patient function to confirm
67
            4. Only proceed if verification succeeds
68
            """
69
        )
70
71
    def _setup_tools(self):
72
        """Configure secure tools."""
73
        self.define_tool(
74
            name="verify_patient",
75
            description="Verify patient identity before accessing records",
76
            parameters={
77
                "type": "object",
78
                "properties": {
79
                    "date_of_birth": {
80
                        "type": "string",
81
                        "description": "Patient date of birth (MM/DD/YYYY)"
82
                    },
83
                    "verification_value": {
84
                        "type": "string",
85
                        "description": "Last 4 of SSN or phone number"
86
                    }
87
                },
88
                "required": ["date_of_birth", "verification_value"]
89
            },
90
            handler=self.verify_patient,
91
            secure=True
92
        )
93
94
        self.define_tool(
95
            name="check_appointments",
96
            description="Check patient appointments (requires prior verification)",
97
            parameters={
98
                "type": "object",
99
                "properties": {}
100
            },
101
            handler=self.check_appointments,
102
            secure=True
103
        )
104
105
    def _log_audit(self, action, success, details=None):
106
        """Log HIPAA audit event."""
107
        audit_logger.info(
108
            f"HIPAA_AUDIT|{datetime.utcnow().isoformat()}|{action}|"
109
            f"{'SUCCESS' if success else 'FAILURE'}|{details or ''}"
110
        )
111
112
    def verify_patient(self, args, raw_data):
113
        """Verify patient identity."""
114
        if self.emergency_mode:
115
            self._log_audit("verify_patient", False, "emergency_mode")
116
            return FunctionResult(
117
                "Our system is temporarily unavailable. Please call back later."
118
            )
119
120
        dob = args.get("date_of_birth", "")
121
        verification = args.get("verification_value", "")
122
123
        # In production, verify against your patient database
124
        # This is a placeholder for the actual verification logic
125
        verified = self._verify_against_database(dob, verification)
126
127
        if verified:
128
            self.verified_caller = True
129
            self._log_audit("verify_patient", True, "identity_confirmed")
130
            return FunctionResult(
131
                "Thank you, I've verified your identity. How can I help you today?"
132
            )
133
        else:
134
            self._log_audit("verify_patient", False, "verification_failed")
135
            return FunctionResult(
136
                "I wasn't able to verify your identity with the information provided. "
137
                "Please contact our office directly for assistance."
138
            )
139
140
    def check_appointments(self, args, raw_data):
141
        """Check appointments - requires prior verification."""
142
        if not self.verified_caller:
143
            self._log_audit("check_appointments", False, "not_verified")
144
            return FunctionResult(
145
                "I need to verify your identity first. "
146
                "Can you provide your date of birth?"
147
            )
148
149
        if self.emergency_mode:
150
            self._log_audit("check_appointments", False, "emergency_mode")
151
            return FunctionResult(
152
                "Our scheduling system is temporarily unavailable."
153
            )
154
155
        # Fetch appointments from your system
156
        # Return minimal necessary information
157
        self._log_audit("check_appointments", True, "appointments_retrieved")
158
        return FunctionResult(
159
            "Your next appointment is scheduled for Monday at 2:30 PM "
160
            "with the medical team. Please arrive 15 minutes early."
161
        )
162
163
    def _verify_against_database(self, dob, verification):
164
        """Verify patient against database. Implement your logic here."""
165
        # Placeholder - implement actual database verification
166
        return True
167
168
if __name__ == "__main__":
169
    agent = HIPAAHealthcareAgent()
170
    agent.run()