Argonaut

Servidor MCP (stdio) em Python que deixa um agente descobrir, inspecionar e diagnosticar filas e tópicos em Redis Streams, Kafka, RabbitMQ, Amazon SQS e Amazon SNS. As operações de escrita existem para resolver o problema encontrado, sempre atrás de modo, allowlist e confirmação em duas etapas.

Faz par com o Nautilus (MCP de bancos de dados): Nautilus para bases, Argonaut para mensageria.

O que faz

O caso de uso central é diagnóstico: "por que essa fila está crescendo?", "qual consumer group está atrasado?", "o que tem na DLQ?", "por que a mensagem publicada no SNS não chegou na fila X?".

Instalação e execução

Requer Python 3.11+. Dependências com uv.

uv sync --extra all          # ou --extra redis / kafka / rabbitmq / aws
ARGONAUT_CONFIG=./argonaut.yaml uv run argonaut

O servidor fala por stdio. Configure-o no seu cliente MCP apontando o comando para uv run argonaut (ou o console script argonaut) com a variável de ambiente ARGONAUT_CONFIG e os segredos que o YAML referencia.

Configuração

Um único arquivo YAML descreve defaults e conexões. Segredos sempre por ${VAR} (resolvidos do ambiente) — nunca valores no arquivo. Veja argonaut.example.yaml. O servidor valida na inicialização e falha cedo, com mensagens claras.

defaults:
  peek: { max_messages: 20, max_payload_bytes: 4096 }
  confirm_token_ttl: 120s
  audit_log: ./argonaut-audit.jsonl
  redact:
    json_paths: ["$.customer.document", "$.card.*"]
    patterns: [cpf, email, credit_card]

connections:
  redis-dev:
    type: redis_streams
    url: "${REDIS_URL}"
    mode: admin
    dlq_convention: "{name}:dlq"

Tipos de conexão: redis_streams, kafka, rabbitmq, aws (SQS + SNS juntos).

Tools

¹ peek_messages com accept_side_effects=true (leitura que altera estado, como em SQS e filas clássicas do RabbitMQ) exige modo operator.

Além das tools, o servidor expõe o resource argonaut://{conexão}/{destino} (a descrição do destino em JSON) e prompts de diagnóstico prontos (investigar_backlog, analisar_dlq, rastrear_mensagem_sns).

Segurança

O padrão é readonly. Toda escrita exige três coisas:

1. Modo por conexão (readonly ⊂ operator ⊂ admin): produção nasce readonly; promover é mudança de config, nunca de runtime.
2. Allowlist/denylist por glob: destino negado não aparece nem na listagem.
3. Confirmação em duas etapas: a 1ª chamada roda em dry-run e devolve o impacto ("vai apagar 48.213 mensagens de orders.dlq") e um token com TTL curto, vinculado ao hash dos argumentos; a execução só acontece na 2ª chamada com esse token.

Além disso: payloads truncados e com dados sensíveis redigidos antes de sair; toda MessageView marcada como untrusted (o conteúdo é dado, não instrução); auditoria JSON lines de toda escrita (inclusive dry-runs); credenciais só por ambiente, nunca em respostas, erros ou logs.

Desenvolvimento

uv sync --extra dev
uv run pytest tests/unit tests/contract      # sem Docker (fakes)
uv run pytest -m integration                 # brokers reais (Docker)

A peça central de teste é uma suíte de contrato parametrizada por adapter: o mesmo cenário (criar destino, publicar, stats, peek, lag, redrive, purge) roda em todos. Um broker novo só entra quando passa nela.

Fora de escopo na v1

Ser consumidor/worker de produção; gestão de usuários e ACLs dos brokers; provisionamento de clusters; monitoramento contínuo e alertas; decodificação de Avro/Protobuf no Kafka; replay de tópicos FIFO do SNS; transporte HTTP remoto.