fluxpy

The problem

Every MCP server today is a vending machine. The agent asks, the server answers, the connection closes, and nothing is remembered. That works fine for "read this file" and badly for anything that is still happening.

If you want an agent to follow a live match, watch a deploy, track a price, or react to a failing build, request/response makes you choose between two bad options: poll in a loop and burn tokens on 200 identical answers, or ask once and miss everything that happens next.

Related MCP server: DEX Pools MCP

What fluxpy does

fluxpy holds persistent connections to live feeds, transforms them through a reactive operator pipeline, and keeps the results in a cursor-addressable buffer that survives between your turns. An agent can then consume a stream three different ways — and it needs all three, because no single one works everywhere:

flowchart LR
    A["SSE · WebSocket · HTTP poll<br/>files · processes · webhooks"] --> B[Source supervisor<br/><i>reconnect + backoff</i>]
    B --> C[Operator pipeline<br/><i>filter · window · throttle</i>]
    C --> D[(Ring buffer<br/>cursors)]
    D --> E[flux_poll / flux_wait]
    D --> F[Watches → alerts]
    D --> G[Sinks · recorder · SQLite]
    D --> H["subscriptions/listen<br/>push"]

It is not only a streaming server. It is a data plane: merge feeds into derived streams, compute rolling statistics, detect anomalies, set standing alerts, record a feed and replay it deterministically, and forward events onward to a file or a webhook. 28 tools in total.

And it is built for three quite different people:

Quickstart

Not on PyPI yet. Until the first release, install straight from this repository — the commands below already do. Once published, drop the --from git+... and plain uvx fluxpy serve works.

Check it runs at all:

uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy doctor
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy tail scoreboard -o seconds=0.3 -n 5

That second command streams a simulated football match — no network, no credentials. If events scroll past, fluxpy works and anything that goes wrong next is client configuration.

Now pick your assistant:

claude mcp add fluxpy -- uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy serve

Then, in a session, ask: "Subscribe to the scoreboard source with seconds: 0.5, wait for a goal, and tell me who scored."

Add to claude_desktop_config.json (where is it?):

{
  "mcpServers": {
    "fluxpy": {
      "command": "uvx",
      "args": ["--from", "git+https://github.com/saudaljuaid/fluxpy", "fluxpy", "serve"]
    }
  }
}

Then fully quit and reopen Claude Desktop — closing the window is not enough.

Add to ~/.cursor/mcp.json:

{
  "mcpServers": {
    "fluxpy": {
      "command": "uvx",
      "args": ["--from", "git+https://github.com/saudaljuaid/fluxpy", "fluxpy", "serve"]
    }
  }
}

Add to .vscode/mcp.json — note the key is servers, not mcpServers:

{
  "servers": {
    "fluxpy": {
      "type": "stdio",
      "command": "uvx",
      "args": ["--from", "git+https://github.com/saudaljuaid/fluxpy", "fluxpy", "serve"]
    }
  }
}

codex mcp add fluxpy -- uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy serve

ChatGPT connects over HTTP only, so run fluxpy as a server and expose it:

uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy serve --transport http --port 8765

Then add the URL under Settings → Connectors → Advanced → Developer mode. See docs/clients/chatgpt.md — read the authentication section before exposing it publicly.

Using something else? fluxpy ships configuration for 24 AI tools — Cline, Roo Code, Kilo Code, Continue, Gemini CLI, Amazon Q, Goose, opencode, Zed, Windsurf, Trae, JetBrains, Visual Studio, Warp, LibreChat, Cherry Studio, BoltAI, Witsy and more. See the client setup index, or let fluxpy do it:

# Add --from-git to any of these while fluxpy is unreleased.
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy install
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy install cline --from-git
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy install cline --from-git --write

Verify it works — ask your agent:

Subscribe to the scoreboard source with seconds: 0.5, then wait for a goal and tell me who scored.

That needs no network and no credentials. If goals arrive, the whole path works.

Presets

Not everyone wants to write a source spec, and nobody wants to look up the USGS GeoJSON schema to find out whether an earthquake happened. A preset is a complete, working stream behind a name and a sentence — with the filtering that makes the feed usable already in place.

uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy presets

An agent reaches these through flux_list_presets and flux_use_preset, so a request phrased in ordinary words — "watch Hacker News for anything about AI" — becomes one tool call rather than a guess at a spec.

A preset builds an ordinary StreamSpec. Nothing about the resulting stream is special: inspect it, edit it, export it with flux_export_config, paste it into a config file. fluxpy presets <name> prints exactly that block, ready to paste.

What it looks like

An agent monitoring live earthquakes, doing all the filtering server-side:

// flux_subscribe
{
  "stream_id": "quakes",
  "source": {
    "type": "http_poll",
    "url": "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson",
    "interval": 60,
    "select": "features"
  },
  "pipeline": [
    { "op": "flatten" },                                       // one event per quake
    { "op": "dedupe", "key": "id", "ttl": 3600 },              // the feed repeats itself
    { "op": "filter", "where": "properties.mag >= 4.5" },      // only significant ones
    { "op": "select", "fields": ["properties.place", "properties.mag", "properties.time"] }
  ]
}

That pipeline turns a 2 MB polled document into a handful of small events. The agent then either polls:

// flux_poll → { "events": [...], "cursor": 47, "missed": 0, "lag": 0 }

…or blocks until something big happens:

// flux_wait  { "stream_id": "quakes", "where": "properties.mag >= 6", "timeout": 300 }

…or sets a standing alert and gets on with something else:

// flux_watch { "stream_id": "quakes", "name": "Major quake",
//              "where": "properties.mag >= 6",
//              "message": "M{properties.mag} near {properties.place}" }

Sources

Full options in docs/sources.md, or run fluxpy sources.

Operators

filter · reject · map · select · enrich · flatten · distinct · dedupe · throttle · debounce · sample · delay · take · skip · scan · window · buffer · rate_limit · tag

Operators run server-side, before events reach the agent, which is the single biggest lever on both noise and token cost. throttle vs debounce vs sample are three different answers to "too much data" and docs/operators.md explains exactly when each is right.

Try it without an agent

# alias flux='uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy'
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy tail scoreboard -o seconds=0.3 --where "kind == 'goal'" -n 5
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy tail system_metrics -o seconds=1 --select cpu_percent,mem_used_percent -n 5
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy sources    # the full catalogue
uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy doctor     # diagnose a broken setup

fluxpy tail uses the same engine, sources, and operators as the MCP server, so it answers "is the feed broken, or is my client config broken?" in one command.

Configuration

fluxpy runs with zero configuration. A config file makes streams permanent:

streams:
  - id: wikipedia
    source:
      type: sse
      url: https://stream.wikimedia.org/v2/stream/recentchange
    pipeline:
      - op: filter
        where: wiki == 'enwiki' and namespace == 0 and bot == false
      - op: select
        fields: [title, user, comment]
      - op: throttle
        seconds: 1

watches:
  - id: cpu-hot
    stream_id: host
    name: CPU saturated
    where: cpu_percent > 90
    cooldown: 300

fluxpy init writes a commented starter. Streams an agent builds interactively can be exported with flux_export_config and pasted straight back in — the formats are identical by design.

Security

fluxpy connects wherever an agent tells it to, so it ships with a boundary:

• Cloud metadata endpoints are blocked unconditionally (169.254.169.254 and friends). This is never a legitimate feed and always a credential-theft target.

• Only stream-shaped schemes — http, https, ws, wss.

• Reading files and running processes are off by default. They turn a data client into something with authority over your machine.

• Credentials never reach a transcript. URLs, headers, and error messages are redacted; secrets belong in ${ENV_VAR} references, not in tool calls.

• Expressions are sandboxed — parsed to an AST and walked against an allowlist. There is no eval, no attribute access, and no import path.

Want none of that? One switch removes every restriction and every ceiling — any host, any path, any command, unlimited streams, buffers, and payloads:

fluxpy serve --unrestricted          # or FLUXPY_UNRESTRICTED=1, or security.unrestricted: true

Every limit is also individually configurable if you only need one raised. The expression sandbox stays on in every mode — that one is not a limit on you, it is what stops feed data from executing code.

Defaults suit a server on your own machine. Read docs/security.md before exposing one over HTTP.

Running it for a team

Four controls turn a laptop tool into shared infrastructure. All off by default, all independent:

access:
  read_only: true                              # withhold every state-changing tool
  tokens: ["sre:${FLUXPY_SRE_TOKEN}"]          # Authorization: Bearer <token>
  metrics: true                                # Prometheus at /metrics
  audit_log: /var/log/fluxpy/audit.jsonl       # one JSON line per request

The arrangement most teams land on: operations declare the streams in a version-controlled config file; agents connect read-only and read them. A new stream becomes a pull request rather than a tool call, and an agent's surface can no longer surprise you.

Withheld tools are not registered at all, so they never reach the model's context. Metrics include per-stream last_event_age_seconds — the one alert that catches a feed which went quiet, since a dead feed and a quiet feed look identical from the inside. The audit log records refused attempts too, and never records event payloads.

See docs/enterprise.md for the full picture, including Kubernetes manifests and a compliance summary.

Documentation

Full documentation index →

Contributing

Issues and pull requests are welcome — see CONTRIBUTING.md. Adding a source is deliberately easy: subclass Source, implement one run method, and the engine handles supervision, backpressure, pipelines, and fan-out for you.

git clone https://github.com/saudaljuaid/fluxpy && cd fluxpy
uv venv && uv pip install -e ".[dev]"
pytest && ruff check . && mypy src

License

MIT — see LICENSE.