SDK Quick Start¶
Learn the essential patterns for using the Aegis SDK effectively.
Basic Usage¶
Initialize the Client¶
from aegis_sdk import Aegis
# Basic usage (no license required)
aegis = Aegis()
# With license key for enterprise features
aegis = Aegis(license_key="aegis_lic_xxx")
# With custom policy configuration
aegis = Aegis(
policy_config={
"destinations": {
"AI_TOOL": {
"masked": ["EMAIL", "PHONE"],
"blocked": ["SSN"]
}
}
}
)
Process Text Content¶
result = aegis.process(
text="Customer email: [email protected], SSN: 123-45-6789",
destination="AI_TOOL"
)
print(f"Decision: {result.decision}")
print(f"Summary: {result.summary}")
print(f"Masked: {result.masked_content}")
Detect Only¶
# Just detect without processing decisions
detected = aegis.detect("Contact [email protected] or call 555-1234")
for item in detected:
print(f"{item.type}: {item.count} occurrence(s)")
Mask Only¶
# Just mask without policy evaluation
masked = aegis.mask("Contact [email protected] or call 555-1234")
print(masked) # Contact j***@example.com or call XXX-XXX-1234
Destinations¶
Aegis supports three destination types with different policy rules:
| Destination | Use Case | Example |
|---|---|---|
AI_TOOL |
AI assistants, chatbots, copilots | ChatGPT, Claude, GitHub Copilot |
VENDOR |
External service providers | Analytics, CRM, contractors |
CUSTOMER |
Customer-facing communications | Emails, reports, exports |
# Different rules apply based on destination
ai_result = aegis.process(text, destination="AI_TOOL")
vendor_result = aegis.process(text, destination="VENDOR")
customer_result = aegis.process(text, destination="CUSTOMER")
Handling Decisions¶
The ProcessingResult Object¶
result = aegis.process(text, destination="AI_TOOL")
# Decision: ALLOWED, BLOCKED, or ALLOWED_WITH_MASKING
result.decision
# Human-readable explanation
result.summary
# Safe content (masked if applicable)
result.masked_content
# List of detected sensitive items
result.detected
# Number of bytes processed
result.bytes_processed
# Suggested fix (if blocked)
result.suggested_fix
Decision Flow¶
from aegis_sdk import Decision
result = aegis.process(text, destination="AI_TOOL")
match result.decision:
case Decision.ALLOWED:
# Content is safe, use original
send_to_ai(text)
case Decision.ALLOWED_WITH_MASKING:
# Use the masked version
send_to_ai(result.masked_content)
case Decision.BLOCKED:
# Content violates policy
log_blocked_attempt(result.summary)
raise ValueError(result.summary)
Working with Detections¶
result = aegis.process(text, destination="AI_TOOL")
for detection in result.detected:
print(f"Type: {detection.type}") # e.g., "SSN", "EMAIL"
print(f"Count: {detection.count}") # Number of occurrences
print(f"Sample: {detection.sample}") # Partially masked example
Detection Types¶
| Type | Description | Pattern Example |
|---|---|---|
SSN |
Social Security Number | 123-45-6789 |
EMAIL |
Email address | [email protected] |
PHONE |
Phone number | (555) 123-4567 |
CREDIT_CARD |
Credit card number | 4111-1111-1111-1111 |
API_KEY |
API keys and tokens | sk-abc123... |
AWS_KEY |
AWS access keys | AKIA... |
IP_ADDRESS |
IP addresses | 192.168.1.1 |
Multi-Department Support¶
Organizations can have different policy groups for departments:
# Initialize with a specific policy group
aegis = Aegis(
license_key="aegis_lic_xxx",
policy_group="engineering"
)
# Switch policy groups dynamically
marketing = aegis.with_policy_group("marketing")
sales = aegis.with_policy_group("sales")
# Each has different rules
eng_result = aegis.process(text, destination="AI_TOOL")
mkt_result = marketing.process(text, destination="AI_TOOL")
Error Handling¶
from aegis_sdk import Aegis, AegisError, LicenseValidationError
try:
aegis = Aegis(license_key="aegis_lic_xxx")
result = aegis.process(text, destination="AI_TOOL")
except LicenseValidationError as e:
# Invalid or expired license
print(f"License error: {e}")
# Fall back to default policy
aegis = Aegis()
except AegisError as e:
# Other SDK errors
print(f"Aegis error: {e}")
Streaming File Processing¶
For large files, use streaming processors:
from aegis_sdk import StreamingProcessor, CSVStreamProcessor
from pathlib import Path
# Text files
processor = StreamingProcessor(chunk_size_mb=10)
result = processor.process_file(
input_path=Path("large_file.txt"),
output_path=Path("output.txt"),
destination="VENDOR"
)
print(f"Processed {result.mb_processed:.2f} MB")
print(f"Decision: {result.decision}")
# CSV files
csv_processor = CSVStreamProcessor()
result = csv_processor.process(
input_path=Path("data.csv"),
output_path=Path("data_masked.csv"),
destination="AI_TOOL"
)
Complete Example: AI Assistant¶
from aegis_sdk import Aegis, Decision
import openai
aegis = Aegis()
openai_client = openai.OpenAI()
def safe_chat(user_message: str) -> str:
"""Chat with AI, protected by Aegis."""
# Check user input before sending
result = aegis.process(
text=user_message,
destination="AI_TOOL"
)
if result.decision == Decision.BLOCKED:
return f"I can't process that request: {result.summary}"
# Use masked content if needed
safe_input = result.masked_content if result.masked_content else user_message
# Call OpenAI
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": safe_input}]
)
return response.choices[0].message.content
# Usage
response = safe_chat("Summarize this: Customer SSN is 123-45-6789")
print(response)
Using LLM Wrappers¶
For even simpler integration, use the built-in wrappers:
from aegis_sdk import AegisOpenAI
# Drop-in replacement for OpenAI client
client = AegisOpenAI(api_key="sk-...")
# PII is automatically masked before sending to OpenAI
response = client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "user",
"content": "Summarize: Customer email [email protected]"
}]
)
# OpenAI only sees: "Summarize: Customer email j***@example.com"
Next Steps¶
- API Reference - Detailed SDK documentation
- CLI Reference - Command-line usage
- LLM Integrations - OpenAI, Anthropic wrappers