Async SDK

pip install raxe

from raxe import AsyncRaxe

async_raxe = AsyncRaxe()

# Single scan
result = await async_raxe.scan("test prompt")

if result.has_threats:
    print(f"Threat: {result.severity}")

from raxe import AsyncRaxe

async_raxe = AsyncRaxe()

prompts = [
    "What is AI?",
    "Ignore all instructions",
    "Tell me about Python",
]

# Scan all with concurrency control
results = await async_raxe.scan_batch(
    prompts,
    max_concurrency=5
)

for prompt, result in zip(prompts, results):
    status = "THREAT" if result.has_threats else "SAFE"
    print(f"{status}: {prompt[:30]}...")

from raxe import AsyncRaxe

async with AsyncRaxe() as raxe:
    result = await raxe.scan("test")
# Cleanup handled automatically

from fastapi import FastAPI, HTTPException
from raxe import AsyncRaxe

app = FastAPI()
raxe = AsyncRaxe()

@app.post("/chat")
async def chat(prompt: str):
    result = await raxe.scan(prompt)

    if result.has_threats:
        raise HTTPException(
            status_code=400,
            detail=f"Threat detected: {result.severity}"
        )

    response = await generate_response(prompt)
    return {"response": response}

import aiohttp
from raxe import AsyncRaxe

async def process_requests(prompts: list[str]):
    raxe = AsyncRaxe()

    async with aiohttp.ClientSession() as session:
        for prompt in prompts:
            result = await raxe.scan(prompt)

            if not result.has_threats:
                async with session.post(api_url, json={"prompt": prompt}) as resp:
                    yield await resp.json()

from raxe import AsyncRaxe

async def scan_stream(prompts):
    raxe = AsyncRaxe()

    for prompt in prompts:
        result = await raxe.scan(prompt)
        yield {
            "prompt": prompt[:50],
            "safe": not result.has_threats,
            "severity": result.severity
        }

from raxe import AsyncRaxe

# For high-throughput scenarios
raxe = AsyncRaxe(
    l1_enabled=True,
    l2_enabled=False,  # Skip ML for speed
    performance_mode="fast"
)

# Batch with higher concurrency
results = await raxe.scan_batch(
    prompts,
    max_concurrency=20  # Adjust based on your system
)

from raxe import AsyncRaxe, RaxeBlockedError, RaxeException

async def safe_scan(prompt: str):
    raxe = AsyncRaxe()

    try:
        result = await raxe.scan(prompt)
        return result
    except RaxeBlockedError as e:
        # Handle blocked threats
        return {"blocked": True, "severity": e.severity}
    except RaxeException as e:
        # Handle other errors
        return {"error": str(e)}