agentpost

A pluggable framework for AI agents to route and answer email.

agentpost is the reusable scaffolding under an agent email system. It does not hard-code a model, a prompt, or a transport — you plug in your own. Our own deployment plugs in the NVIDIA Nemotron API with our prompting layers, but the framework itself is brain-, prompt-, and transport-agnostic.

What it does

Two capabilities on one pluggable scaffold:

1. Route — an incoming message is classified by the brain against your routing prompt, and the correct target agent is woken through the pluggable Delivery layer. The proven, universal wake is tmux send-keys of a FIXED pointer string into the target's registered tmux session — the same mechanism a proven agent fleet runs on. Delivery stays pull-based / self-custody: routing decides where, the wake just tells the right agent to go look; the message body never travels through the wake.
2. Answer — reply to a message directly, backed by an FAQ section: when the same question recurs, the vetted answer is reused instead of re-asking the brain — faster, cheaper, consistent.

The recommended flow is answer_or_route (FAQ-FIRST): each message is answered first (marketing "not interested" and already-answered questions get a canned FAQ reply with no routing); only a genuine miss falls through to routing + a Delivery wake.

Pluggable everywhere

• Brain — any model behind one Brain interface (Nemotron, Claude, GPT, local…).
• Prompting — your own routing/answer prompt layers, supplied as config.
• Transport — the same core drops onto the agent-to-agent relay, Gmail, or anything else, behind one Transport interface.
• Scheduler — the wake mechanism (cron) is an interface too, so it's testable and swappable.

Reference implementations (an echo brain, an in-memory transport, a no-op scheduler) ship so the framework runs and tests standalone with zero external dependencies.

Install

pip install -e .          # core has NO third-party dependencies (stdlib only)
python -m unittest discover -s tests

Package layout

agentpost/
  message.py        Message / OutboundMessage — the transport-neutral envelope
  brains/           Brain ABC + injection defense + fail-safe JSON; EchoBrain
  transports/       Transport ABC; InMemoryTransport
  prompting/        PromptLayers (routing/answer templates) + default_templates/
  faq.py            FAQStore — the answer-reuse layer (find/add/suggestion_block)
  directory.py      AgentDirectory — targets + how to wake them (tmux_session)
  scheduler.py      Scheduler ABC; NoopScheduler, CronScheduler, TimerScheduler
  delivery/         Delivery ABC; TmuxSendKeysDelivery (send-keys), NoopDelivery
  router.py         Router — brain.route → resolve target → Delivery wake
  answerer.py       Answerer — FAQ-first, brain on miss, save new answers back
  pipeline.py       Pipeline — run_once() in route|answer|answer_or_route mode
  config.py         build_pipeline(dict) + register_brain/transport/scheduler/delivery

The interfaces you plug into

Interface	Contract	Reference impl
Brain	route(msg, ctx) -> RouteDecision and/or answer(msg, ctx) -> AnswerResult|None. Implement one or both.	EchoBrain
Transport	name, fetch_new() -> [Message], send(OutboundMessage) -> id, mark_handled(id)	InMemoryTransport
Scheduler	schedule_wake(target, delay_seconds=60, prompt) -> handle, cancel(handle)	NoopScheduler, CronScheduler
Delivery	verify(target, session) -> bool, deliver(target, session, pointer) -> DeliveryResult	TmuxSendKeysDelivery, NoopDelivery
PromptLayers	render_routing(ctx), render_answer(ctx) — your own template strings	shipped defaults

Delivering a wake by hand (verify a session, then send the fixed pointer):

python3 -m agentpost.delivery.tmux --verify agent-billing          # is it up?
python3 -m agentpost.delivery.tmux agent-billing "[agentpost] You have new mail waiting -- check your inbox."

Every brain inherits the injection defense for free: build your model prompt via agentpost.build_prompt(instructions, message, context_block, task) (wraps the untrusted message in a per-message random nonce delimiter + control-token sanitizer) and parse the model's reply with parse_route / parse_answer (fail-safe: garbage raises BrainError, never a fabricated result).

Plug in your own brain

from agentpost import Brain, RouteDecision, AnswerResult, build_prompt, parse_answer
from agentpost import register_brain

class MyBrain(Brain):
    name = "mybrain"

    def answer(self, message, context):
        prompt = build_prompt(
            instructions=context.get("prompt", ""),   # your rendered template (trusted)
            message=message,                           # untrusted — auto-nonce-wrapped
            context_block=context.get("suggestion", ""),  # FAQ hint (trusted, still scrubbed)
            task='Respond with JSON: {"reply": "...", "reason": "..."}',
        )
        raw = my_model_call(prompt)          # <-- your provider (Nemotron, Claude, …)
        return parse_answer(raw)             # None = decline; BrainError on garbage

    # (implement route() too, or leave it raising NotImplementedError)

# make it selectable from a config dict by string name:
register_brain("mybrain", MyBrain)

A transport is the same shape — subclass Transport, then register_transport("relay", RelayTransport).

Wire a whole pipeline from a config dict

from agentpost import build_pipeline

pipe = build_pipeline({
    "mode": "answer",                                  # "route" | "answer"
    "brain":     {"type": "echo", "answer_template": "Ack: {subject}"},
    "transport": {"type": "memory"},
    "scheduler": {"type": "noop"},                     # or "cron"
    "prompting": {"dir": "/path/to/templates"},        # or routing_template/answer_template
    "faq":       {"path": "~/.agentpost/faq.json", "threshold": 0.35},
    "directory": {"path": "~/.agentpost/directory.json",
                  "targets": {"billing": {"wake_command": "wake billing",
                                          "description": "invoices and payments"}}},
    "router":    {"fallback_target": "human", "min_confidence": 0.4},
    "answerer":  {"faq_threshold": 0.35, "save_new_answers": True},
})

run = pipe.run_once()   # fetch new mail → route or answer each → mark handled

type names resolve through the registries; the reference impls are registered by default, and external adapters add themselves with register_brain / register_transport / register_scheduler — no edit to the core.

Status

v0.1 — core framework + reference implementations, plus a full stdlib test suite (python -m unittest discover -s tests). The example deployment (examples/our_instance/) plugs in NVIDIA Nemotron + relay/Gmail transports as separate adapter packages that register themselves.