Agentic DM Gateway

Plano de controle de segurança para agentes de LLM em chats privados (normalmente DMs do Discord).

Ele fica na frente do seu agente. Decide quem pode conversar, se a sessão está desbloqueada, se o processo está pausado e se esta mensagem é segura o suficiente para ser encaminhada. Seu modelo e suas ferramentas permanecem atrás desse portão. A biblioteca não chama um LLM. Ela não implementa funcionalidades de produto além de segurança.

Inspirado no Hermes. O design segue as mesmas ideias de control plane usadas nos gateways de mensagens do Hermes Agent: entrega priorizando DM, allowlists de identidade, abertura no estilo pareamento, kill switch do proprietário e uma separação rígida entre quem pode agir (control plane) e o texto da mensagem que o modelo vê (data plane). Este pacote é uma extração pequena e independente desse padrão para qualquer agente que possa ser chamado. Não é afiliado à Nous Research.

Maturidade: implementado · validado de forma independente · mantido. Consulte STATUS.md. Reproduzir: python scripts/repro.py (espera REPRO_OK).

Testes offline:

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

Ao vivo: https://github.com/SamsonCyber/agentic-dm-gateway

O problema

Se você colocar um agente no Discord (ou em qualquer API de chat) com ferramentas, qualquer pessoa que consiga enviar mensagem ao bot pode tentar:

• usar o agente sem permissão
• queimar a cota da API com enxurradas de mensagens
• injetar prompts no estilo "ignore as instruções anteriores"
• enganar o modelo para repetir chaves de API ou outros segredos

Você precisa de um control plane (controles de identidade e de processo) separado do data plane (o texto da mensagem que o modelo vê).

Este pacote é esse control plane.

O que ele faz

Escopo: somente gate de segurança. Não é um chatbot, bot de trading, scanner ou framework de agentes. Passe um agent(user_id, text) -> str (ou assíncrono) se você usar Discord. O núcleo funciona com qualquer ID de usuário inteiro e texto simples.

Demonstração (copiar e colar)

$ 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

Como integrar

Três caminhos de integração. Escolha um.

1) Integração direta com Discord (mais fácil)

Instale com suporte ao Discord, aponte as variáveis de ambiente para seus IDs de usuário, registre o gateway e execute o 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

No seu próprio 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)

O que register_dm_gateway faz:

1. Instala um handler on_message no seu discord.Client / bot.
2. Ignora bots e mensagens de servidores (apenas DMs).
3. Executa InboundSecurityPipeline.precheck antes do seu agente.
4. Envia respostas de negação / controle quando necessário.
5. Chama seu agent(user_id, sanitized_text, is_owner=...).
6. Limpa a resposta do agente (segredos + beacons de imagem) e divide respeitando o limite de 2000 caracteres do Discord.

Mensagens de servidores nunca chegam ao agente. Somente DMs de usuários na allowlist chegam.

2) Hook manual no Discord (você já tem on_message)

Se você não puder usar register_dm_gateway (cadeia de handlers existente), chame o pipeline você mesmo:

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) Independente de protocolo (Hermes, CLI, Telegram, qualquer coisa)

Nenhum import do Discord é necessário. Use o mesmo precheck em qualquer turno do agente:

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

Campos de PrecheckResult:

• run_agent: encaminhar ao modelo somente se verdadeiro
• sanitized_text: entrada limpa
• reply_text: resposta de negação / comando de controle
• stage: allowlist | kill | pin | rate | injection | ok | ...

Checklist de integração:

1. Construa InboundSecurityPipeline uma vez no início do processo (config + env).
2. Para cada mensagem recebida: pre = pipe.precheck(user_id, text).
3. Se pre.run_agent: chame seu agente somente com pre.sanitized_text.
4. Sempre passe a saída do modelo por sanitize_agent_output antes de enviar.
5. Trate as respostas de controle (/auth, /kill, …) como concluídas quando run_agent for false.

Pipeline (uma mensagem recebida)

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

Control plane: quem é o usuário (allowlist / proprietário). Data plane: corpo da mensagem (sempre não confiável até que as verificações passem).

Estrutura do pacote

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

Zero dependências de runtime obrigatórias. Discord é opcional: pip install agentic-dm-gateway[discord].

Instalação

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

Configuração

Dicionário de configuração

Variáveis de ambiente

Diretório de estado padrão: ./data/agentic_dm/.

Comandos do proprietário e da sessão

Eles nunca chamam seu modelo.

Limitações (honestas)

• A detecção de injeção é uma lista heurística de regex, não um juiz ou classificador LLM completo.
• A redação é uma correspondência de padrões best-effort; ela pode não detectar formatos novos de segredos.
• A inspiração no Hermes é arquitetural (control plane de DM). Isso não é um gateway Hermes completo ou uma stack de pareamento.
• Você ainda precisa de armazenamento seguro de tokens, ferramentas com privilégio mínimo e endurecimento do host fora desta biblioteca.

Licença

MIT. Consulte LICENSE.