argos-python 1.1.0

argos Security SDK for Python

Argos Security SDK

A production-grade Python SDK for real-time threat detection and fraud prevention. Argos monitors every request, scores risk using ML models, and lets you block threats instantly or asynchronously.

Features

• AI-Powered Detection: ML models analyze requests in real-time and return risk scores
• Zero Latency Option: Async mode queues events without blocking your application
• Sync Enforcement: Block malicious requests before they reach your business logic
• Framework Middleware: Drop-in middleware for FastAPI, Flask, Django, and Starlette
• Circuit Breaker: Built-in resilience with automatic recovery
• Retry Logic: Exponential backoff with jitter for failed requests
• Blocklist Management: Programmatic IP and user blocking
• High Throughput: Event queuing for peak traffic scenarios

Installation

pip install argos-python

# With framework support
pip install argos-python[fastapi]    # FastAPI + Starlette
pip install argos-python[flask]      # Flask
pip install argos-python[django]     # Django
pip install argos-python[all]        # All frameworks

Quick Start

from argos import create_client

# Initialize the client
client = create_client(
    api_key="your-api-key",
)

# Ingest an event for analysis
event = client.ingest({
    "event_type": "login",
    "user_id": "user123",
    "ip_address": "192.168.1.1",
    "status": "success"
})

# Check the verdict
if event.signal == "BLOCK":
    print("Threat detected! Block this request.")
else:
    print("Request allowed.")

Configuration

Creating a Client

from argos import create_client, ArgosClient, ArgosConfig

# Using the convenience function (recommended)
client = create_client(
    api_key="your-api-key",
    timeout=30.0,                    # Request timeout in seconds
    max_retries=3,                   # Maximum retry attempts
    retry_strategy="exponential",    # "exponential", "linear", or "none"
    retry_base_delay=0.5,            # Base delay between retries
    retry_max_delay=30.0,            # Maximum delay cap
    retry_jitter=True,               # Add randomness to delays
    circuit_breaker_threshold=5,     # Failures before opening circuit
    circuit_breaker_timeout=60.0,    # Seconds before attempting recovery
    queue_size=1000,                 # Maximum queued events
    auto_block_on_block=False,       # Auto-block IP on BLOCK verdict
    auto_block_ttl=3600.0,           # TTL for auto-blocked IPs (seconds)
    trusted_proxies=["10.0.0.0/8"],  # Trusted proxy CIDRs for IP extraction
)

# Or using the config class directly
config = ArgosConfig(
    api_key="your-api-key",
    # ... other options
)
client = ArgosClient(config)

Configuration Options

Option	Type	Default	Description
api_key	str	Required	Your Argos API key
timeout	float	30.0	Request timeout in seconds
max_retries	int	3	Maximum retry attempts
retry_strategy	str	"exponential"	Retry strategy: exponential, linear, or none
retry_base_delay	float	0.5	Base delay for retries
retry_max_delay	float	30.0	Maximum delay between retries
retry_jitter	bool	True	Add random jitter to delays
circuit_breaker_threshold	int	5	Failures before opening circuit
circuit_breaker_timeout	float	60.0	Seconds before attempting recovery
queue_size	int	1000	Maximum events in queue
auto_block_on_block	bool	False	Automatically block IP on BLOCK verdict
auto_block_ttl	float	3600.0	TTL for auto-blocked IPs
trusted_proxies	list[str]	None	Trusted proxy CIDRs

Core API

Ingesting Events

# Single event ingestion
event = client.ingest(
    payload={
        "event_type": "login",           # Required: event type identifier
        "user_id": "user123",            # Recommended
        "ip_address": "192.168.1.1",     # Recommended
        "status": "success",             # Optional
        # ... any custom fields
    },
    metadata={
        "externalID": "req_123",         # External identifier
        "sessionID": "sess_456",         # Session identifier
        "environment_id": "env_789",     # Environment ID
        "client_ip": "192.168.1.1",      # Client IP (for auto-block)
        "source": "api",                 # Event source
    }
)

# Access event properties
print(event.id)              # Event ID from Argos
print(event.signal)          # Verdict: "BLOCK", "ALLOW", or "PENDING"
print(event.anomaly_score)   # ML risk score (0.0 - 1.0)
print(event.payload)         # Original payload

Getting Decisions

# Get decision for a specific event
decision = client.get_decision(event_id)

print(decision.verdict)  # "allow", "block", or "pending"
print(decision.reason)   # Reason for the decision

# Convenience properties
if decision.is_blocked:
    print("Request blocked!")
elif decision.is_allowed:
    print("Request allowed")

Blocklist Management

# Check if an IP is blocked
is_blocked, reason = client.is_blocked("192.168.1.1")
if is_blocked:
    print(f"IP blocked: {reason}")

# Check if a user is blocked
is_blocked, reason = client.is_user_blocked("user123")
if is_blocked:
    print(f"User blocked: {reason}")

# Check both IP and user at once
status = client.check_access(ip="192.168.1.1", user_id="user123")
print(status.blocked)           # Overall blocked status
print(status.ip_blocked)        # IP blocked status
print(status.user_blocked)      # User blocked status

# Block an IP
entry = client.block_ip(
    ip="192.168.1.1",
    environment_id="env_123",
    reason="Malicious activity detected"
)
print(f"Blocked IP: {entry.id}")

# Block a user
entry = client.block_user(
    user_id="user123",
    environment_id="env_123",
    reason="Account compromise suspected"
)
print(f"Blocked user: {entry.id}")

# Unblock an entry
client.unblock(entry_id="entry_123")

# Get all blocklist entries
entries = client.get_blocklist()
for entry in entries:
    print(f"{entry.type}: {entry.value} - {entry.reason}")

Event Queuing

For high-throughput scenarios, queue events and flush them in batches:

# Queue events without blocking
for request in requests:
    client.queue_event(
        payload={
            "event_type": "request",
            "path": request.url,
            "user_id": request.user_id,
        },
        metadata={
            "externalID": request.id,
            "environment_id": "env_123"
        }
    )

# Process all queued events
events = client.flush_queue()
print(f"Processed {len(events)} events")

Async Usage

All client methods have async counterparts prefixed with a:

import asyncio
from argos import create_client

async def main():
    client = create_client(api_key="your-api-key")
    
    # Async ingest
    event = await client.aingest({
        "event_type": "login",
        "user_id": "user123"
    })
    print(event.signal)
    
    # Async batch
    events = await client.aingest_batch([
        {"event_type": "login", "user_id": "user1"},
        {"event_type": "login", "user_id": "user2"},
    ])
    
    # Async blocklist checks
    is_blocked, reason = await client.ais_blocked("192.168.1.1")
    
    # Async blocklist management
    await client.ablock_ip("192.168.1.1", "env_123", "reason")
    
    # Async queuing
    await client.aqueue_event({"event_type": "request"})
    events = await client.aflush_queue()

asyncio.run(main())

Framework Middleware

FastAPI

from fastapi import FastAPI
from argos import create_client
from argos.middleware.fastapi import FastAPIMiddleware
from argos.middleware.base import MiddlewareConfig

app = FastAPI()
client = create_client(api_key="your-api-key")

# Configure middleware
config = MiddlewareConfig(
    mode="async",                    # "sync" blocks requests, "async" queues
    include_headers=True,            # Include request headers in events
    include_body=False,              # Include request body (careful with sensitive data)
    exclude_paths=["/health", "/metrics"],  # Paths to skip
    identity_header="X-User-ID",     # Header to extract user ID
)

app.add_middleware(FastAPIMiddleware, client=client, config=config)

@app.get("/protected")
def protected_route():
    return {"message": "This route is protected by Argos"}

Flask

from flask import Flask
from argos import create_client
from argos.middleware.flask import FlaskMiddleware
from argos.middleware.base import MiddlewareConfig

app = Flask(__name__)

# Create client with auto-block enabled (optional)
client = create_client(
    api_key="your-api-key",
    auto_block_on_block=True,  # Auto-block when ML returns BLOCK
)

# Configure middleware with blocklist checking
config = MiddlewareConfig(
    mode="sync",              # 'sync' blocks bad requests, 'async' queues
    include_headers=True,
    include_body=True,
)
FlaskMiddleware(app, client, config)

@app.route("/protected")
def protected_route():
    return {"message": "This route is protected by Argos"}

Starlette

from starlette.applications import Starlette
from starlette.middleware import Middleware
from argos import create_client
from argos.middleware.starlette import StarletteMiddleware

app = Starlette(
    routes=[...],
    middleware=[
        Middleware(
            StarletteMiddleware,
            client=create_client(api_key="your-api-key"),
            mode="async"
        )
    ]
)

Django

Add to your settings.py:

# settings.py
ARGOS_API_KEY = "your-api-key"
ARGOS_BASE_URL = "https://api.your-argos-deployment.com"  # Optional

# Add middleware (order matters!)
MIDDLEWARE = [
    # ... other middleware
    'argos.middleware.django.DjangoMiddleware',
]

Or configure with custom settings:

# settings.py
ARGOS = {
    "api_key": "your-api-key",
    "base_url": "https://api.your-argos-deployment.com",
    "mode": "async",
    "exclude_paths": ["/health/", "/admin/"],
}

Middleware Configuration Options

Option	Type	Default	Description
mode	str	"async"	"sync" waits for decision, "async" queues events
include_headers	bool	True	Include request headers in events
include_body	bool	False	Include request body (use carefully)
exclude_paths	list[str]	["/health", "/metrics", "/_health"]	Paths to skip
identity_header	str	"X-User-ID"	Header to extract user ID
custom_payload_builder	Callable	None	Custom function to build payload

Error Handling

from argos import (
    ArgosError,
    ArgosAPIError,
    ArgosAuthenticationError,
    ArgosRateLimitError,
    ArgosTimeoutError,
    ArgosCircuitOpenError,
)

try:
    event = client.ingest({...})
except ArgosAuthenticationError as e:
    print(f"Invalid API key: {e.message}")
except ArgosRateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ArgosTimeoutError as e:
    print(f"Request timed out: {e.message}")
except ArgosCircuitOpenError as e:
    print(f"Argos unavailable (circuit open): {e.message}")
except ArgosAPIError as e:
    print(f"API error ({e.status_code}): {e.message}")
except ArgosError as e:
    print(f"Generic error: {e.message}")

Error Types

Error	Description
ArgosError	Base exception for all Argos errors
ArgosAPIError	Non-2xx API responses
ArgosAuthenticationError	Invalid API key (401)
ArgosRateLimitError	Rate limit exceeded (429), includes retry_after
ArgosTimeoutError	Request timeout
ArgosCircuitOpenError	Circuit breaker is open due to failures
ArgosValidationError	Invalid request data
ArgosQueueFullError	Event queue is full

Advanced Features

Circuit Breaker

The SDK includes a circuit breaker that automatically opens after repeated failures to prevent cascade failures:

config = ArgosConfig(
    api_key="your-api-key",
    circuit_breaker_threshold=5,    # Open after 5 consecutive failures
    circuit_breaker_timeout=60.0,   # Attempt recovery after 60 seconds
)

client = ArgosClient(config)

# When circuit is open, raises ArgosCircuitOpenError
try:
    event = client.ingest({...})
except ArgosCircuitOpenError:
    print("Argos is temporarily unavailable")

Auto-Block

Automatically block IPs that receive a BLOCK verdict:

config = ArgosConfig(
    api_key="your-api-key",
    auto_block_on_block=True,      # Enable auto-blocking
    auto_block_ttl=3600.0,         # Block for 1 hour
)

client = ArgosClient(config)

# When ingest returns BLOCK, IP is automatically blocked
event = client.ingest({
    "event_type": "login",
    "ip_address": "192.168.1.1",
}, metadata={
    "client_ip": "192.168.1.1",   # Required for auto-block
    "environment_id": "env_123",  # Required for auto-block
})

Trusted Proxies

Properly extract client IP when behind a proxy or load balancer:

config = ArgosConfig(
    api_key="your-api-key",
    trusted_proxies=[
        "10.0.0.0/8",      # Private network
        "172.16.0.0/12",   # Docker network
        "192.168.0.0/16",  # Local network
    ]
)

client = ArgosClient(config)
# X-Forwarded-For headers from trusted proxies will be used

Custom Payload Builder

For complete control over the event payload:

from argos.middleware.base import MiddlewareConfig

def custom_builder(request):
    return {
        "event_type": "api_request",
        "path": request.url.path,
        "method": request.method,
        "user_id": get_user_from_token(request),
        "ip_address": get_client_ip(request),
        "custom_field": "value",
    }

config = MiddlewareConfig(
    custom_payload_builder=custom_builder
)

app.add_middleware(FastAPIMiddleware, client=client, config=config)

Creating Custom Middleware

You can create custom middleware for any Python framework by subclassing BaseMiddleware:

from argos import create_client
from argos.middleware.base import BaseMiddleware, MiddlewareConfig
from argos.models import Decision

class MyFrameworkMiddleware(BaseMiddleware):
    """Custom middleware for any Python web framework."""

    def __init__(self, app, client, config=None):
        super().__init__(client, config)
        self.app = app  # Your framework's app

    def get_request_data(self, request):
        """Extract data from your framework's request object."""
        return {
            "method": request.method,
            "path": request.path,
            "headers": dict(request.headers),
        }

    def get_client_ip(self, request):
        """Extract client IP from request."""
        # Customize based on your framework
        return request.client_ip or request.headers.get("X-Forwarded-For", "unknown")

    def get_user_id(self, request):
        """Extract user ID from request."""
        # Customize based on your auth system
        return request.headers.get("X-User-ID")

    def process_request(self, request):
        """Process a request and return a Decision."""
        return self.config.mode == "sync"

    def __call__(self, environ, start_response):
        """WSGI interface example."""
        # Wrap your framework's request/response
        request = self._create_request_object(environ)
        
        if self.should_skip(request.path):
            return self.app(environ, start_response)
        
        # Check blocklist first
        blocked, reason = self.check_blocklist(request)
        if blocked:
            start_response("403 Forbidden", [("Content-Type", "application/json")])
            return [b'{"error": "blocked", "reason": "' + reason.encode() + b'"}']
        
        # Process based on mode
        if self.config.mode == "sync":
            decision = self.run_sync(request)
            if decision.verdict.lower() == "block":
                start_response("403 Forbidden", [("Content-Type", "application/json")])
                return [b'{"error": "blocked"}']
        else:
            self.run_async(request)
        
        # Add headers to response
        def custom_start_response(status, headers):
            headers.extend([
                ("X-Argos-Verdict", "allow"),
                ("X-Argos-Reason", "async_pending"),
            ])
            return start_response(status, headers)
        
        return self.app(environ, custom_start_response)

Minimal Custom Middleware Example

Here's a minimal example for a simple custom framework:

from argos import create_client
from argos.models import Decision

class SimpleArgosMiddleware:
    """Minimal middleware for custom frameworks."""

    def __init__(self, app, api_key, mode="async", exclude_paths=None):
        self.app = app
        self.client = create_client(api_key=api_key)
        self.mode = mode
        self.exclude_paths = exclude_paths or ["/health", "/metrics"]

    def should_skip(self, path):
        return any(path.startswith(p) for p in self.exclude_paths)

    def get_client_ip(self, environ):
        """Extract IP from WSGI environ."""
        forwarded = environ.get("HTTP_X_FORWARDED_FOR")
        if forwarded:
            return forwarded.split(",")[0].strip()
        return environ.get("REMOTE_ADDR", "unknown")

    def __call__(self, environ, start_response):
        path = environ.get("PATH_INFO", "")

        if self.should_skip(path):
            return self.app(environ, start_response)

        # Build payload
        payload = {
            "event_type": environ.get("REQUEST_METHOD", "unknown").lower(),
            "path": path,
            "ip_address": self.get_client_ip(environ),
            "user_agent": environ.get("HTTP_USER_AGENT", ""),
        }

        if self.mode == "sync":
            # Wait for decision
            try:
                event = self.client.ingest(payload)
                if event.signal == "BLOCK":
                    start_response("403 Forbidden", [("Content-Type", "application/json")])
                    return [b'{"error": "blocked"}']
            except Exception:
                pass  # Allow on errors
        else:
            # Queue and continue
            try:
                self.client.queue_event(payload)
            except Exception:
                pass

        # Add Argos headers
        def wrapped_start_response(status, headers):
            headers.extend([
                ("X-Argos-Verdict", "allow"),
                ("X-Argos-Reason", "async_pending" if self.mode == "async" else "ok"),
            ])
            return start_response(status, headers)

        return self.app(environ, wrapped_start_response)

Using with Custom Framework

# Example with a custom WSGI app
from my_framework import App

app = App()
wrapped_app = SimpleArgosMiddleware(
    app,
    api_key="your-api-key",
    mode="sync",  # or "async"
    exclude_paths=["/health", "/static"]
)

# Run with any WSGI server
if __name__ == "__main__":
    from wsgiref.simple_server import make_server
    server = make_server("localhost", 8000, wrapped_app)
    server.serve_forever()

Extending Existing Middleware

You can also extend the existing middleware classes to customize behavior:

from argos.middleware.fastapi import FastAPIMiddleware
from argos.middleware.base import MiddlewareConfig

class CustomFastAPIMiddleware(FastAPIMiddleware):
    """Extended FastAPI middleware with custom behavior."""

    async def process_sync(self, request):
        """Override to add custom logic before/after processing."""
        # Custom pre-processing
        custom_data = await self.extract_custom_data(request)

        # Call parent processing
        decision = await super().process_sync(request)

        # Custom post-processing
        if decision.verdict.lower() == "block":
            await self.log_block_event(request, decision)

        return decision

    async def extract_custom_data(self, request):
        """Extract custom data for your use case."""
        return {"custom_field": "value"}

    async def log_block_event(self, request, decision):
        """Log blocked requests to your own logging system."""
        print(f"Blocked: {request.url.path} - {decision.reason}")

# Use it exactly like the regular middleware
app.add_middleware(CustomFastAPIMiddleware, client=client, config=config)

Response Headers

When using middleware, responses include Argos headers:

X-Argos-Verdict: block
X-Argos-Reason: anomaly_score=0.85

Health Check

# Check if Argos API is healthy
health = client.health_check()
print(health)

TypeScript/JavaScript SDK

For frontend and Node.js integrations, see the official Argos JavaScript SDK:

npm install @argos.dev/js-sdk

License

MIT License - see LICENSE for details.

Support

• Documentation: https://docs.argos.dev
• Issues: https://github.com/argosdev/argos-python/issues
• Email: [e-mail]