langchain-senthex 0.1.0

LangChain provider for Senthex Proxy — EU-hosted AI firewall with parallel runtime shields.

langchain-senthex

LangChain provider for Senthex Proxy — EU-hosted AI firewall with parallel runtime shields.

A first-class LangChain chat-model integration for Senthex Proxy — an EU-hosted AI firewall that runs 26 parallel shields (prompt-injection detection, PII redaction, secrets, unicode steganography, behavioural anomaly, semantic hijack, …) on every LLM call and emits an EU AI Act Article 15-aligned audit trail.

langchain-senthex is a thin facade over langchain-openai, langchain-anthropic and langchain-mistralai. It routes calls through Senthex Proxy and surfaces every shield verdict on response.response_metadata["senthex"] so you can compose them with RunnableBranch, retries and fallbacks.

---

Quickstart — 60 seconds

pip install langchain-senthex[openai]
export SENTHEX_API_KEY="snx-..."   # free tier: 1000 req/month, no credit card
export OPENAI_API_KEY="sk-..."

from langchain_senthex import ChatSenthex

chat = ChatSenthex(provider="openai", model="gpt-4o-mini")
print(chat.invoke("Hello world").content)

That's it. The call goes through https://app.senthex.com/v1, all 26 shields fire in parallel server-side, and the upstream OpenAI response flows back with shield verdicts attached.

---

What you get out of the box

• 26 parallel runtime shields — prompt-injection detection, PII redaction, secrets, unicode steganography (homoglyphs / zero-width / bidi), behavioural fingerprinting, semantic hijack, output validation, code-danger scoring, response policy enforcement, … See docs/shields.md for the full list.
• EU AI Act Article 15 audit trail — every call logs {timestamp, action_taken, reason, data_classification} for traceability. Exportable from the Senthex dashboard.
• RGPD by design — Senthex Proxy hosted at Hetzner DE, no log-rotation outside the EU, no third-party processors.
• Multi-provider — provider="openai", provider="anthropic", provider="mistral" with the same idiomatic LangChain surface.
• Free tier — 1 000 requests/month, no credit card needed. Upgrade only when you ship.

---

Reading shield verdicts

Every response carries a normalised senthex payload on response.response_metadata:

resp = chat.invoke("My email is [e-mail], please send the report.")

verdict = resp.response_metadata["senthex"]
verdict["shield_status"]       # "pass" | "warn" | "block"
verdict["injection_score"]     # 0.0 .. 1.0 — prompt-injection probability
verdict["pii_found"]           # int — number of PII findings
verdict["data_types"]          # ["EMAIL"] — PII categories detected
verdict["data_classification"] # "PUBLIC" | "INTERNAL" | "CONFIDENTIAL" | "RESTRICTED"
verdict["latency_ms"]          # float — proxy latency excluding upstream
verdict["request_id"]          # "snx_req_xxx" — visible in the Senthex dashboard
verdict["plan"]["tier"]        # "free" | "pro" | "enterprise"
verdict["plan"]["usage_month"] # int — month-to-date request count

verdict["raw_headers"]         # full X-Senthex-* dict for power users

Compose with LangChain runnables:

from langchain_core.runnables import RunnableBranch

pipeline = RunnableBranch(
    (lambda r: r.response_metadata["senthex"]["shield_status"] == "block",
     lambda _: AIMessage(content="Blocked by Senthex shields.")),
    chat,  # default branch
)

---

Agent Trail — session tracking (preview)

Group multi-turn agent calls under a single session id and retrieve the full trail from the Senthex dashboard:

chat = ChatSenthex(
    provider="anthropic",
    model="claude-3-5-sonnet-latest",
    senthex_session_id="sess_abc123",
)

for prompt in ["plan the trip", "book the flight", "summarise"]:
    chat.invoke(prompt)

# → All three calls grouped at https://app.senthex.com/sessions/sess_abc123
# → Cumulative injection score, turn count, and risk score surfaced on
#   every response.response_metadata["senthex"]["session"].

Session ids are alphanumerics + dash/underscore, max 128 chars. Senthex Proxy enforces per-tier session caps; over-quota session ids are silently dropped (the request still succeeds — only grouping is skipped).

---

Streaming

Streaming is supported on every provider. Shield verdicts arrive at the last chunk of the stream so accumulators don't double-merge them:

for chunk in chat.stream("Tell me a short joke about Kubernetes"):
    print(chunk.content, end="", flush=True)

# Final accumulated message carries response_metadata["senthex"] as usual.

Async variants (ainvoke, astream) work the same way.

---

Tool calling and structured output

ChatSenthex forwards .bind_tools(...) and .with_structured_output(...) to the underlying partner package, so anything that works on ChatOpenAI / ChatAnthropic / ChatMistralAI works on ChatSenthex:

from pydantic import BaseModel

class WeatherQuery(BaseModel):
    city: str
    units: str = "metric"

chat.with_structured_output(WeatherQuery).invoke("Paris in Celsius")
# → WeatherQuery(city='Paris', units='metric')

Senthex Proxy logs every tool call into the audit trail and can enforce an allowed_tools policy at the project level.

---

Side-by-side — without and with Senthex

Without Senthex (raw OpenAI, no shields, no audit):

from langchain_openai import ChatOpenAI
chat = ChatOpenAI(model="gpt-4o-mini")
chat.invoke("Hello world")

With Senthex (1 import change, 26 shields + Article 15 audit):

from langchain_senthex import ChatSenthex
chat = ChatSenthex(provider="openai", model="gpt-4o-mini")
chat.invoke("Hello world")

That's the cost of getting shields. One line.

---

Error handling

The package exposes typed exceptions you can branch on:

from langchain_senthex import (
    SenthexShieldBlockError,   # a shield blocked the request
    SenthexQuotaError,         # plan / quota exhausted
    SenthexSessionQuarantinedError,  # session id has been quarantined
    SenthexAuthError,          # missing or invalid Senthex / upstream key
    SenthexUpstreamError,      # upstream provider error, normalised
)

try:
    chat.invoke(suspicious_input)
except SenthexShieldBlockError as e:
    print("Blocked:", e.verdict["shield_status"], e.verdict["injection_score"])

Upstream provider exceptions (openai.APIError, anthropic.APIError, …) bubble up unchanged so your existing LangChain error handlers keep working.

---

Installation variants

Command	What it pulls
pip install langchain-senthex[openai]	+ langchain-openai
pip install langchain-senthex[anthropic]	+ langchain-anthropic
pip install langchain-senthex[mistral]	+ langchain-mistralai
pip install langchain-senthex[all]	all three partner packages
pip install langchain-senthex	base only — pick a provider extra before instantiating

The base package depends on langchain-core>=0.3,<0.4, httpx>=0.27, pydantic>=2. No heavy dependencies; install completes under 5 seconds.

---

Authentication

Keys are resolved in this order:

1. Explicit kwarg to ChatSenthex(...)
2. Environment variable (SENTHEX_API_KEY, OPENAI_API_KEY, ANTHROPIC_API_KEY, MISTRAL_API_KEY)
3. SenthexAuthError at construction time

SENTHEX_API_KEY is always required. The upstream-provider key (one of OPENAI_*, ANTHROPIC_*, MISTRAL_*) is required based on provider=.

Both are wrapped in pydantic.SecretStr so they don't leak in logs or repr.

---

Links

• Sign up free — no credit card
• Senthex docs
• Manifesto: agents make the proxy load-bearing
• Examples repo
• Issues

---

Disclaimer

Senthex is not affiliated with LangChain, Inc. This is a community-maintained integration. LangChain is a trademark of LangChain, Inc.

"Article 15 support" means this package and Senthex Proxy emit audit-trail metadata aligned with the requirements of EU AI Act Article 15. It is not a certification of compliance — your compliance posture also depends on how you store and act on the audit trail.

License

MIT — see LICENSE.