laroguard 2.0.0

Official Python SDK for the LaroGuard AI security gateway

LaroGuard Python SDK

Official Python SDK for the LaroGuard AI security gateway. Route your LLM traffic through LaroGuard to apply input/output security rules, PII masking, prompt-injection detection, and rate limiting — without changing your application logic.

---

Requirements

	Minimum
Python	3.9
httpx	0.27

---

Installation

pip install laroguard

---

Quick start

from laroguard import LaroGuard

lg = LaroGuard(
    api_key="your-project-api-key",   # → X-API-Key header
    project_id="proj_xxx",            # → X-Project-ID header + body field
    integration_id="openai_primary",  # → integration_id body field
    base_url="https://gateway.example.com",
)

# Non-streaming — returns a complete ProcessResponse
resp = lg.process(messages=[{"role": "user", "content": "Hello"}])
print(resp.content)          # cleaned / masked LLM response text
print(resp.decision)         # "ALLOW" | "MASK" | "REDACT" | "BLOCK"
print(resp.triggered_rules)  # e.g. ["pii_email", "prompt_injection"]

# Streaming — yields StreamEvent objects
for event in lg.stream(messages=[{"role": "user", "content": "Tell me a story"}]):
    if event.type == "chunk":
        print(event.content, end="", flush=True)
    elif event.type == "blocked":
        print("\n[BLOCKED BY SECURITY RULE]")
    elif event.type == "done":
        print()  # newline after stream ends

---

Constructor parameters

Parameter	Type	Required	Default	Description
api_key	str	✓	—	Project API key. Sent as the X-API-Key request header.
project_id	str	✓	—	Project identifier. Sent as the X-Project-ID header and project_id body field.
integration_id	str	✓	—	Default integration to invoke. Can be overridden per call. See Finding your integration_id.
base_url	str		http://localhost:8000	Root URL of the LaroGuard gateway.
timeout	float		120.0	Per-request timeout in seconds.

---

process() — non-streaming

resp = lg.process(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of France?"},
    ],
    session_id="sess_abc",          # optional: link turns into a conversation
    integration_id="openai_gpt4o",  # optional: override the instance default
    skip_guard=False,               # optional: bypass security rules (use with care)
)

Response fields

Field	Type	Description
request_id	str	Unique request identifier assigned by the gateway.
trace_id	str	Internal trace identifier for debugging.
decision	str	Gateway decision: "ALLOW", "MASK", "REDACT", or "BLOCK".
content	str	The LLM response after any masking or redaction has been applied. Empty when decision == "BLOCK".
blocked	bool	True when the request was blocked before reaching the LLM.
masked	bool	True when at least one rule masked or redacted part of the output.
triggered_rules	list[str]	Names of the security rules that fired, e.g. ["pii_email", "api_key_leak"].

---

stream() — server-sent events

for event in lg.stream(
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    session_id="sess_xyz",
):
    if event.type == "chunk":
        print(event.content, end="", flush=True)
    elif event.type == "blocked":
        print("\n[Stream terminated by security rule]")
    elif event.type == "done":
        print("\n[Stream complete]")

Stream event types

event.type	Class	Attributes	When emitted
"chunk"	StreamChunkEvent	content: str	For every safe text token from the LLM.
"blocked"	StreamBlockedEvent	(none)	When the gateway terminates the stream due to a triggered rule. Iteration stops.
"done"	StreamDoneEvent	(none)	After the stream completes normally. Always the last event.

---

Async usage

import asyncio
from laroguard import AsyncLaroGuard

async def main() -> None:
    async with AsyncLaroGuard(
        api_key="your-project-api-key",
        project_id="proj_xxx",
        integration_id="openai_primary",
        base_url="https://gateway.example.com",
    ) as lg:
        # Non-streaming
        resp = await lg.process(
            messages=[{"role": "user", "content": "Hello"}],
        )
        print(resp.content)

        # Streaming
        async for event in lg.stream(
            messages=[{"role": "user", "content": "Tell me a story"}],
        ):
            if event.type == "chunk":
                print(event.content, end="", flush=True)
            elif event.type == "blocked":
                print("\n[BLOCKED]")

asyncio.run(main())

AsyncLaroGuard accepts the same constructor parameters as LaroGuard and exposes identical process() and stream() methods, both async.

---

Error handling

All SDK exceptions inherit from LaroGuardError, so you can use a single broad handler or catch specific subclasses for fine-grained control.

from laroguard import (
    LaroGuardError,      # base — catch-all
    SecurityBlockError,  # 403 — input or output blocked by a rule
    AuthenticationError, # 401 — invalid or missing API key
    NotFoundError,       # 404 — integration_id not found / project not found
    RateLimitError,      # 429 — quota exceeded
    APIError,            # other 4xx / 5xx HTTP errors
    ConnectionError,     # network failure (timeout, DNS, refused)
    StreamBlockedError,  # mid-stream block sentinel (raised inside stream loop)
)

try:
    resp = lg.process(messages=[{"role": "user", "content": "Hello"}])
except SecurityBlockError as e:
    print(f"Blocked: {e.message}")
except AuthenticationError:
    print("Check your API key")
except RateLimitError:
    print("Back off and retry")
except NotFoundError:
    print("Verify your integration_id in the dashboard")
except ConnectionError:
    print("Cannot reach the gateway — check your base_url")
except APIError as e:
    print(f"Gateway error {e.status_code}: {e.message}")

Exception hierarchy

LaroGuardError
├── APIError               — 4xx / 5xx HTTP responses
│   ├── SecurityBlockError — 403
│   ├── AuthenticationError— 401
│   ├── NotFoundError      — 404
│   └── RateLimitError     — 429
├── StreamBlockedError     — mid-stream block sentinel
└── ConnectionError        — network-level failure

---

Context managers

Both clients support the context-manager protocol, which ensures the underlying HTTP connection pool is closed cleanly on exit.

# Synchronous
with LaroGuard(api_key=..., project_id=..., integration_id=...) as lg:
    resp = lg.process(messages=[...])

# Asynchronous
async with AsyncLaroGuard(api_key=..., project_id=..., integration_id=...) as lg:
    resp = await lg.process(messages=[...])

---

Finding your integration_id

1. Open the LaroGuard management dashboard.
2. Navigate to Project → Integrations.
3. Copy the identifier shown next to your LLM integration (e.g. openai_primary).

The integration_id tells the gateway which LLM provider credential to use when forwarding your request. You can configure multiple integrations per project and switch between them per call by passing integration_id= to process() or stream().

---

License

MIT — see LICENSE for details.