A transparent reverse proxy that mimics Anthropic's /v1/messages API shape,
so it's a drop-in replacement: point your existing client at this proxy's
URL instead of api.anthropic.com and change nothing else. It runs every
request through a multi-tier detection pipeline before forwarding it
upstream, and exposes two extra endpoints (/v1/tool-calls,
/v1/tool-results) so an agent framework can also gate and scan tool
activity that never goes through /v1/messages at all.
pip install -r requirements.txt
export ANTHROPIC_API_KEY=sk-ant-...
uvicorn main:app --reload --port 8000
curl http://localhost:8000/healthz
curl http://localhost:8000/v1/messages \
-H "x-api-key: $ANTHROPIC_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "content-type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 100,
"messages": [{"role": "user", "content": "Say hello in 5 words."}]
}'
Or point the Python SDK at it directly — zero code changes beyond base_url:
import anthropic
client = anthropic.Anthropic(
api_key="sk-ant-...",
base_url="http://localhost:8000",
)
msg = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=100,
messages=[{"role": "user", "content": "Say hello in 5 words."}],
)
print(msg)
Every request logs an audit line (request_id, action, tier,
latency_ms) and pushes a record onto a background-drained audit queue —
see Audit trail below.
/v1/messages, /v1/score-prompt
Tier 1 (DeBERTa prompt-injection classifier)
confident allow/block -> done
ambiguous -> Tier 2 (MiniLM + logistic regression) -> allow/block
/v1/tool-calls (call-gate: agent about to ACT)
risk_categories.py category (FINANCIAL / EXTERNAL_COMMS / READ_ONLY / DESTRUCTIVE / ...)
+ pii_detection.py (regex + checksum, e.g. Luhn) -> always block on hit
+ argument_validation.py (known recipient/contact identity check)
+ amount_anomaly.py (banking: statistical outlier on transaction amount)
+ content_risk_embeddings.py (message-body suspicion score)
+ url_reputation.py + url_lexical_classifier.py (any url/link argument)
-> confident local verdict, or escalate to:
+ llm_judge.py (Gemini one-shot judge, only for genuinely ambiguous cases)
/v1/tool-results (read-gate: agent about to READ a tool's output)
tool_result_scanner.py — DeBERTa PI detector at a fixed threshold, redacts on detection
Tier 1/2 gate the prompt; Tier 3 (/v1/tool-calls + /v1/tool-results)
gates the actions and tool I/O around it — both moments in an
agent's loop where an attack actually has to show up.
| Piece | Status |
|---|---|
/v1/messages pass-through to upstream | Real — forwards and streams |
| Async audit logging (ring buffer + background drain thread) | Real |
| Tier 1 — DeBERTa prompt-injection classifier | Real — see below |
| Tier 2 — MiniLM embedding + logistic regression, ambiguous-band only | Real — see below |
Tier 3 call-gate (/v1/tool-calls) | Real — tier3_policy.py, see below |
Tier 3 read-gate (/v1/tool-results) | Real — tool_result_scanner.py |
Dashboard (Streamlit, reads runtime/audit.db) | Real |
Domain-age/registrar/cert-age signals (domain_age.py) | Built + evaluated, not wired into the pipeline — age alone doesn't correlate with maliciousness for this attacker population, see the module's docstring |
Tier 0 — MCP tool-description scanner (tier0_mcp_scanner.py, structural_injection_signals.py) | Built + validated, then removed from main.py — real generalization gaps, backlogged |
| Cross-request / multi-step session-hijacking detection | Not started — needs new session-state architecture |
| Persistent-memory poisoning defense | Not started |
tier1_deberta_scorer.py scores every prompt with
protectai/deberta-v3-base-prompt-injection-v2. tau_low/tau_high come
from tier1_deberta_index.pkl (built by build_tier1_deberta_index.py),
calibrated only on a held-out calibration split
(eval_lodo_deberta_calibrated.py) — never the test set.
Honest cross-source numbers (real, not same-source): 79.2% fast-path resolution, 0.99 AUC. Everything that doesn't resolve confidently at Tier 1 escalates to Tier 2 instead of being forced into a threshold.
The original Tier 1 was TF-IDF + cosine similarity over an AdvBench-derived
vector library (ingest.py), with thresholds from a same-source ROC curve
(AUC 0.993). A true leave-one-dataset-out test against JailbreakHub (1,405
real in-the-wild jailbreak prompts, zero overlap with the training library)
told the real story:
| Metric | Same-source (ingest.py) | True LODO (JailbreakHub) |
|---|---|---|
| AUC | 0.993 | 0.706 |
| Fast-path resolution rate | 92% | 8% |
TF-IDF term-overlap generalizes poorly from AdvBench's templated strings to
real paraphrased jailbreak phrasing — almost nothing scored confidently
enough to resolve at Tier 1, so 92% of traffic escalated to Tier 2 instead
of the "mostly resolved at Tier 1" latency story holding up. Swapping to
all-MiniLM-L6-v2 sentence embeddings helped, but the DeBERTa
prompt-injection classifier (fine-tuned specifically for this task, not a
general-purpose embedding model repurposed for it) is what actually closed
the gap: 79.2% fast-path resolution vs. 8% for TF-IDF, on the same
cross-source test.
tier2_scorer.py only runs on Tier 1's ambiguous band (score_only
verdicts), using tier2_model.pkl (build_tier2_model.py) — a MiniLM
embedding feeding a logistic regression classifier, with its decision
threshold frozen from calibration-only data, never the held-out test set.
Honest held-out numbers: full pipeline (Tier 1 + Tier 2) recall 94.8%, FPR 2.9%, vs. Tier 1 alone's 88.6% recall / 0.86% FPR — a real uplift from escalating only the genuinely ambiguous slice rather than trying to force Tier 1 to resolve everything on its own.
/v1/tool-calls runs tier3_policy.py, which combines several
independently-validated mechanisms into one verdict (see the module's
docstring for the full escalation-scoping table):
risk_categories.py — categorizes a tool by what its name
suggests it does (financial, destructive, credential, access-control,
external-comms, read-only), replacing an earlier hardcoded banking tool
list. Token-based, not a per-tool-name blocklist — fixed a real bug
where \bdelete\b never matched inside delete_email (_ counts as a
word character).pii_detection.py — regex + checksum (Luhn for credit cards)
structured-identifier detection. Always blocks on a hit; deliberately
narrow and high-precision, complementing the broader semantic check
below rather than replacing it.argument_validation.py — known-recipient/contact identity check
(banking FINANCIAL, travel/workspace EXTERNAL_COMMS).amount_anomaly.py — Tukey's-fences statistical outlier detection
on transaction amounts, in log-space (financial amounts are
log-normally distributed; a raw-dollar fence let $0.01 probing attacks
through as "normal"). Catches wildly-anomalous amounts regardless of
whether the recipient is otherwise known.content_risk_embeddings.py — MiniLM-embedding suspicion score on
free-text arguments (message body/subject), for attacks where the
recipient is legitimate and only the content is malicious.url_reputation.py + url_lexical_classifier.py — any
url/link/webpage_url argument is checked against a local
URLhaus-derived blocklist (full-URL match, not domain-only — domain-only
false-positived on github.com hosting one attacker file) and scored by
a tiny BERT phishing classifier on the URL string itself.llm_judge.py — a Gemini one-shot judge, called only when the
fast local checks land on a genuinely ambiguous case (e.g. an unknown
recipient with a normal-looking amount) and user_task_prompt was
supplied. Deliberately scoped as an escalation, not a default path — real
LLM latency and API-quota risk make it unsuitable as the fast path.Real before/after on AgentDojo ground truth (eval_agentdojo_ground_truth.py,
zero API key required — tests whether the guard blocks AgentDojo's own
labeled attack calls):
| Fixed banking blocklist | + risk categories | + argument/content/amount checks | |
|---|---|---|---|
| Overall attack block rate | 19% | 38% | 60% |
| Overall legit false-block rate | 2% | 9% | 17% |
Final-call-in-chain block rate (many ground-truth attacks are multi-step chains where only the last call is actually harmful):
| Suite | Final (real) attack blocked |
|---|---|
| Banking | 9/9 = 100% |
| Slack | 4/5 = 80% |
| Travel | 5/6 = 83% |
| Workspace | 5/6 = 83% |
/v1/tool-results runs tool_result_scanner.py — the same DeBERTa PI
detector as Tier 1, but at a single fixed threshold (0.5, matching
AgentDojo's own published TransformersBasedPIDetector design) rather than
a calibrated ambiguous band — calibration was tried and abandoned here
because legitimate and malicious tool-result content overlap too much in
score, degenerating to tau_high=1.0. Detection redacts the flagged
content rather than aborting the run (a hard abort has a real utility
cost — AgentDojo's own numbers show utility under attack roughly halving
with abort-on-detect). Real, measured: 70.0% recall, 2.9% FPR on its own;
88.5% task-level attack stop rate from the call-gate alone, 93.8%
combined with this read-gate.
audit_ring_buffer.py — the request path never touches SQLite directly.
It pushes a record onto an in-memory queue (microseconds, no I/O) and a
single dedicated background thread drains it into runtime/audit.db in
batches. This replaced an earlier version that called sqlite3.connect()
synchronously on the request path, which serialized every concurrent
request behind each other's DB write. The queue is bounded — under
sustained overload it drops and counts (audit_dropped_total on
/healthz) rather than growing unbounded and OOMing the process.
The Streamlit dashboard reads runtime/audit.db live: event feed,
tier/action breakdown, latency distribution.
Run it in its own virtual environment, separate from the proxy's —
Streamlit and FastAPI want different starlette versions.
uv venv .venv-dashboard
source .venv-dashboard/bin/activate
uv pip install streamlit pandas plotly
streamlit run dashboard.py
Leave the proxy running in its original environment — the dashboard only
reads audit.db, it doesn't import anything from main.py.
eval_agentdojo_ground_truth.py — free, no API key: does the guard
block AgentDojo's own labeled attack tool calls. This is what the Tier 3
table above is built from.run_agentdojo_benchmark.py — real end-to-end run (needs
ANTHROPIC_API_KEY): runs the banking suite with and without
Tier3Guard inserted into the pipeline, reporting AgentDojo's own
utility-pass-rate / attack-success-rate metrics.run_agentdojo_benchmark_gemini.py — same harness against Google's free
AI Studio tier, for a free sanity check that the mechanism holds with a
different model in the loop (not a substitute for the Anthropic number,
since different models get fooled by injected content at different
rates).uvicorn main:app --port 8000 &
python eval_agentdojo_ground_truth.py
The proxy is intentionally framework-agnostic — it mimics the raw
/v1/messages shape rather than any orchestration framework's interface,
since LangChain, LangGraph, CrewAI, AutoGen, and hand-rolled loops all
ultimately call a raw LLM API underneath. langgraph_demo.py shows the
intended integration shape: every tool call routes through
/v1/tool-calls before it executes, demonstrated against an AgentDojo-style
indirect prompt injection (read_email returning content that tries to
hijack the agent into calling send_payment).
uvicorn main:app --port 8000 &
python langgraph_demo.py
/v1/score-prompt exists specifically for callers like a LangChain
BaseCallbackHandler that need a Tier 1/2 verdict on prompt content
without the proxy also making a real, billed call to the upstream model on
their behalf.
tier0_mcp_scanner.py, structural_injection_signals.py), then removed
from main.py. Real gaps found: generalization beyond the eval set, and
needs execution-layer signal it doesn't have yet.domain_age.py) — built and evaluated, but not
wired into tier3_policy.py: real data shows domain age alone doesn't
correlate with maliciousness for this attacker population (median 138
days, some 19+ years old). Registrar and cert-age were evaluated
separately and may still be worth wiring in; raw age is not.
5 commits
Python
85.6%
TypeScript
13.9%
A transparent reverse proxy that mimics Anthropic's /v1/messages API shape,
so it's a drop-in replacement: point your existing client at this proxy's
URL instead of api.anthropic.com and change nothing else. It runs every
request through a multi-tier detection pipeline before forwarding it
upstream, and exposes two extra endpoints (/v1/tool-calls,
/v1/tool-results) so an agent framework can also gate and scan tool
activity that never goes through /v1/messages at all.
pip install -r requirements.txt
export ANTHROPIC_API_KEY=sk-ant-...
uvicorn main:app --reload --port 8000
curl http://localhost:8000/healthz
curl http://localhost:8000/v1/messages \
-H "x-api-key: $ANTHROPIC_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "content-type: application/json" \
-d '{
"model": "claude-sonnet-4-6",
"max_tokens": 100,
"messages": [{"role": "user", "content": "Say hello in 5 words."}]
}'
Or point the Python SDK at it directly — zero code changes beyond base_url:
import anthropic
client = anthropic.Anthropic(
api_key="sk-ant-...",
base_url="http://localhost:8000",
)
msg = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=100,
messages=[{"role": "user", "content": "Say hello in 5 words."}],
)
print(msg)
Every request logs an audit line (request_id, action, tier,
latency_ms) and pushes a record onto a background-drained audit queue —
see Audit trail below.
/v1/messages, /v1/score-prompt
Tier 1 (DeBERTa prompt-injection classifier)
confident allow/block -> done
ambiguous -> Tier 2 (MiniLM + logistic regression) -> allow/block
/v1/tool-calls (call-gate: agent about to ACT)
risk_categories.py category (FINANCIAL / EXTERNAL_COMMS / READ_ONLY / DESTRUCTIVE / ...)
+ pii_detection.py (regex + checksum, e.g. Luhn) -> always block on hit
+ argument_validation.py (known recipient/contact identity check)
+ amount_anomaly.py (banking: statistical outlier on transaction amount)
+ content_risk_embeddings.py (message-body suspicion score)
+ url_reputation.py + url_lexical_classifier.py (any url/link argument)
-> confident local verdict, or escalate to:
+ llm_judge.py (Gemini one-shot judge, only for genuinely ambiguous cases)
/v1/tool-results (read-gate: agent about to READ a tool's output)
tool_result_scanner.py — DeBERTa PI detector at a fixed threshold, redacts on detection
Tier 1/2 gate the prompt; Tier 3 (/v1/tool-calls + /v1/tool-results)
gates the actions and tool I/O around it — both moments in an
agent's loop where an attack actually has to show up.
| Piece | Status |
|---|---|
/v1/messages pass-through to upstream | Real — forwards and streams |
| Async audit logging (ring buffer + background drain thread) | Real |
| Tier 1 — DeBERTa prompt-injection classifier | Real — see below |
| Tier 2 — MiniLM embedding + logistic regression, ambiguous-band only | Real — see below |
Tier 3 call-gate (/v1/tool-calls) | Real — tier3_policy.py, see below |
Tier 3 read-gate (/v1/tool-results) | Real — tool_result_scanner.py |
Dashboard (Streamlit, reads runtime/audit.db) | Real |
Domain-age/registrar/cert-age signals (domain_age.py) | Built + evaluated, not wired into the pipeline — age alone doesn't correlate with maliciousness for this attacker population, see the module's docstring |
Tier 0 — MCP tool-description scanner (tier0_mcp_scanner.py, structural_injection_signals.py) | Built + validated, then removed from main.py — real generalization gaps, backlogged |
| Cross-request / multi-step session-hijacking detection | Not started — needs new session-state architecture |
| Persistent-memory poisoning defense | Not started |
tier1_deberta_scorer.py scores every prompt with
protectai/deberta-v3-base-prompt-injection-v2. tau_low/tau_high come
from tier1_deberta_index.pkl (built by build_tier1_deberta_index.py),
calibrated only on a held-out calibration split
(eval_lodo_deberta_calibrated.py) — never the test set.
Honest cross-source numbers (real, not same-source): 79.2% fast-path resolution, 0.99 AUC. Everything that doesn't resolve confidently at Tier 1 escalates to Tier 2 instead of being forced into a threshold.
The original Tier 1 was TF-IDF + cosine similarity over an AdvBench-derived
vector library (ingest.py), with thresholds from a same-source ROC curve
(AUC 0.993). A true leave-one-dataset-out test against JailbreakHub (1,405
real in-the-wild jailbreak prompts, zero overlap with the training library)
told the real story:
| Metric | Same-source (ingest.py) | True LODO (JailbreakHub) |
|---|---|---|
| AUC | 0.993 | 0.706 |
| Fast-path resolution rate | 92% | 8% |
TF-IDF term-overlap generalizes poorly from AdvBench's templated strings to
real paraphrased jailbreak phrasing — almost nothing scored confidently
enough to resolve at Tier 1, so 92% of traffic escalated to Tier 2 instead
of the "mostly resolved at Tier 1" latency story holding up. Swapping to
all-MiniLM-L6-v2 sentence embeddings helped, but the DeBERTa
prompt-injection classifier (fine-tuned specifically for this task, not a
general-purpose embedding model repurposed for it) is what actually closed
the gap: 79.2% fast-path resolution vs. 8% for TF-IDF, on the same
cross-source test.
tier2_scorer.py only runs on Tier 1's ambiguous band (score_only
verdicts), using tier2_model.pkl (build_tier2_model.py) — a MiniLM
embedding feeding a logistic regression classifier, with its decision
threshold frozen from calibration-only data, never the held-out test set.
Honest held-out numbers: full pipeline (Tier 1 + Tier 2) recall 94.8%, FPR 2.9%, vs. Tier 1 alone's 88.6% recall / 0.86% FPR — a real uplift from escalating only the genuinely ambiguous slice rather than trying to force Tier 1 to resolve everything on its own.
/v1/tool-calls runs tier3_policy.py, which combines several
independently-validated mechanisms into one verdict (see the module's
docstring for the full escalation-scoping table):
risk_categories.py — categorizes a tool by what its name
suggests it does (financial, destructive, credential, access-control,
external-comms, read-only), replacing an earlier hardcoded banking tool
list. Token-based, not a per-tool-name blocklist — fixed a real bug
where \bdelete\b never matched inside delete_email (_ counts as a
word character).pii_detection.py — regex + checksum (Luhn for credit cards)
structured-identifier detection. Always blocks on a hit; deliberately
narrow and high-precision, complementing the broader semantic check
below rather than replacing it.argument_validation.py — known-recipient/contact identity check
(banking FINANCIAL, travel/workspace EXTERNAL_COMMS).amount_anomaly.py — Tukey's-fences statistical outlier detection
on transaction amounts, in log-space (financial amounts are
log-normally distributed; a raw-dollar fence let $0.01 probing attacks
through as "normal"). Catches wildly-anomalous amounts regardless of
whether the recipient is otherwise known.content_risk_embeddings.py — MiniLM-embedding suspicion score on
free-text arguments (message body/subject), for attacks where the
recipient is legitimate and only the content is malicious.url_reputation.py + url_lexical_classifier.py — any
url/link/webpage_url argument is checked against a local
URLhaus-derived blocklist (full-URL match, not domain-only — domain-only
false-positived on github.com hosting one attacker file) and scored by
a tiny BERT phishing classifier on the URL string itself.llm_judge.py — a Gemini one-shot judge, called only when the
fast local checks land on a genuinely ambiguous case (e.g. an unknown
recipient with a normal-looking amount) and user_task_prompt was
supplied. Deliberately scoped as an escalation, not a default path — real
LLM latency and API-quota risk make it unsuitable as the fast path.Real before/after on AgentDojo ground truth (eval_agentdojo_ground_truth.py,
zero API key required — tests whether the guard blocks AgentDojo's own
labeled attack calls):
| Fixed banking blocklist | + risk categories | + argument/content/amount checks | |
|---|---|---|---|
| Overall attack block rate | 19% | 38% | 60% |
| Overall legit false-block rate | 2% | 9% | 17% |
Final-call-in-chain block rate (many ground-truth attacks are multi-step chains where only the last call is actually harmful):
| Suite | Final (real) attack blocked |
|---|---|
| Banking | 9/9 = 100% |
| Slack | 4/5 = 80% |
| Travel | 5/6 = 83% |
| Workspace | 5/6 = 83% |
/v1/tool-results runs tool_result_scanner.py — the same DeBERTa PI
detector as Tier 1, but at a single fixed threshold (0.5, matching
AgentDojo's own published TransformersBasedPIDetector design) rather than
a calibrated ambiguous band — calibration was tried and abandoned here
because legitimate and malicious tool-result content overlap too much in
score, degenerating to tau_high=1.0. Detection redacts the flagged
content rather than aborting the run (a hard abort has a real utility
cost — AgentDojo's own numbers show utility under attack roughly halving
with abort-on-detect). Real, measured: 70.0% recall, 2.9% FPR on its own;
88.5% task-level attack stop rate from the call-gate alone, 93.8%
combined with this read-gate.
audit_ring_buffer.py — the request path never touches SQLite directly.
It pushes a record onto an in-memory queue (microseconds, no I/O) and a
single dedicated background thread drains it into runtime/audit.db in
batches. This replaced an earlier version that called sqlite3.connect()
synchronously on the request path, which serialized every concurrent
request behind each other's DB write. The queue is bounded — under
sustained overload it drops and counts (audit_dropped_total on
/healthz) rather than growing unbounded and OOMing the process.
The Streamlit dashboard reads runtime/audit.db live: event feed,
tier/action breakdown, latency distribution.
Run it in its own virtual environment, separate from the proxy's —
Streamlit and FastAPI want different starlette versions.
uv venv .venv-dashboard
source .venv-dashboard/bin/activate
uv pip install streamlit pandas plotly
streamlit run dashboard.py
Leave the proxy running in its original environment — the dashboard only
reads audit.db, it doesn't import anything from main.py.
eval_agentdojo_ground_truth.py — free, no API key: does the guard
block AgentDojo's own labeled attack tool calls. This is what the Tier 3
table above is built from.run_agentdojo_benchmark.py — real end-to-end run (needs
ANTHROPIC_API_KEY): runs the banking suite with and without
Tier3Guard inserted into the pipeline, reporting AgentDojo's own
utility-pass-rate / attack-success-rate metrics.run_agentdojo_benchmark_gemini.py — same harness against Google's free
AI Studio tier, for a free sanity check that the mechanism holds with a
different model in the loop (not a substitute for the Anthropic number,
since different models get fooled by injected content at different
rates).uvicorn main:app --port 8000 &
python eval_agentdojo_ground_truth.py
The proxy is intentionally framework-agnostic — it mimics the raw
/v1/messages shape rather than any orchestration framework's interface,
since LangChain, LangGraph, CrewAI, AutoGen, and hand-rolled loops all
ultimately call a raw LLM API underneath. langgraph_demo.py shows the
intended integration shape: every tool call routes through
/v1/tool-calls before it executes, demonstrated against an AgentDojo-style
indirect prompt injection (read_email returning content that tries to
hijack the agent into calling send_payment).
uvicorn main:app --port 8000 &
python langgraph_demo.py
/v1/score-prompt exists specifically for callers like a LangChain
BaseCallbackHandler that need a Tier 1/2 verdict on prompt content
without the proxy also making a real, billed call to the upstream model on
their behalf.
tier0_mcp_scanner.py, structural_injection_signals.py), then removed
from main.py. Real gaps found: generalization beyond the eval set, and
needs execution-layer signal it doesn't have yet.domain_age.py) — built and evaluated, but not
wired into tier3_policy.py: real data shows domain age alone doesn't
correlate with maliciousness for this attacker population (median 138
days, some 19+ years old). Registrar and cert-age were evaluated
separately and may still be worth wiring in; raw age is not.
5 commits
Python
85.6%
TypeScript
13.9%