SDK API Reference¶

Complete API reference for the Aegis Python SDK.

Core Classes¶

Aegis¶

The main client class for PII detection and masking.

from aegis_sdk import Aegis

# Create client instance
aegis = Aegis()

# Process content
result = aegis.process(text="Contact [email protected]", destination="AI_TOOL")

Constructor¶

def __init__(
    self,
    license_key: str = None,
    include_samples: bool = True,
    policy_config: dict = None,
    policy_group: str = None,
    auto_fetch_policy: bool = True,
    api_endpoint: str = None,
) -> None:

Parameter	Type	Default	Description
`license_key`	`str`	`None`	License key for enterprise features
`include_samples`	`bool`	`True`	Include masked samples in detection results
`policy_config`	`dict`	`None`	Custom policy configuration
`policy_group`	`str`	`None`	Policy group/department to use
`auto_fetch_policy`	`bool`	`True`	Auto-fetch policy from cloud if license provided
`api_endpoint`	`str`	`None`	Custom API endpoint for license validation

Methods¶

process()¶

Process text with full policy decision logic.

def process(
    self,
    text: str,
    destination: str = "AI_TOOL",
) -> ProcessingResult:

Parameter	Type	Required	Description
`text`	`str`	Yes	Text content to process
`destination`	`str`	No	Target: AI_TOOL, VENDOR, or CUSTOMER

Returns: ProcessingResult

result = aegis.process(
    text="Call me at 555-123-4567",
    destination="AI_TOOL"
)
print(result.decision)        # Decision.ALLOWED_WITH_MASKING
print(result.masked_content)  # Call me at XXX-XXX-4567

detect()¶

Detect sensitive data without policy evaluation.

def detect(self, text: str) -> list[DetectedItem]:

Returns: List of DetectedItem

detected = aegis.detect("Email: [email protected], Phone: 555-1234")
for item in detected:
    print(f"{item.type}: {item.count}")

mask()¶

Mask sensitive data without policy evaluation.

def mask(self, text: str) -> str:

Returns: Masked text string

masked = aegis.mask("My SSN is 123-45-6789")
print(masked)  # My SSN is XXX-XX-6789

with_policy_group()¶

Create instance with different policy group.

def with_policy_group(self, group: str) -> "Aegis":

marketing = aegis.with_policy_group("marketing")
engineering = aegis.with_policy_group("engineering")

ProcessingResult¶

Result object returned from process().

@dataclass
class ProcessingResult:
    decision: Decision
    """Policy decision: ALLOWED, BLOCKED, or ALLOWED_WITH_MASKING"""

    summary: str
    """Human-readable explanation"""

    detected: List[DetectedItem]
    """List of detected sensitive items"""

    masked_content: Optional[str]
    """Masked content (if applicable)"""

    suggested_fix: Optional[str]
    """Suggestion for fixing blocked content"""

    bytes_processed: int
    """Number of bytes processed"""

    @property
    def is_blocked(self) -> bool:
        """True if content was blocked"""

DetectedItem¶

Represents detected sensitive data.

@dataclass
class DetectedItem:
    type: str
    """Detection type (e.g., 'SSN', 'EMAIL')"""

    count: int
    """Number of occurrences found"""

    sample: Optional[str] = None
    """Partially masked sample"""

Decision¶

Policy decision enumeration.

class Decision(str, Enum):
    ALLOWED = "ALLOWED"
    """Content is safe to send as-is"""

    BLOCKED = "BLOCKED"
    """Content violates policy"""

    ALLOWED_WITH_MASKING = "ALLOWED_WITH_MASKING"
    """Content allowed after masking PII"""

Destination¶

Valid destination types.

class Destination(str, Enum):
    AI_TOOL = "AI_TOOL"
    """AI assistants, chatbots, LLMs"""

    VENDOR = "VENDOR"
    """External vendors and service providers"""

    CUSTOMER = "CUSTOMER"
    """Customer-facing communications"""

Streaming Processors¶

StreamingProcessor¶

Process large text files in chunks.

from aegis_sdk import StreamingProcessor
from pathlib import Path

processor = StreamingProcessor(chunk_size_mb=10)

result = processor.process_file(
    input_path=Path("large_file.txt"),
    output_path=Path("output.txt"),
    destination="VENDOR",
    on_progress=lambda bytes, total, chunks: print(f"Processed: {bytes}"),
    stop_on_block=False
)

print(f"Processed {result.mb_processed:.2f} MB in {result.chunks_processed} chunks")

CSVStreamProcessor¶

Process CSV files with row-level streaming.

from aegis_sdk import CSVStreamProcessor
from pathlib import Path

processor = CSVStreamProcessor()

result = processor.process(
    input_path=Path("data.csv"),
    output_path=Path("data_masked.csv"),
    destination="AI_TOOL",
    has_header=True
)

LLM Wrappers¶

AegisOpenAI¶

Drop-in replacement for OpenAI client with automatic PII masking.

from aegis_sdk import AegisOpenAI

# Requires: pip install aegis-sdk[openai]
client = AegisOpenAI(api_key="sk-...")

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{
        "role": "user",
        "content": "Customer email: [email protected]"
    }]
)
# PII automatically masked before sending to OpenAI

AegisAnthropic¶

Drop-in replacement for Anthropic client.

from aegis_sdk import AegisAnthropic

# Requires: pip install aegis-sdk[anthropic]
client = AegisAnthropic(api_key="...")

message = client.messages.create(
    model="claude-3-opus-20240229",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": "Customer SSN: 123-45-6789"
    }]
)
# PII automatically masked before sending to Anthropic

AegisLLMGateway¶

Generic gateway for any LLM provider.

from aegis_sdk import AegisLLMGateway

gateway = AegisLLMGateway(destination="AI_TOOL")

# Mask single prompt
masked = gateway.mask_prompt("Email: [email protected]")

# Mask message list
messages = [
    {"role": "user", "content": "My SSN is 123-45-6789"}
]
masked_messages = gateway.mask_messages(messages)

License Management¶

LicenseManager¶

Online license validation with caching.

from aegis_sdk import LicenseManager

manager = LicenseManager(
    license_key="aegis_lic_xxx",
    policy_group="engineering"
)

info = manager.validate()
print(f"Organization: {info.org_id}")
print(f"Expires: {info.expires}")
print(f"Policy Groups: {info.policy_groups}")

OfflineLicenseManager¶

Air-gapped license validation.

from aegis_sdk import OfflineLicenseManager
from pathlib import Path

manager = OfflineLicenseManager(
    license_file=Path("aegis_license.json")
)

info = manager.validate()

Metrics & Audit¶

MetricsReporter¶

Async metrics reporting to Aegis Cloud.

from aegis_sdk import MetricsReporter

reporter = MetricsReporter(
    license_key="aegis_lic_xxx",
    batch_size=100,
    flush_interval=300
)

# Record processing events (non-blocking)
reporter.record(
    decision="ALLOWED_WITH_MASKING",
    bytes_processed=1024,
    detected_types=["EMAIL", "PHONE"],
    detected_counts={"EMAIL": 2, "PHONE": 1}
)

# Flush on shutdown
reporter.stop()

LocalMetricsCollector¶

Local-only metrics for air-gapped environments.

from aegis_sdk import LocalMetricsCollector
from pathlib import Path

collector = LocalMetricsCollector(
    output_path=Path("/var/log/aegis/metrics.jsonl")
)

collector.record(
    decision="BLOCKED",
    bytes_processed=512,
    detected_types=["SSN"],
    detected_counts={"SSN": 1}
)

summary = collector.get_summary()

AuditLog¶

Audit logging with hash chain verification.

from aegis_sdk import AuditLog
from pathlib import Path

audit = AuditLog(
    log_path=Path("/var/log/aegis/audit.jsonl"),
    verify_chain=True
)

audit.log_processing(
    decision="ALLOWED_WITH_MASKING",
    destination="AI_TOOL",
    detected=[{"type": "EMAIL", "count": 2}],
    bytes_processed=1024,
    user_id="user123"
)

# Verify integrity
is_valid, broken_at = audit.verify_chain()

GDPRAuditLog¶

GDPR-compliant audit log (no sample storage).

from aegis_sdk import GDPRAuditLog

audit = GDPRAuditLog(log_path=Path("gdpr_audit.jsonl"))

# Samples are automatically stripped
audit.log_processing(
    decision="ALLOWED",
    destination="VENDOR",
    detected=[{"type": "EMAIL", "count": 1, "sample": "j***@example.com"}],
    bytes_processed=256,
    user_id="eu_user_123"
)

# Data subject access request
report = audit.get_data_subject_report(user_id="eu_user_123")

Exceptions¶

AegisError¶

Base exception for all SDK errors.

from aegis_sdk import AegisError

try:
    result = aegis.process(text, destination="AI_TOOL")
except AegisError as e:
    print(f"SDK error: {e}")

LicenseValidationError¶

Invalid or expired license.

from aegis_sdk import LicenseValidationError

try:
    aegis = Aegis(license_key="invalid")
except LicenseValidationError as e:
    print(f"License error: {e}")

AegisBlockedError¶

Content blocked by policy (for strict mode).

from aegis_sdk import AegisBlockedError

try:
    # In strict mode
    result = gateway.process_strict(text)
except AegisBlockedError as e:
    print(f"Blocked: {e.result.summary}")

Detection Types¶

Type	Description	Example Pattern
`SSN`	Social Security Number	`123-45-6789`
`EMAIL`	Email address	`[email protected]`
`PHONE`	Phone number	`(555) 123-4567`
`CREDIT_CARD`	Credit card (Luhn validated)	`4111-1111-1111-1111`
`AWS_KEY`	AWS access key	`AKIA...`
`API_KEY`	Generic API keys	`sk-...`
`IP_ADDRESS`	IPv4/IPv6 address	`192.168.1.1`

Type Hints¶

The SDK is fully typed:

from aegis_sdk import (
    # Main class
    Aegis,
    # Types
    Decision,
    Destination,
    DetectedItem,
    ProcessingResult,
    BatchProcessingResult,
    # Errors
    AegisError,
    AegisBlockedError,
    LicenseValidationError,
    # Processors
    StreamingProcessor,
    CSVStreamProcessor,
    # LLM
    AegisOpenAI,
    AegisAnthropic,
    AegisLLMGateway,
    # Enterprise
    LicenseManager,
    MetricsReporter,
    AuditLog,
    GDPRAuditLog,
)