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

$ 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

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

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)

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])

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

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

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

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