Integration Examples

Examples of integrating elsai Guardrails with popular frameworks and agent runtimes.

Overview

Integration	Guardrail Type	Sample
Flask / FastAPI / Django / Streamlit	Content + token budget via LLMRails	Below
LangChain	Input/output wrapper	Below
LangGraph — Tool Authorization	Agent hooks	toolauth/agent_langgraph.py
LangGraph — Rate Limiting	Agent hooks + sessions	ratelimit/agent_langgraph.py
FastAPI — Data Exfiltration	Output checks on API responses	Below
FastAPI — ARMS Storage	Persist runs per request	Below

Content guardrails (toxicity, PII/PHI, semantic) integrate through LLMRails or GuardrailSystem.check_input() / check_output(). Agent guardrails (tool authorization, rate limiting) require hook nodes in your agent graph.

---

Flask Integration

from flask import Flask, request, jsonify
from elsai_guardrails.guardrails import LLMRails, RailsConfig

app = Flask(__name__)

config = RailsConfig.from_content(config_path="config.yml")
rails = LLMRails(config=config)

@app.route('/chat', methods=['POST'])
def chat():
    try:
        data = request.json
        messages = data.get('messages', [])

        if not messages:
            return jsonify({'error': 'No messages provided'}), 400

        result = rails.generate(messages=messages, return_details=True)

        if result['blocked']:
            return jsonify({
                'error': result['block_reason'],
                'message': result['final_response'],
                'token_budget_input_check': result.get('token_budget_input_check'),
                'token_budget_output_check': result.get('token_budget_output_check'),
            }), 400

        return jsonify({'response': result['final_response']})

    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True)

Handle common block_reason values: input, output, token_budget_input, token_budget_output, llm_error.

---

FastAPI Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from elsai_guardrails.guardrails import LLMRails, RailsConfig
from typing import List

app = FastAPI()

config = RailsConfig.from_content(config_path="config.yml")
rails = LLMRails(config=config)

class Message(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    messages: List[Message]

@app.post("/chat")
async def chat(request: ChatRequest):
    try:
        messages = [{"role": m.role, "content": m.content} for m in request.messages]
        result = await rails.generate_async(messages=messages, return_details=True)

        if result['blocked']:
            raise HTTPException(
                status_code=400,
                detail={
                    'error': result['block_reason'],
                    'message': result['final_response'],
                },
            )

        return {"response": result['final_response']}

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

---

Django Integration

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
import json
from elsai_guardrails.guardrails import LLMRails, RailsConfig

config = RailsConfig.from_content(config_path="config.yml")
rails = LLMRails(config=config)

@csrf_exempt
@require_http_methods(["POST"])
def chat_view(request):
    try:
        data = json.loads(request.body)
        messages = data.get('messages', [])

        if not messages:
            return JsonResponse({'error': 'No messages provided'}, status=400)

        result = rails.generate(messages=messages, return_details=True)

        if result['blocked']:
            return JsonResponse({
                'error': result['block_reason'],
                'message': result['final_response'],
            }, status=400)

        return JsonResponse({'response': result['final_response']})

    except Exception as e:
        return JsonResponse({'error': str(e)}, status=500)

---

Streamlit Integration

import streamlit as st
from elsai_guardrails.guardrails import LLMRails, RailsConfig

@st.cache_resource
def get_rails():
    config = RailsConfig.from_content(config_path="config.yml")
    return LLMRails(config=config)

rails = get_rails()

st.title("Chat with Guardrails")

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("What is up?"):
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("user"):
        st.markdown(prompt)

    result = rails.generate(
        messages=st.session_state.messages,
        return_details=True,
    )

    if result['blocked']:
        response = f"⚠️ [{result['block_reason']}] {result['final_response']}"
    else:
        response = result['final_response']

    st.session_state.messages.append({"role": "assistant", "content": response})

    with st.chat_message("assistant"):
        st.markdown(response)

---

LangChain Integration

Wrap a LangChain chain with input and output guardrail checks:

from elsai_guardrails.guardrails import GuardrailSystem, GuardrailConfig

guardrail = GuardrailSystem(config=GuardrailConfig())

class GuardedLLMChain:
    def __init__(self, chain, guardrail):
        self.chain = chain
        self.guardrail = guardrail

    def run(self, input_text):
        input_result = self.guardrail.check_input(input_text)
        if not input_result.passed:
            return f"Input blocked: {input_result.message}"

        response = self.chain.run(input_text)

        output_result = self.guardrail.check_output(response)
        if not output_result.passed:
            return f"Output blocked: {output_result.message}"

        return response

---

LangGraph — Tool Authorization

Restrict which tools each role can invoke using a pre-execution authorization node.

Repository samples:

• Config: toolauth/config.yaml
• Agent: toolauth/agent_langgraph.py

Graph flow: agent → authorization → tools → agent

Policy

guardrails:
  tool_authorization:
    enabled: true
    denied_tools:
      - execute_shell
    sensitive_tools:
      - delete_record
    roles:
      analyst:
        allowed_tools:
          - search_web
          - calculator
      engineer:
        allowed_tools:
          - search_web
          - calculator
          - delete_record

Setup

from elsai_guardrails.guardrails import GuardrailSystem
from elsai_guardrails.guardrails.guardrail_policy import GuardrailPolicy
from langchain_core.messages import AIMessage, ToolMessage
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode

guardrails = GuardrailSystem(
    guardrail_policy=GuardrailPolicy.from_file("toolauth/config.yaml"),
)

Authorization Node

def authorization_node(state):
    """Run before ToolNode — block unauthorized calls without executing tools."""
    last_message: AIMessage = state["messages"][-1]
    if not getattr(last_message, "tool_calls", None):
        return {}

    blocked = []
    for call in last_message.tool_calls:
        result = guardrails.before_tool_call(
            tool_name=call["name"],
            user_role=state.get("user_role", "unknown"),
            raise_on_block=False,
        )
        if not result.passed:
            blocked.append(
                ToolMessage(
                    tool_call_id=call["id"],
                    content=f"GUARDRAIL BLOCKED: {result.error}",
                )
            )

    return {"messages": blocked} if blocked else {}

Routing

def should_continue(state):
    last = state["messages"][-1]
    if isinstance(last, ToolMessage):
        return "agent"   # blocked — surface error to LLM
    if isinstance(last, AIMessage) and last.tool_calls:
        return "authorization"
    return END

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("authorization", authorization_node)
graph.add_node("tools", ToolNode(TOOLS))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "authorization": "authorization", END: END,
})
graph.add_conditional_edges("authorization", lambda s: (
    "tools" if isinstance(s["messages"][-1], AIMessage) else "agent"
), {"tools": "tools", "agent": "agent"})
graph.add_edge("tools", "agent")
app = graph.compile()

Run the demo:

cd toolauth
pip install langgraph langchain-openai langchain-core pyyaml python-dotenv
python agent_langgraph.py

See Tool Authorization for policy details.

---

LangGraph — Rate Limiting

Enforce per-session request and tool call quotas with a rate-limit node and session tracking.

Repository samples:

• Config: ratelimit/config.yaml
• Agent: ratelimit/agent_langgraph.py

Graph flow: agent → rate_limit → tools → agent

Policy

guardrails:
  rate_limit:
    enabled: true
    max_requests_per_session: 5
    max_tool_calls_per_session: 50
    max_tool_execution_seconds: 60

Setup

from elsai_guardrails.guardrails import GuardrailPolicy, GuardrailSystem

guardrails = GuardrailSystem(
    guardrail_policy=GuardrailPolicy.from_file("ratelimit/config.yaml"),
)
rate_limit_config = guardrails.guardrail_policy.to_rate_limit_config()

session = guardrails.create_session()
session_id = session.session_id

Agent Node (Request Limit)

def agent_node(state):
    result = guardrails.before_request(state["session_id"], raise_on_block=False)
    if not result.passed:
        return {"messages": [AIMessage(content=f"RATE LIMIT BLOCKED: {result.error}")]}
    response = llm_with_tools.invoke([SYSTEM_PROMPT] + state["messages"])
    return {"messages": [response]}

Rate Limit Node (Tool Call Quota)

def rate_limit_node(state):
    last: AIMessage = state["messages"][-1]
    if not getattr(last, "tool_calls", None):
        return {}

    blocked = []
    for call in last.tool_calls:
        result = guardrails.check_tool_call_limit(
            state["session_id"], raise_on_block=False,
        )
        if not result.passed:
            blocked.append(ToolMessage(
                tool_call_id=call["id"],
                content=f"RATE LIMIT BLOCKED: {result.error}",
            ))
    return {"messages": blocked} if blocked else {}

Inside Tools (Record + Timer)

@tool
def search_web(query: str, session_id: str) -> str:
    guardrails.record_tool_call(session_id)
    t = guardrails.start_execution_timer()
    result = f"Results for: {query}"
    guardrails.end_execution_timer(t)
    return result

Run the demo:

cd ratelimit
pip install langgraph langchain-openai langchain-core pyyaml python-dotenv
python agent_langgraph.py

See Rate Limiting for full details.

---

LangGraph — Combined Agent Safety

Use one policy file for both tool authorization and rate limiting. See ratelimit/config.yaml which includes both sections.

guardrails:
  rate_limit:
    enabled: true
    max_requests_per_session: 5
    max_tool_calls_per_session: 50
    max_tool_execution_seconds: 60

  tool_authorization:
    enabled: true
    denied_tools:
      - execute_shell
    sensitive_tools:
      - delete_record
    roles:
      analyst:
        allowed_tools:
          - search_web
          - calculator

Recommended graph with both guardrails:

agent (before_request)
  → authorization (before_tool_call)
  → rate_limit (check_tool_call_limit)
  → tools (record_tool_call + timer)
  → agent

See Advanced Examples Example 17 for a programmatic combined safety step.

---

FastAPI — Data Exfiltration on Output

Validate LLM responses before returning them to API clients. Block or mask credential leaks and bulk PII exports.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from elsai_guardrails.guardrails import GuardrailConfig, GuardrailSystem
from elsai_guardrails.guardrails.guardrail_policy import GuardrailPolicy

app = FastAPI()

policy = GuardrailPolicy.from_file("config.yaml")
guardrails = GuardrailSystem(
    config=GuardrailConfig(
        check_toxicity=False,
        check_sensitive_data=False,
        check_semantic=False,
    ),
    output_checks=True,
    guardrail_policy=policy,
)

class GenerateRequest(BaseModel):
    llm_output: str

@app.post("/validate-output")
def validate_output(body: GenerateRequest):
    result = guardrails.check_output(body.llm_output)

    if not result.passed:
        raise HTTPException(
            status_code=400,
            detail={
                "message": result.message,
                "exfiltration": result.exfiltration,
            },
        )

    exfil = result.exfiltration or {}
    response_text = exfil.get("processed_text", body.llm_output)

    return {
        "response": response_text,
        "exfiltration": exfil,
    }

See Data Exfiltration Detection.

---

FastAPI — ARMS Storage per Request

Persist guardrail check results to the ARMS Backend for each API request.

import os
import uuid
from fastapi import FastAPI
from pydantic import BaseModel
from elsai_guardrails.guardrails import GuardrailConfig, GuardrailSystem
from elsai_guardrails.guardrails.guardrail_policy import GuardrailPolicy

app = FastAPI()

policy = GuardrailPolicy.from_file("config.yaml")
guardrails = GuardrailSystem(
    config=GuardrailConfig(),
    guardrail_policy=policy,
)._with_storage_hook()

class ChatRequest(BaseModel):
    message: str

@app.post("/chat")
def chat(body: ChatRequest):
    run_id = uuid.uuid4().hex
    project_id = os.environ.get("GUARDRAILS_ARMS_PROJECT_ID", "demo-project-001")

    guardrails.link_run_context(
        run_id=run_id,
        project_id=project_id,
        project="demo-app",
    )

    with guardrails.storage_run_context(session_id=run_id):
        input_result = guardrails.check_input(body.message)
        if not input_result.passed:
            guardrails.end_run()
            return {"blocked": True, "stage": "input", "message": input_result.message}

        # Replace with your LLM call
        llm_output = f"Echo: {body.message}"
        output_result = guardrails.check_output(llm_output)
        if not output_result.passed:
            guardrails.end_run()
            return {"blocked": True, "stage": "output", "message": output_result.message}

    saved = guardrails.end_run()
    return {"response": llm_output, "run_id": saved}

Environment:

export API_BASE_URL=https://your-arms-backend
export ELSAI_ARMS_API_KEY=your-api-key
export GUARDRAILS_ARMS_PROJECT_ID=demo-project-001

The Backend routes persistence to MongoDB, DynamoDB, or ClickHouse based on your ARMS deployment. See ARMS Storage.

---

Next Steps

• Basic Examples — Token budget, PII/PHI, tool auth, rate limit, exfiltration, storage
• Advanced Examples — Combined agent safety, exfiltration tuning, redacted storage
• Tool Authorization
• Rate Limiting
• Data Exfiltration Detection
• ARMS Storage
• Token Budget Enforcement