argos-python 1.0.2

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, and lets you block threats instantly or asynchronously.

Features

• AI-Powered Detection: ML models analyze requests in real-time
• Zero Latency: Async mode queues events without blocking your app
• Sync Enforcement: Block malicious requests before they reach your logic
• Circuit Breaker: Built-in resilience with exponential backoff
• Framework Middleware: Drop-in support for FastAPI, Flask, Django, Starlette

Installation

pip install argos

# With framework support
pip install argos[fastapi]    # FastAPI + Starlette
pip install argos[flask]     # Flask
pip install argos[django]     # Django
pip install argos[all]      # Everything

Quick Start

from argos import create_client

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

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

if event.signal == "BLOCK":
    print("Threat detected!")

With Middleware

FastAPI

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

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

app.add_middleware(FastAPIMiddleware, client=client, mode="async")

Flask

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

app = Flask(__name__)
client = create_client(api_key="your-api-key")
FlaskMiddleware(app, client)

Django

Add to settings.py:

ARGOS_API_KEY = "your-api-key"

MIDDLEWARE = [
    ...
    'argos.middleware.django.DjangoMiddleware',
    ...
]

Configuration

from argos import create_client

client = create_client(
    api_key="your-api-key",
    timeout=30.0,
    max_retries=3,
    queue_size=1000,
    circuit_breaker_threshold=5,
    circuit_breaker_timeout=60.0,
)

Middleware Options

from argos.middleware.base import MiddlewareConfig

config = MiddlewareConfig(
    mode="async",              # "sync" to block, "async" to queue
    include_headers=True,       # include request headers in events
    exclude_paths=["/health"], # skip these paths
    identity_header="X-User-ID",
)

Error Handling

from argos import (
    argosError,
    argosAuthenticationError,
    argosRateLimitError,
    argosTimeoutError,
    argosCircuitOpenError,
)

try:
    event = client.ingest({...})
except argosAuthenticationError:
    print("Invalid API key")
except argosRateLimitError as e:
    print(f"Rate limited, retry in {e.retry_after}s")
except argosCircuitOpenError:
    print("argos is unavailable")

Async Usage

import asyncio
from argos import create_client

async def main():
    client = create_client(api_key="your-api-key")
    event = await client.aingest({"event_type": "login", "user_id": "user123"})
    print(event.signal)

asyncio.run(main())

High Throughput

Queue events during peak traffic, flush when convenient:

# Queue without blocking
for req in requests:
    client.queue_event({"event_type": "request", "path": req.url})

# Process queued events
events = client.flush_queue()

License

MIT