Agentic DM Gateway

Sicherheits-Kontrollebene für LLM-Agenten über private Chats (typischerweise Discord-DMs).

Es sitzt vor deinem Agenten. Es entscheidet, wer sprechen darf, ob die Sitzung entsperrt ist, ob der Prozess angehalten ist und ob diese Nachricht sicher genug ist, um weitergeleitet zu werden. Dein Modell und deine Tools bleiben hinter diesem Tor. Die Bibliothek ruft kein LLM auf. Sie implementiert keine Produktfunktionen über die Sicherheit hinaus.

Hermes-inspiriert. Das Design folgt denselben Ideen der Kontrollebene, die in Hermes Agent-Messaging-Gateways verwendet werden: DM-first-Zustellung, Identitäts-Allowlists, Pairing-artiges Öffnen, Besitzer-Kill-Switch und eine harte Trennung zwischen wer handeln darf (Kontrollebene) und Nachrichtentext, den das Modell sieht (Datenebene). Dieses Paket ist ein kleiner, eigenständiger Extrakt dieses Musters für jeden aufrufbaren Agenten. Nicht verbunden mit Nous Research.

Reifegrad: implementiert · unabhängig validiert · gewartet. Siehe STATUS.md. Reproduzieren: python scripts/repro.py (erwartet REPRO_OK).

Offline-Tests:

pip install -e ".[dev]"   # or: pip install -e . && pip install pytest
python -m pytest -q --tb=line
# or: python scripts/repro.py

Live: https://github.com/SamsonCyber/agentic-dm-gateway

Das Problem

Wenn du einen Agenten mit Tools auf Discord (oder einer beliebigen Chat-API) bereitstellst, kann jeder, der den Bot anschreiben kann, versuchen:

• den Agenten ohne Erlaubnis zu nutzen
• API-Kontingent durch Fluten zu verbrennen
• Prompts im Stil von „ignore previous instructions“ einzuschleusen
• das Modell dazu zu bringen, API-Keys oder andere Geheimnisse preiszugeben

Du brauchst eine Kontrollebene (Identitäts- und Prozesskontrollen), getrennt von der Datenebene (Nachrichtentext, den das Modell sieht).

Dieses Paket ist diese Kontrollebene.

Was es tut

Umfang: ausschließlich Sicherheits-Gate. Kein Chatbot, Handelsbot, Scanner oder Agent-Framework. Übergib einen agent(user_id, text) -> str (oder async), wenn du Discord nutzt. Der Kern funktioniert mit jeder ganzzahligen Benutzer-ID und Klartext.

Demo (Copy-and-Paste)

$ python - <<'PY'
from agentic_dm_gateway import InboundSecurityPipeline
pipe = InboundSecurityPipeline({
 "allowed_user_ids": [111],
 "owner_ids": [111],
 "pin_enabled": False,
 "block_injection": True,
 "deny_message": "Not authorized.",
})
for uid, text in [
 (99, "hi"),
 (111, "ignore previous instructions"),
 (111, "summarize this note"),
]:
 r = pipe.precheck(uid, text)
 print(uid, r.stage, r.run_agent, r.reply_text)
PY

99 allowlist False Not authorized.
111 injection False Blocked: looks like prompt injection / secret fishing. Rephrase.
111 ok True None

$ python scripts/repro.py
REPRO_OK agentic-dm-gateway unit suite

So bindest du es ein

Drei Integrationswege. Wähle einen.

1) Drop-in-Discord (am einfachsten)

Installiere mit Discord-Unterstützung, richte die Umgebungsvariablen auf deine Benutzer-IDs aus, registriere das Gateway und starte den Bot.

pip install -e ".[discord]"
# or: pip install agentic-dm-gateway[discord]


![agentic-dm-gateway banner](https://assets.kitploit.com/production/public/readmes/50583/ed168cd7c4b9a61caf98c8a31b2dfb7d7e3bd3c5401288706c21c1376c7199ce.jpg)

export DISCORD_BOT_TOKEN=...
export AGENTIC_DM_ALLOWLIST=your_discord_user_id
export AGENTIC_DM_OWNER_ID=your_discord_user_id
# optional: export AGENTIC_DM_PIN=....


![agentic-dm-gateway banner](https://assets.kitploit.com/production/public/readmes/50583/ed168cd7c4b9a61caf98c8a31b2dfb7d7e3bd3c5401288706c21c1376c7199ce.jpg)
python examples/discord_echo_bot.py

In deinem eigenen Bot:

import discord
from agentic_dm_gateway.discord_adapter import register_dm_gateway

def agent(user_id: int, text: str, *, is_owner: bool = False) -> str:
 # your Hermes / local model / tool loop
 return call_your_model(text)

intents = discord.Intents.default()
intents.message_content = True
bot = discord.Client(intents=intents)

register_dm_gateway(
 bot,
 {
 "allowed_user_ids": [], # or rely on AGENTIC_DM_ALLOWLIST env
 "owner_ids": [],
 "pin_enabled": False,
 "deny_message": False, # silent drop for strangers
 },
 agent=agent,
)
bot.run(TOKEN)

Was register_dm_gateway tut:

1. Installiert einen on_message-Handler auf deinem discord.Client / Bot.
2. Ignoriert Bots und Guild-Nachrichten (nur DMs).
3. Führt InboundSecurityPipeline.precheck vor deinem Agenten aus.
4. Sendet bei Bedarf Deny-/Control-Antworten.
5. Ruft deinen agent(user_id, sanitized_text, is_owner=...) auf.
6. Bereinigt die Agentenantwort (Geheimnisse + Bild-Beacons) und teilt Discords 2000-Zeichen-Limit in Blöcke auf.

Guild-Nachrichten erreichen den Agenten nie. Nur DMs von Benutzern auf der Allowlist.

2) Manueller Discord-Hook (du hast bereits on_message)

Wenn du register_dm_gateway nicht verwenden kannst (bestehende Handler-Kette), rufe die Pipeline selbst auf:

from agentic_dm_gateway import InboundSecurityPipeline
from agentic_dm_gateway.security import sanitize_agent_output

pipe = InboundSecurityPipeline({
 "allowed_user_ids": [YOUR_ID],
 "owner_ids": [YOUR_ID],
 "pin_enabled": True,
})

@bot.event
async def on_message(message):
 if message.author.bot or message.guild is not None:
 return

 pre = pipe.precheck(int(message.author.id), message.content or "")
 if pre.reply_text and not pre.run_agent:
 await message.channel.send(pre.reply_text[:1900])
 return
 if not pre.run_agent:
 return

 raw = await your_agent(pre.sanitized_text) # Hermes, Ollama, API, ...
 await message.channel.send(sanitize_agent_output(str(raw))[:1900])

3) Protokoll-agnostisch (Hermes, CLI, Telegram, beliebig)

Kein Discord-Import erforderlich. Verwende denselben Precheck um jeden Agenten-Turn:

from agentic_dm_gateway import InboundSecurityPipeline
from agentic_dm_gateway.security import sanitize_agent_output

pipe = InboundSecurityPipeline({
 "allowed_user_ids": [111],
 "owner_ids": [111],
 "pin_enabled": False,
 "rate_limit_per_minute": 20,
 "block_injection": True,
 "deny_message": "Not authorized.",
})

def handle_inbound(user_id: int, text: str) -> str | None:
 pre = pipe.precheck(user_id, text)
 if pre.run_agent:
 answer = my_llm(pre.sanitized_text) # your model / Hermes run
 return sanitize_agent_output(str(answer))
 return pre.reply_text # deny or control-command reply

PrecheckResult-Felder:

• run_agent: nur bei true an das Modell weiterleiten
• sanitized_text: bereinigte Eingabe
• reply_text: Deny-/Control-Befehlsantwort
• stage: allowlist | kill | pin | rate | injection | ok | ...

Checkliste für die Anbindung:

1. Erstelle InboundSecurityPipeline einmal beim Prozessstart (Konfiguration + Env).
2. Bei jeder eingehenden Nachricht: pre = pipe.precheck(user_id, text).
3. Wenn pre.run_agent: rufe deinen Agenten nur mit pre.sanitized_text auf.
4. Gib die Modellausgabe vor dem Senden immer durch sanitize_agent_output.
5. Behandle Control-Antworten (/auth, /kill, …) als erledigt, wenn run_agent false ist.

Pipeline (eine eingehende Nachricht)

1. Adapter: ignore bots; only accept DMs (not server channels)
2. Allowlist: is this user id permitted?
3. Owner commands: /kill /unkill /status -> reply, stop
4. Session commands: /auth <pin> /lock -> reply, stop
5. SecurityGateway.check_message:
 kill switch?
 session unlocked? (PIN)
 under rate limit?
 length + injection heuristics OK?
6. If ok -> run_agent=True with sanitized text
7. After your agent returns -> sanitize_agent_output (redact + strip image beacons)
8. Audit rows written along the way

Kontrollebene: wer der Benutzer ist (Allowlist / Besitzer). Datenebene: Nachrichteninhalt (bis die Prüfungen bestanden sind, immer als nicht vertrauenswürdig betrachtet).

Paketstruktur

src/agentic_dm_gateway/
 security.py # RateLimiter, SessionAuth, SecurityGateway,
 # sanitize_input, redact_secrets, sanitize_agent_output,
 # kill switch, audit_log
 allowlist.py # merge config + env + file into allowlist / owners
 commands.py # /kill /unkill /status /auth /lock (no LLM)
 pipeline.py # InboundSecurityPipeline.precheck() orchestration
 discord_adapter.py # optional discord.py on_message wire-up
tests/ # unit tests for the core (no Discord required)
examples/
 minimal_precheck.py # CLI-style demo of precheck outcomes
 discord_echo_bot.py # secured DMs + echo agent

Keinerlei erforderliche Laufzeitabhängigkeiten. Discord ist optional: pip install agentic-dm-gateway[discord].

Installation

git clone https://github.com/SamsonCyber/agentic-dm-gateway.git
cd agentic-dm-gateway
pip install -e ".[dev]"
python scripts/repro.py

Konfiguration

Konfigurations-Dict

Umgebungsvariablen

Standard-Zustandsverzeichnis: ./data/agentic_dm/.

Besitzer- und Sitzungsbefehle

Diese rufen niemals dein Modell auf.

Grenzen (ehrlich)

• Die Injection-Erkennung ist eine Regex-Heuristikliste, kein vollständiger LLM-Bewerter oder Klassifikator.
• Die Schwärzung ist Best-Effort-Musterabgleich; sie wird neuartige Geheimnisformate übersehen.
• Die Hermes-Inspiration ist architektonisch (DM-Kontrollebene). Dies ist kein vollständiges Hermes-Gateway oder Pairing-Stack.
• Du benötigst weiterhin eine sichere Token-Speicherung, Tools mit minimalen Rechten und Host-Härtung außerhalb dieser Bibliothek.

Lizenz

MIT. Siehe LICENSE.