langchain-deepkeep 0.1.0

DeepKeep AI Firewall integration for LangChain (LCEL)

langchain-deepkeep

LangChain integration for the DeepKeep AI Firewall. Slot DeepKeep guardrails into any LCEL chain to check user input and LLM output before they reach — or leave — your model.

Installation

pip install langchain-deepkeep

Dependencies: langchain-core >= 0.2, httpx >= 0.27

Quick start

DeepKeepFirewall — wrap any chain (recommended)

The simplest way to add guardrails. Pass your existing chain and DeepKeep will check both the user's input and the LLM's response, sharing a single conversation.

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_deepkeep import DeepKeepFirewall

chain = DeepKeepFirewall(
    chain=ChatOpenAI(model="gpt-4o") | StrOutputParser(),
    api_key="dk-...",
    subdomain="acme",        # your DeepKeep environment subdomain
    firewall_id="fw-xxx",    # from the DeepKeep dashboard
)

result = chain.invoke("Tell me something interesting.")

To skip output checking:

chain = DeepKeepFirewall(..., check_output=False)

DeepKeepGuardrails — compose manually with |

Use this when you need finer control — for example, only one direction, or sharing a conversation ID across multiple guards.

from langchain_deepkeep import DeepKeepGuardrails

input_guard = DeepKeepGuardrails(
    api_key="dk-...", subdomain="acme", firewall_id="fw-xxx", mode="input"
)
output_guard = DeepKeepGuardrails(
    api_key="dk-...", subdomain="acme", firewall_id="fw-xxx", mode="output",
    conversation_id=input_guard.conversation_id,   # share the same conversation
)

chain = input_guard | llm | StrOutputParser() | output_guard

Async

Both classes support ainvoke. If you are in an async context and want to avoid a blocking conversation-creation call at startup, use the async_create classmethod:

guard = await DeepKeepGuardrails.async_create(
    api_key="dk-...", subdomain="acme", firewall_id="fw-xxx",
)

fw = await DeepKeepFirewall.async_create(
    chain=llm | StrOutputParser(),
    api_key="dk-...", subdomain="acme", firewall_id="fw-xxx",
)

Utilities

from langchain_deepkeep import create_conversation, acreate_conversation, list_firewalls

# List firewalls on your account
firewalls = list_firewalls(api_key="dk-...", subdomain="acme")
# [{"id": "fw-xxx", "name": "Production Firewall"}, ...]

# Create a conversation explicitly (e.g. one per user session)
conv_id = create_conversation(api_key="dk-...", subdomain="acme", firewall_id="fw-xxx")
conv_id = await acreate_conversation(...)   # async variant

Error handling

from langchain_deepkeep import ContentBlockedError, DeepKeepAPIError

try:
    result = chain.invoke(user_message)
except ContentBlockedError as e:
    # Content violated a firewall policy — show e.args[0] to the user
    return str(e)
except DeepKeepAPIError as e:
    # Unexpected API error (4xx/5xx) or network failure
    # e.status_code is the HTTP status, or None for network errors
    log.error("DeepKeep error %s: %s", e.status_code, e)
    raise

Retry behaviour

All HTTP calls automatically retry on 429 Too Many Requests (rate limits and firewall warmup). The Retry-After response header is respected; exponential backoff with random jitter is applied on top.

Default: max_retries=3. To disable retries entirely pass max_retries=0.

guard = DeepKeepGuardrails(..., max_retries=5)   # more patient
guard = DeepKeepGuardrails(..., max_retries=0)   # fail fast

Configuration reference

Parameter	Type	Default	Description
api_key	str	—	DeepKeep API key
subdomain	str	—	Your environment subdomain (e.g. "acme")
firewall_id	str	—	Firewall ID from the dashboard
conversation_id	str | None	None	Existing conversation; auto-created if omitted
mode	"input" | "output"	"input"	Which endpoint to call (DeepKeepGuardrails only)
error_message	str	"Your message was blocked..."	Message inside ContentBlockedError
logs	bool	False	Return all violations instead of stopping at the first
timeout	float	30.0	HTTP timeout in seconds
max_retries	int	3	Max retries on HTTP 429

DeepKeepFirewall additionally accepts chain, check_output, input_error_message, and output_error_message.

Running the tests

pip install -e ".[dev]"
pytest

The test suite uses respx to mock all HTTP calls — no API credentials or network access required.