aquifer

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jun 6, 2026 License: MIT Imports: 27 Imported by: 0

README

Aquifer — MCP Traffic Framework

Self-hosted MCP server framework for coordinating HTTP traffic from distributed agents. Aquifer queues requests durably, controls dispatch pace, and exposes the same traffic core through pluggable adapters.


The problem

Distributed agents call tools and APIs in bursts. Your backend gets overwhelmed on inbound. Your app gets 429s on outbound. One slow dependency takes everything else down with it.

Aquifer gives those agents a coordination layer. It absorbs the burst, queues requests durably to SQLite, and releases them at the rate you configure. Your backend decides the pace. The upstream decides the pace. Whoever needs to slow things down — wins.


Two ways to use it

MCP tools — coordinate distributed agents

agents / MCP clients  →  aquifer_enqueue_job  →  Aquifer queue  →  target API

Agents call Aquifer as an MCP server instead of racing each other directly against the same backend or external API. Aquifer returns a job id immediately, dispatches the request at a controlled rate, and delivers the result to your webhook.

HTTP API — protect your API

agents / clients  →  POST /jobs to Aquifer  →  your backend (at controlled RPS)

Agents hammering your API over HTTP? Aquifer queues their requests and drains them to your backend at a pace it can handle. Your backend returns X-Aquifer-Rps headers to signal how fast it wants traffic in real time.

Outbound — respect external APIs

your app  →  POST /jobs to Aquifer  →  OpenAI / Stripe / any API (at controlled RPS)

Calling a rate-limited upstream? Aquifer queues the calls and dispatches them at your configured rate. If the upstream signals a slowdown via headers, Aquifer backs off automatically.

In both cases — the upstream response headers are the final say on pace. Your config sets the ceiling. Headers can only reduce below it, never exceed it. When pressure clears, the rate recovers gradually back to your ceiling.


How it works

  1. Client submits a job through an adapter (MCP tool or HTTP endpoint) and moves on
  2. Aquifer persists it to SQLite — survives crashes, re-dispatches on restart
  3. A per-upstream worker dispatches at your configured RPS with jitter
  4. On completion Aquifer POSTs your webhook with the response body and status
  5. The upstream can adjust the rate live via X-Aquifer-* response headers

Quick start

Binary

go install github.com/rjpruitt16/aquifer/cmd/aquifer@latest
aquifer

Docker

docker run -p 8080:8080 -v $(pwd)/data:/data \
  -e AQUIFER_ADAPTER=http \
  -e DB_PATH=/data/aquifer.db \
  ghcr.io/rjpruitt16/aquifer

Fly.io

git clone https://github.com/rjpruitt16/aquifer
cd aquifer
flyctl launch --name my-aquifer --no-deploy
flyctl volumes create aquifer_data --size 1 --region iad
flyctl deploy

Configuration

Set CONFIG_PATH to a YAML file to configure rate limits per upstream hostname:

# aquifer.yml — copy from aquifer.example.yml
defaults:
  rps: 2
  max_concurrent: 1

upstreams:
  api.openai.com:
    rps: 10
    max_concurrent: 3
  api.stripe.com:
    rps: 20
    max_concurrent: 5
  your-backend.internal:
    rps: 50
    max_concurrent: 10
Env var Default Description
AQUIFER_ADAPTER http for binary, mcp-stdio in Docker image Runtime adapter: http or mcp-stdio
PORT 8080 HTTP listen port
DB_PATH aquifer.db SQLite database path
CONFIG_PATH (none) Path to rate limit config YAML

Framework adapters

Aquifer has a framework-neutral core and adapter front doors. The core owns idempotency, persistence, rate control, dispatch, SSE events, L8 signing, and webhook delivery. Adapters translate framework-specific calls into that core.

type FrameworkAdapter interface {
    Name() string
    Start(ctx context.Context, aquifer *Aquifer) error
}

Current adapters:

Adapter Env Purpose
HTTP AQUIFER_ADAPTER=http Existing REST/SSE API on PORT
MCP stdio AQUIFER_ADAPTER=mcp-stdio MCP server exposing Aquifer tools over stdio

Run as an MCP stdio server:

AQUIFER_ADAPTER=mcp-stdio aquifer

The published Docker image defaults to AQUIFER_ADAPTER=mcp-stdio so MCP directories such as Glama can start and introspect it directly. Set AQUIFER_ADAPTER=http when running Aquifer as an HTTP queue service.

MCP tools:

Tool Purpose
aquifer_enqueue_job Queue an HTTP request for durable, rate-controlled dispatch
aquifer_get_job Fetch job status and metadata
aquifer_health Return health and protocol metadata
aquifer_l8_metadata Return L8 public key metadata
aquifer_l8_challenge Answer an L8 challenge

MCP resources:

Resource Purpose
aquifer://jobs/{job_id} Read current job status and metadata as JSON

The HTTP adapter remains the default so existing deployments do not change.

Writing an adapter

Adapter authors import Aquifer as a Go package, implement FrameworkAdapter, and pass the shared core into their framework. Built-in adapters are selected with AQUIFER_ADAPTER; third-party adapters normally ship as small custom binaries that call aquifer.RunAdapter.

package myframework

import (
    "context"

    "github.com/rjpruitt16/aquifer"
)

type Adapter struct{}

func (a *Adapter) Name() string {
    return "my-mcp-framework"
}

func (a *Adapter) Start(ctx context.Context, app *aquifer.Aquifer) error {
    // Register framework handlers that call:
    // app.Enqueue(req)
    // app.GetJob(jobID)
    // app.SubscribeJob(jobID)
    // app.Health()
    return nil
}

Custom binaries can reuse Aquifer's runtime wiring:

package main

import (
    "context"
    "log"

    "github.com/rjpruitt16/aquifer"
    myadapter "github.com/you/your-adapter"
)

func main() {
    runtime := aquifer.NewRuntime(aquifer.RuntimeOptions{
        DBPath:     "aquifer.db",
        ConfigPath: "aquifer.yml",
    })
    runtime.RecoverQueuedJobs("aquifer.db")

    adapter := myadapter.New()
    log.Fatal(adapter.Start(context.Background(), runtime.Aquifer))
}

For the shortest form, let Aquifer create the runtime and start your adapter:

adapter := myadapter.New()
log.Fatal(aquifer.RunAdapter(context.Background(), adapter, aquifer.RuntimeOptions{
    DBPath:     "aquifer.db",
    ConfigPath: "aquifer.yml",
}))

See examples/custom_adapter for a complete compile-tested adapter binary.


Metrics adapter

Aquifer emits lifecycle events through a pluggable metrics adapter. Implement MetricsAdapter and pass it into NewRegistry:

type MetricsAdapter interface {
    JobQueued(userID, upstream string)
    JobDispatched(userID, upstream string)
    JobCompleted(userID, upstream string, durationMs int64)
    JobFailed(userID, upstream string, reason string)
    WebhookDelivered(url string, attempt int)
    WebhookFailed(url string, attempts int)
    QueueDepth(upstream string, depth int)
    FlowRate(upstream string, rps float64)
}

Aquifer ships with NoopMetricsAdapter, so existing deployments do not change.


API

POST /jobs
{
  "user_id":        "user-123",
  "idempotent_key": "invoice-42-notify",
  "url":            "https://api.openai.com/v1/chat/completions",
  "method":         "POST",
  "headers":        { "Authorization": "Bearer sk-..." },
  "body":           "{\"model\":\"gpt-4o\",\"messages\":[...]}",
  "webhook_url":    "https://yourapp.com/webhooks/aquifer"
}

Idempotent — duplicate idempotent_key per user_id returns the existing job.

201 new job queued · 200 + "duplicate": true already exists

GET /jobs/:id
{
  "job_id":     "a3f9...",
  "status":     "queued | in_flight | completed | failed",
  "url":        "https://api.openai.com/v1/chat/completions",
  "method":     "POST",
  "created_at": 1715000000000
}
GET /jobs/:id/stream

Server-Sent Events stream for live job updates.

event: queued
data: {"job_id":"a3f9...","status":"queued"}

event: dispatching
data: {"job_id":"a3f9..."}

event: completed
data: {"job_id":"a3f9...","response_status":200,"body":"..."}

Or event: failed with {"job_id":"...","reason":"..."}.

Position updates — while the job waits in queue, a position event is broadcast every 2 seconds:

event: position
data: {"job_id":"a3f9...","position":4}
curl -N http://localhost:8080/jobs/<id>/stream

Connecting late is safe — you'll receive synthetic queued and dispatching catchup events for states you missed.

The Aqueduct Protocol — SSE is the live view. Webhook is the guaranteed delivery. Both always fire regardless of whether the stream was open. Think of it like a phone call with voicemail: stay on the line (SSE) for real-time updates, or hang up and the result goes to voicemail (webhook). You never lose the result.

GET /health
{ "status": "ok" }

Webhook payload

Completed

{
  "job_id":          "a3f9...",
  "status":          "completed",
  "response_status": 200,
  "body":            "..."
}

Failed (after 4 retries with exponential backoff)

{
  "job_id": "a3f9...",
  "status": "failed",
  "reason": "connection refused"
}

Webhook delivery retries 4 times: 1 s · 2 s · 4 s · 8 s.


L8 Protocol — trustless webhook delivery

Traditional webhook security requires sharing a secret between sender and receiver and storing it in a database on both sides. Aquifer implements L8 v0.1, a lightweight challenge-response protocol that eliminates shared secrets entirely.

The attack surface problem L8 solves: A shared HMAC secret is something that can be stolen, accidentally logged, forgotten to rotate, or compromised on either side. A stolen secret lets anyone forge webhook deliveries forever. L8 replaces that shared secret with public key cryptography — there is no secret to steal from a database.

How it works:

  1. The receiver publishes a public key at GET /.well-known/l8
  2. Before the first delivery, Aquifer challenges the receiver to prove ownership of the corresponding private key — a one-time handshake
  3. Trust is cached to disk as l8-trust/{domain}.json — the handshake never runs again for that domain
  4. Every webhook delivery carries X-L8-Signature headers the receiver verifies locally with no database lookup and no round-trip to any authority

Why this keeps things fast: Verification is a single local Ed25519 verify() call against a cached public key. No database query, no HTTP call, no shared state. Microseconds.

Key management:

Set L8_PRIVATE_KEY (base64 Ed25519 private key) for a stable identity across restarts. Without it, Aquifer auto-generates a key and saves it to .l8-key on first start.

To revoke trust with a domain: delete l8-trust/{domain}.json. The handshake re-runs on next delivery.

Aquifer exposes:

Endpoint Purpose
GET /.well-known/l8 Aquifer's public key and capabilities — receivers discover Aquifer here
POST /l8/challenge Handles incoming challenges from receivers verifying Aquifer's identity
GET /l8-spec The full L8 protocol spec — served on any running Aquifer instance

Protocol version: 0.1. The version is advertised in /.well-known/l8 and GET /health so agents can detect what capabilities are available. Future versions will add payload encryption (0.2) and formalized key rotation (0.3).

The full protocol spec and verification examples are in L8-SPEC.md, also browsable at GET /l8-spec on any running instance. The spec documents the receiver-side endpoints any service needs to implement to receive signed webhooks.

See tests/l8_receiver.py for a complete reference implementation of the receiver side, and tests/test_l8.py for end-to-end tests that verify the handshake, signed delivery, and cryptographic signature validation.


Dynamic rate control

The upstream controls pace at runtime via response headers:

Header Effect
X-Aquifer-Rps Reduce dispatch rate to this value
X-Aquifer-Max-Concurrent Reduce max in-flight requests
X-Aquifer-Account-Queue enabled — isolate each tenant's queue

With X-Aquifer-Account-Queue: enabled, each (user_id, api_key) pair gets its own independently paced queue. One tenant's burst can't slow down another.


Autoscaling

Aquifer sends machine load data as headers on every outgoing request to your service:

Header Value
X-Aquifer-Total-Jobs Total jobs on this machine right now
X-Aquifer-Queue-Depth Jobs waiting to be dispatched
X-Aquifer-Flow-Rate Current dispatch rate (RPS) for this queue

Your service reads these headers and calls your autoscaler when the queue is growing:

total_jobs = int(request.headers.get("X-Aquifer-Total-Jobs", 0))

if total_jobs > 500:
    scale_up()  # call Fly.io, AWS ASG, k8s HPA, etc.

This keeps the autoscaling decision in your hands — Aquifer exposes the signal, your service acts on it however fits your infrastructure.


Reliability

  • Durable queue — jobs persist to SQLite on every write
  • Crash recovery — queued jobs re-dispatched automatically on restart
  • In-flight tracking — jobs marked in_flight before dispatch; recovered immediately on panic without waiting for full restart
  • Stale job safety net — in-flight jobs older than 5 min automatically reset to queued
  • Per-job panic isolation — a panic in one job marks it failed and delivers the webhook; the worker keeps running

Job TTLs

Status TTL
queued 24 h
completed 30 min
failed 2 h

Deployment model

Aquifer is designed as a sidecar on a single machine. One instance per app server, SQLite on a local persistent volume — no external database, no coordination overhead.

Running multiple instances against the same upstream without partitioning will multiply your request rate. If you scale horizontally, partition by upstream domain or tenant so each instance owns a distinct key space.


License

MIT

Documentation

Index

Constants

This section is empty.

Variables

View Source
var ErrJobNotFound = errors.New("job not found")

Functions

func RunAdapter

func RunAdapter(ctx context.Context, adapter FrameworkAdapter, opts RuntimeOptions) error

Types

type AccountQueue

type AccountQueue struct {
	// contains filtered or unexported fields
}

func NewAccountQueue

func NewAccountQueue(key, upstream string, rps float64, maxConc int, store *Store, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, onIdle func(string)) *AccountQueue

func (*AccountQueue) Enqueue

func (q *AccountQueue) Enqueue(job *Job)

func (*AccountQueue) RPS

func (q *AccountQueue) RPS() float64

type Aquifer

type Aquifer struct {
	// contains filtered or unexported fields
}

func NewAquifer

func NewAquifer(store *Store, registry *Registry, broker *Broker, l8 *L8Registry) *Aquifer

func (*Aquifer) Enqueue

func (a *Aquifer) Enqueue(req JobRequest) (EnqueueResult, error)

func (*Aquifer) GetJob

func (a *Aquifer) GetJob(id string) (*Job, error)

func (*Aquifer) HandleL8Challenge

func (a *Aquifer) HandleL8Challenge(req L8ChallengeReq) (*L8ChallengeResp, error)

func (*Aquifer) Health

func (a *Aquifer) Health() map[string]any

func (*Aquifer) L8Metadata

func (a *Aquifer) L8Metadata(host string) L8Meta

func (*Aquifer) SubscribeJob

func (a *Aquifer) SubscribeJob(id string) (*Job, <-chan SSEEvent, func(), error)

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

Broker is the pub/sub layer for SSE streams. Each job gets its own set of subscriber channels.

func NewBroker

func NewBroker() *Broker

func (*Broker) Publish

func (b *Broker) Publish(jobID string, event SSEEvent)

func (*Broker) Subscribe

func (b *Broker) Subscribe(jobID string) (<-chan SSEEvent, func())

type Config

type Config struct {
	Defaults  RateConfig            `yaml:"defaults"`
	Upstreams map[string]RateConfig `yaml:"upstreams"`
}

func LoadConfig

func LoadConfig(path string) *Config

func (*Config) ForURL

func (c *Config) ForURL(rawURL string) RateConfig

type EnqueueResult

type EnqueueResult struct {
	JobID     string `json:"job_id"`
	Status    Status `json:"status"`
	Duplicate bool   `json:"duplicate,omitempty"`
}

type FrameworkAdapter

type FrameworkAdapter interface {
	Name() string
	Start(ctx context.Context, aquifer *Aquifer) error
}

type HTTPAdapter

type HTTPAdapter struct {
	// contains filtered or unexported fields
}

func NewHTTPAdapter

func NewHTTPAdapter(addr string) *HTTPAdapter

func (*HTTPAdapter) Name

func (a *HTTPAdapter) Name() string

func (*HTTPAdapter) Start

func (a *HTTPAdapter) Start(ctx context.Context, aquifer *Aquifer) error

type Job

type Job struct {
	ID            string            `json:"id"`
	UserID        string            `json:"user_id"`
	IdempotentKey string            `json:"idempotent_key"`
	URL           string            `json:"url"`
	Method        string            `json:"method"`
	Headers       map[string]string `json:"headers,omitempty"`
	Body          string            `json:"body,omitempty"`
	WebhookURL    string            `json:"webhook_url"`
	Status        Status            `json:"status"`
	CreatedAt     int64             `json:"created_at"`
}

func NewJob

func NewJob(r *JobRequest) *Job

type JobRequest

type JobRequest struct {
	UserID        string            `json:"user_id"`
	IdempotentKey string            `json:"idempotent_key"`
	URL           string            `json:"url"`
	Method        string            `json:"method"`
	Headers       map[string]string `json:"headers,omitempty"`
	Body          string            `json:"body,omitempty"`
	WebhookURL    string            `json:"webhook_url"`
}

func (*JobRequest) Validate

func (r *JobRequest) Validate() string

type L8ChallengeReq

type L8ChallengeReq struct {
	ChallengeID     string `json:"challenge_id"`
	Nonce           string `json:"nonce"`
	Timestamp       int64  `json:"timestamp"`
	SenderPublicKey string `json:"sender_public_key"`
	Signature       string `json:"signature"`
}

type L8ChallengeResp

type L8ChallengeResp struct {
	ChallengeID       string `json:"challenge_id"`
	Nonce             string `json:"nonce"`
	ReceiverSignature string `json:"receiver_signature"`
	ReceiverPublicKey string `json:"receiver_public_key"`
}

type L8Meta

type L8Meta struct {
	ProtocolVersion   string   `json:"protocol_version"`
	ServiceName       string   `json:"service_name"`
	PublicKey         string   `json:"public_key"`
	ChallengeEndpoint string   `json:"challenge_endpoint"`
	SupportedAlgos    []string `json:"supported_algorithms"`
	Capabilities      []string `json:"capabilities"`
	SpecURL           string   `json:"spec_url"`
}

type L8Registry

type L8Registry struct {
	PubB64 string // exported so server can embed in responses
	// contains filtered or unexported fields
}

func NewL8Registry

func NewL8Registry(keyPath, trustDir string) *L8Registry

func (*L8Registry) EnsureTrust

func (r *L8Registry) EnsureTrust(webhookURL string)

EnsureTrust runs the L8 handshake with the webhook domain if not already trusted. Silently does nothing if the receiver doesn't support L8 — delivery still proceeds unsigned.

func (*L8Registry) HandleChallenge

func (r *L8Registry) HandleChallenge(req L8ChallengeReq) (*L8ChallengeResp, error)

HandleChallenge verifies the sender's ownership proof and returns Aquifer's signed response.

func (*L8Registry) IsTrusted

func (r *L8Registry) IsTrusted(webhookURL string) bool

IsTrusted returns true if the L8 handshake has been completed for this webhook domain.

func (*L8Registry) Meta

func (r *L8Registry) Meta(host string) L8Meta

func (*L8Registry) SignHeaders

func (r *L8Registry) SignHeaders(body []byte) map[string]string

SignHeaders returns X-L8-* headers to attach to an outgoing webhook delivery.

type MCPStdioAdapter

type MCPStdioAdapter struct {
	// contains filtered or unexported fields
}

func NewMCPStdioAdapter

func NewMCPStdioAdapter(in io.Reader, out io.Writer) *MCPStdioAdapter

func (*MCPStdioAdapter) Name

func (a *MCPStdioAdapter) Name() string

func (*MCPStdioAdapter) Start

func (a *MCPStdioAdapter) Start(ctx context.Context, aquifer *Aquifer) error

type MetricsAdapter

type MetricsAdapter interface {
	JobQueued(userID, upstream string)
	JobDispatched(userID, upstream string)
	JobCompleted(userID, upstream string, durationMs int64)
	JobFailed(userID, upstream string, reason string)
	WebhookDelivered(url string, attempt int)
	WebhookFailed(url string, attempts int)
	QueueDepth(upstream string, depth int)
	FlowRate(upstream string, rps float64)
}

type NoopMetricsAdapter

type NoopMetricsAdapter struct{}

func (NoopMetricsAdapter) FlowRate

func (NoopMetricsAdapter) FlowRate(upstream string, rps float64)

func (NoopMetricsAdapter) JobCompleted

func (NoopMetricsAdapter) JobCompleted(userID, upstream string, durationMs int64)

func (NoopMetricsAdapter) JobDispatched

func (NoopMetricsAdapter) JobDispatched(userID, upstream string)

func (NoopMetricsAdapter) JobFailed

func (NoopMetricsAdapter) JobFailed(userID, upstream string, reason string)

func (NoopMetricsAdapter) JobQueued

func (NoopMetricsAdapter) JobQueued(userID, upstream string)

func (NoopMetricsAdapter) QueueDepth

func (NoopMetricsAdapter) QueueDepth(upstream string, depth int)

func (NoopMetricsAdapter) WebhookDelivered

func (NoopMetricsAdapter) WebhookDelivered(url string, attempt int)

func (NoopMetricsAdapter) WebhookFailed

func (NoopMetricsAdapter) WebhookFailed(url string, attempts int)

type RateConfig

type RateConfig struct {
	RPS           float64 `yaml:"rps"`
	MaxConcurrent int     `yaml:"max_concurrent"`
}

type Registry

type Registry struct {
	// contains filtered or unexported fields
}

func NewRegistry

func NewRegistry(store *Store, cfg *Config, broker *Broker, l8 *L8Registry, metrics MetricsAdapter) *Registry

func (*Registry) Enqueue

func (r *Registry) Enqueue(job *Job)

func (*Registry) JobDispatched

func (r *Registry) JobDispatched()

func (*Registry) JobDone

func (r *Registry) JobDone()

type Runtime

type Runtime struct {
	Aquifer  *Aquifer
	Store    *Store
	Broker   *Broker
	Registry *Registry
	L8       *L8Registry
	Config   *Config
}

func NewRuntime

func NewRuntime(opts RuntimeOptions) *Runtime

func (*Runtime) DBPath

func (r *Runtime) DBPath() string

func (*Runtime) RecoverQueuedJobs

func (r *Runtime) RecoverQueuedJobs(dbPath string)

type RuntimeOptions

type RuntimeOptions struct {
	DBPath     string
	ConfigPath string
	Config     *Config
	L8KeyPath  string
	L8TrustDir string
	Metrics    MetricsAdapter
}

type SSEEvent

type SSEEvent struct {
	Event string
	Data  map[string]any
}

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer(aquifer *Aquifer) *Server

func (*Server) Routes

func (s *Server) Routes() http.Handler

type Status

type Status string
const (
	StatusQueued    Status = "queued"
	StatusInFlight  Status = "in_flight"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
)

type Store

type Store struct {
	// contains filtered or unexported fields
}

func NewStore

func NewStore(path string) *Store

func (*Store) CheckOrInsert

func (s *Store) CheckOrInsert(job *Job) (string, bool)

func (*Store) Counts

func (s *Store) Counts() StoreCounts

func (*Store) GetJob

func (s *Store) GetJob(jobID string) *Job

func (*Store) GetQueuedJobs

func (s *Store) GetQueuedJobs() []*Job

func (*Store) MarkInFlight

func (s *Store) MarkInFlight(jobID string)

func (*Store) Path

func (s *Store) Path() string

func (*Store) RecoverInFlight

func (s *Store) RecoverInFlight(queueKey string) []*Job

func (*Store) SetQueueKey

func (s *Store) SetQueueKey(jobID, queueKey string)

func (*Store) UpdateStatus

func (s *Store) UpdateStatus(jobID string, status Status)

type StoreCounts

type StoreCounts struct {
	TotalJobs  int64
	QueueDepth int64
}

type URLWorker

type URLWorker struct {
	// contains filtered or unexported fields
}

func NewURLWorker

func NewURLWorker(domain string, rps float64, maxConc int, store *Store, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, onIdle func(string)) *URLWorker

func (*URLWorker) Enqueue

func (w *URLWorker) Enqueue(job *Job)

Directories

Path Synopsis
cmd
aquifer command
examples
custom_adapter command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL