argos-python 1.2.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 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 Ingestion: Background worker automatically flushes event queues without blocking your application.
• Smart Caching: Built-in TTL caching for blocklist checks ensures sub-millisecond response times for repeated IPs.
• Framework Middleware: Drop-in middleware for FastAPI, Flask, Django, and Starlette.
• Resilient Design: Built-in Circuit Breaker and Retry logic with exponential backoff.
• Zero Dependencies: Core SDK uses only the Python standard library.

---

📦 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

import atexit
from argos import create_client

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

# Ensure clean shutdown and final flush
atexit.register(client.close)

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

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

---

🛠️ Configuration

Creating a Client

from argos import create_client

client = create_client(
    api_key="your-api-key",
    base_url="https://api.argossecops.com",
    
    # Reliability
    max_retries=3,                   # Max retry attempts
    circuit_breaker_threshold=5,     # Failures before opening circuit
    
    # Performance & Locality
    blocklist_cache_ttl=60.0,        # Cache blocklist results for 60s
    flush_interval=1.0,              # How often to flush background queue (seconds)
    batch_size=50,                   # Max events per batch flush
    
    # Behavior
    auto_block_on_block=False,       # Auto-block IP on BLOCK verdict
)

Configuration Options

Option	Default	Description
api_key	Required	Your Argos API key
base_url	https://...	Base URL of the Argos API
sync_mode	False	If True, disables background flushing (all ingestions block)
blocklist_cache_ttl	60.0	Time-to-live for local blocklist cache (seconds)
flush_interval	1.0	Frequency of background queue flushing (seconds)
batch_size	50	Maximum events per batch ingestion
max_retries	3	Maximum retry attempts for failed requests
circuit_breaker_threshold	5	Failures before temporarily suspending requests

---

🧩 Middleware Integration

FastAPI / Starlette

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")

app.add_middleware(FastAPIMiddleware, client=client, config=MiddlewareConfig(
    mode="async",                    # "sync" blocks requests, "async" queues
    include_body=True,               # Include request body in events
    exclude_paths=["/static"],       # Paths to skip
))

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, mode="sync") # Blocks bad requests immediately

---

🛡️ Blocklist Management

Argos provides a powerful programmatic blocklist that is synced to your local application cache for maximum performance.

# Check if an IP is blocked (uses local cache)
is_blocked, reason = client.is_blocked("192.168.1.1")

# Programmatically block a user
client.block_user("user123", environment_id="env_123", reason="Suspicious behavior")

# Unblock instantly
client.unblock(entry_id="...")

---

📖 Advanced Usage

Manual Queue Management

While the SDK flushes events in the background by default, you can still queue events manually for specific logic:

client.queue_event({"event_type": "nav"}, metadata={"path": "/home"})
# Background worker will pick this up automatically!

Clean Shutdown

Always call client.close() or register it with atexit to ensure all queued events are sent before your process exits.

import atexit
atexit.register(client.close)

---

📄 License

MIT License - see LICENSE for details.

🤝 Contributing

We welcome contributions! Please see our CONTRIBUTING.md for philosophy and coding guidelines.

• Support: [e-mail]
• Documentation: docs.argos.dev