← Back to Documentation

Framework Middleware

v0.9.0

Drop-in security adapters for LangChain, CrewAI, Haystack, and AutoGen/AG2. Add runtime prompt injection defence to your existing agent stack with one line of code.

InstallationLangChain MiddlewareLangChain CallbackCrewAIHaystackAutoGen / AG2Policy OptionsChoosing an Adapter

Installation

Install the extras package for your framework. Each package pulls in the framework dependency alongside sentinel-security.

All frameworks Recommended

pip install "sentinel-security[frameworks]"

Individual frameworks

LangChain

pip install "sentinel-security[langchain]"

CrewAI

pip install "sentinel-security[crewai]"

Haystack

pip install "sentinel-security[haystack]"

AutoGen / AG2

pip install "sentinel-security[autogen]"

LangChain — SentinelMiddleware

langchain >= 1.0.0Supports blocking

The SentinelMiddleware integrates with LangChain's AgentMiddleware API (LangChain ≥ 1.0.0). It hooks into before_model and after_model to scan input messages before they reach the LLM, and scan the LLM's response before it enters the agent loop. Supports full blocking — raises SentinelBlockedError when a threat is detected.

What it scans

  • Input messages before the LLM call (before_model hook)
  • LLM response and tool output (after_model hook)

Usage

from langchain_openai import ChatOpenAI
from sentinel_security.middleware import SentinelMiddleware

# Add to create_agent() middleware list
agent = create_agent(
    model=ChatOpenAI(),
    tools=[...],
    middleware=[SentinelMiddleware()],
)

# With explicit policy and threshold
agent = create_agent(
    model=ChatOpenAI(),
    tools=[...],
    middleware=[
        SentinelMiddleware(
            policy="block",          # "log" | "warn" | "block"
            risk_threshold=7,        # 0–10, default from global config
        )
    ],
)
⚠

Requires LangChain ≥ 1.0.0 for the AgentMiddleware API. For older LangChain or LangGraph setups, use SentinelCallbackHandler instead.

Handling blocked content

from sentinel_security.middleware._types import SentinelBlockedError

try:
    result = agent.invoke({"messages": [...]})
except SentinelBlockedError as e:
    print(f"Blocked: {e.result.threats}")
    # Handle gracefully — e.g. return a safe fallback response

LangChain — SentinelCallbackHandler

langchain-core >= 0.3.0Observation only

The SentinelCallbackHandler extends LangChain's BaseCallbackHandler to scan LLM inputs, outputs, and tool results at every step. It works with legacy LangChain (< 1.0), LangGraph's create_react_agent, and raw BaseChatModel usage.

The callback handler is observation-only — it logs and emits events but cannot block requests. Use SentinelMiddleware with policy="block" if you need blocking support.

What it scans

  • String prompts sent to non-chat LLMs (on_llm_start)
  • Chat messages sent to the model (on_chat_model_start)
  • LLM response generations (on_llm_end)
  • Tool output for injection and exfiltration (on_tool_end)

Usage with LangGraph

from sentinel_security.middleware import SentinelCallbackHandler

# LangGraph create_react_agent
result = agent.invoke(
    {"messages": [HumanMessage(content="...")]},
    config={"callbacks": [SentinelCallbackHandler()]},
)

# With policy and threshold
handler = SentinelCallbackHandler(
    policy="warn",      # "log" | "warn" | "block" (ignored — observation only)
    risk_threshold=6,
)
result = agent.invoke({"messages": [...]}, config={"callbacks": [handler]})

Usage with raw BaseChatModel

from langchain_openai import ChatOpenAI
from sentinel_security.middleware import SentinelCallbackHandler

model = ChatOpenAI(callbacks=[SentinelCallbackHandler()])
response = model.invoke("Summarise this document: ...")

SentinelMiddleware vs SentinelCallbackHandler

CapabilitySentinelMiddlewareSentinelCallbackHandler
Blocking support✓ Yes✗ No
LangChain ≥ 1.0 AgentMiddleware✓ Required✗ N/A
LangGraph / legacy LangChain✗ No✓ Yes
Raw BaseChatModel✗ No✓ Yes
Tool output scanning✓ Yes✓ Yes

CrewAI — SentinelCrewAI

crewai >= 0.80Supports blocking

SentinelCrewAI wraps CrewAI's step_callback and task_callback hooks on each agent to scan step outputs and inter-agent task results. Supports blocking via SentinelBlockedError.

When LangChain's SentinelMiddleware is also active (CrewAI uses LangChain internally), the adapter automatically deduplicates scans of identical content within the same session to avoid double-scanning.

Protect an entire crew

from crewai import Agent, Crew, Task
from sentinel_security.middleware.crewai import SentinelCrewAI

researcher = Agent(
    role="Researcher",
    goal="Research AI security topics",
    backstory="Expert security researcher",
)

writer = Agent(
    role="Writer",
    goal="Write clear technical content",
    backstory="Technical writer with security focus",
)

crew = Crew(agents=[researcher, writer], tasks=[...])

# Wrap all agents in the crew with Sentinel security scanning
SentinelCrewAI.protect(crew, policy="block", risk_threshold=7)

Protect a single agent

from crewai import Agent
from sentinel_security.middleware.crewai import SentinelCrewAI

agent = Agent(
    role="Data Analyst",
    goal="Analyse external datasets",
    backstory="Data analyst working with untrusted external data",
)

# Protect just this agent
SentinelCrewAI.protect_agent(agent, policy="warn")

Haystack — SentinelGuard

haystack-ai >= 2.0Routes blocked docs to separate output

SentinelGuard is a native Haystack @component that scans a batch of Document objects for prompt injection. It splits output into two ports: documents (clean) and blocked (detected threats). Each document's meta dict is annotated with scan results.

Document metadata annotations

Every scanned document receives these keys in its meta dict:

KeyDescription
sentinel_risk_scoreRisk score 0–10
sentinel_threatsList of detected threat type strings
sentinel_actionAction taken: "log", "warn", or "block"
sentinel_scan_idUnique scan ID for audit trail

Usage in a pipeline

from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from sentinel_security.middleware.haystack import SentinelGuard

pipeline = Pipeline()
pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=...))
pipeline.add_component("sentinel", SentinelGuard(policy="block"))
pipeline.add_component("prompt_builder", PromptBuilder(template="..."))

# Connect retriever output through sentinel before the prompt builder
pipeline.connect("retriever.documents", "sentinel.documents")
pipeline.connect("sentinel.documents", "prompt_builder.documents")

# Run the pipeline — blocked documents are routed to sentinel.blocked
result = pipeline.run({"retriever": {"query": "AI security"}})

Handling blocked documents

# Access blocked documents from the result
blocked_docs = result.get("sentinel", {}).get("blocked", [])
for doc in blocked_docs:
    print(f"Blocked: {doc.meta['sentinel_threats']} (score: {doc.meta['sentinel_risk_score']})")

AutoGen / AG2 — SentinelAutoGen

ag2 >= 0.4Replaces blocked messages

SentinelAutoGen registers process_last_received_message (inbound) and process_message_before_send (outbound) hooks on any ConversableAgent to scan messages in both directions. When a message is blocked, the hook returns a safe replacement string rather than raising an exception — following the AutoGen hook API contract.

Both ag2 and the legacy autogen package are supported.

Usage

import ag2
from sentinel_security.middleware.autogen import SentinelAutoGen

# Create your AG2 agent
assistant = ag2.AssistantAgent(
    name="assistant",
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": "..."}]},
)

# Register Sentinel hooks on the agent
SentinelAutoGen.protect(assistant, policy="block", risk_threshold=7)

# Scan inbound and outbound messages from this point forward
user_proxy = ag2.UserProxyAgent(name="user_proxy", human_input_mode="NEVER")
user_proxy.initiate_chat(assistant, message="Analyse this data: ...")

Custom blocked message

SentinelAutoGen.protect(
    assistant,
    policy="block",
    blocked_message="I cannot process this content as it appears to contain unsafe instructions.",
)

How blocking works in AutoGen

Unlike SentinelMiddleware which raises an exception, the AutoGen adapter returns a replacement message when content is blocked. This keeps the agent loop running — the agent receives a safe fallback string and can continue the conversation. The detection event is still logged and alerts are dispatched to your configured destinations.

Policy Options

All adapters accept the same policy and risk_threshold arguments. If omitted, the values fall back to your global sentinel_security.configure() settings.

PolicyBehaviour
logDetect and log threats. No blocking. Lowest friction.
warnLog and dispatch alerts to configured destinations. No blocking.
blockBlock content when risk score exceeds risk_threshold. Log and alert.

Global configuration

import sentinel_security

sentinel_security.configure(
    policy="block",
    risk_threshold=7,           # 0–10, default is 6
    event_handler=my_handler,   # optional: receives ScanEvent objects
)

# All middleware adapters inherit this config unless overridden per-instance

Choosing an Adapter

LangChain ≥ 1.0

Use SentinelMiddleware with create_agent() for full blocking support at both the input and response scan points.

SentinelMiddleware →

LangGraph / legacy LangChain

Use SentinelCallbackHandler — works anywhere callbacks are accepted, including raw BaseChatModel usage.

SentinelCallbackHandler →

CrewAI crews

Use SentinelCrewAI.protect(crew) to wrap every agent at once. Use protect_agent(agent) for targeted protection.

SentinelCrewAI →

Haystack RAG pipelines

Insert SentinelGuard between your retriever and prompt builder to filter injected documents before they reach the LLM.

SentinelGuard →

AutoGen / AG2 agents

Use SentinelAutoGen.protect(agent) to register inbound and outbound hooks on any ConversableAgent.

SentinelAutoGen →

Need help?

Questions about integrating Sentinel into your agent stack? We're here to help.

Contact SupportBack to Docs