<h1 align="center">fluxpy</h1>

<p align="center">
  <strong>Reactive, real-time data streaming for AI agents — as a first-class citizen of MCP.</strong>
</p>

<p align="center">
  <a href="https://github.com/saudaljuaid/fluxpy/actions/workflows/ci.yml"><img alt="CI" src="https://github.com/saudaljuaid/fluxpy/actions/workflows/ci.yml/badge.svg"></a>
  <a href="https://pypi.org/project/fluxpy/"><img alt="PyPI" src="https://img.shields.io/pypi/v/fluxpy.svg"></a>
  <a href="https://pypi.org/project/fluxpy/"><img alt="Python" src="https://img.shields.io/pypi/pyversions/fluxpy.svg"></a>
  <a href="LICENSE"><img alt="License: MIT" src="https://img.shields.io/badge/license-MIT-green.svg"></a>
  <a href="https://modelcontextprotocol.io"><img alt="MCP" src="https://img.shields.io/badge/MCP-2026--07--28-blueviolet.svg"></a>
  <a href="docs/clients/README.md"><img alt="Clients" src="https://img.shields.io/badge/clients-24_AI_tools-orange.svg"></a>
</p>

<p align="center">
  <a href="docs/getting-started.md">Getting started</a> ·
  <a href="docs/concepts.md">Concepts</a> ·
  <a href="docs/tools.md">Tools</a> ·
  <a href="docs/clients/README.md">Client setup</a> ·
  <a href="docs/recipes.md">Recipes</a> ·
  <a href="docs/faq.md">FAQ</a>
</p>

---

## 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.

## 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:

| | How | When to use it |
|---|---|---|
| **Push** | `subscriptions/listen` → `ResourceUpdated` | The client speaks MCP 2026-07-28. True server-initiated push, zero polling. |
| **Pull** | `flux_poll(cursor)` | Everywhere. Exactly-once and gap-aware — you learn what you missed. |
| **Block** | `flux_wait(where=...)` | React *inside one turn*. One tool call parks until the thing you care about happens. |

```mermaid
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](docs/tools.md) in total.

And it is built for three quite different people:

| | |
|---|---|
| **Anyone** | 15 [presets](#presets) — ready-made streams behind a plain-language name. "Watch Hacker News", "tell me if my site goes down", "is my computer running hot". No spec to author, no API to learn. → **[fluxpy for everyone](docs/for-everyone.md)** |
| **Developers** | 13 source types, 19 operators, a sandboxed filter language, cursor semantics, and a CLI that reproduces the whole engine outside any client. → **[Concepts](docs/concepts.md)** |
| **Organisations** | Read-only mode, bearer auth, Prometheus metrics, JSONL audit logging, Kubernetes manifests. All off by default. → **[For organisations](docs/enterprise.md)** |

---

## 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:

```bash
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:

<details open>
<summary><b>Claude Code</b></summary>

```bash
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."*
</details>

<details>
<summary><b>Claude Desktop</b></summary>

Add to `claude_desktop_config.json` ([where is it?](docs/clients/claude-desktop.md)):

```json
{
  "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.
</details>

<details>
<summary><b>Cursor</b></summary>

Add to `~/.cursor/mcp.json`:

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

<details>
<summary><b>VS Code / Copilot</b></summary>

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

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

<details>
<summary><b>Codex CLI</b></summary>

```bash
codex mcp add fluxpy -- uvx --from git+https://github.com/saudaljuaid/fluxpy fluxpy serve
```
</details>

<details>
<summary><b>ChatGPT</b> (remote, HTTP)</summary>

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

```bash
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](docs/clients/chatgpt.md) — read the authentication section before exposing it publicly.
</details>

**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](docs/clients/README.md)**, or let fluxpy do it:

```bash
# 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.

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

| | |
|---|---|
| `demo_match` · `my_computer` | Nothing required. No network, no credentials. |
| `news` · `hacker_news` · `youtube_channel` · `reddit` · `web_page` | Feeds, and any page at all watched for changes. |
| `github_releases` · `github_commits` · `ci_webhooks` · `app_log` · `website_up` | Software and work. |
| `crypto_price` · `earthquakes` · `wikipedia` | Money and the world. |

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:

```jsonc
// 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:

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

…or blocks until something big happens:

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

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

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

---

## Sources

| Type | What it connects to |
|---|---|
| `sse` | Server-Sent Events endpoints. Resumes with `Last-Event-ID` across reconnects. |
| `websocket` | WebSocket feeds, with subscribe-on-connect frames re-sent on every reconnect. |
| `http_poll` | Any REST endpoint, on an interval. `emit: on_change` turns state into events. |
| `rss` | Any RSS or Atom feed — news, blogs, podcasts, YouTube, releases. Deduplicated. |
| `web_page` | Any page at all, watched for changes. The fallback when there is no feed. |
| `webhook` | **Inbound** HTTP. GitHub, Stripe, Sentry, CI — push *to* the agent. |
| `file_tail` | A growing file, surviving truncation and log rotation. |
| `process` | The stdout of a long-running command (`kubectl logs -f`, `journalctl -f`). |
| `interval` | A clock-driven heartbeat or synthetic feed. |
| `replay` | A recorded capture, replayed at any speed. Makes streaming testable. |
| `derived` | Other fluxpy streams, merged. Composes to any depth. |
| `scoreboard`&nbsp;·&nbsp;`random_walk`&nbsp;·&nbsp;`system_metrics` | Zero-setup demos. No network, no credentials. |

Full options in [docs/sources.md](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](docs/operators.md) explains exactly when each is right.

---

## Try it without an agent

```bash
# 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:

```yaml
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:

```bash
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](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:

```yaml
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](docs/enterprise.md)** for the full picture, including Kubernetes manifests and a compliance summary.

---

## Documentation

**[Full documentation index →](docs/README.md)**

| | |
|---|---|
| **[fluxpy for everyone](docs/for-everyone.md)** | **No programming.** What it can watch, in plain English. |
| [Getting started](docs/getting-started.md) | Install, first stream, first watch. |
| [FAQ](docs/faq.md) | Short answers, including "why is my filter matching nothing?" |
| [Concepts](docs/concepts.md) | Streams, events, cursors, backpressure, push vs pull. |
| [Tool reference](docs/tools.md) | All 28 MCP tools with arguments and examples. |
| [Sources](docs/sources.md) | Every source type, and how to write your own. |
| [Operators](docs/operators.md) | Every operator, with the throttle/debounce/sample decision. |
| [Expressions](docs/expressions.md) | The filter language and its sandbox. |
| [Recipes](docs/recipes.md) | Complete worked setups for real problems. |
| [Architecture](docs/architecture.md) | How it works inside, and why. |
| [Security](docs/security.md) | Threat model and the hardened profile. |
| [Deployment](docs/deployment.md) | Docker, remote HTTP, tunnels, systemd. |
| **[For organisations](docs/enterprise.md)** | Read-only, auth, metrics, audit, Kubernetes. |
| **[Client setup](docs/clients/README.md)** | All 24 tools. Dedicated guides: [Claude Code](docs/clients/claude-code.md) · [Claude Desktop](docs/clients/claude-desktop.md) · [Cursor](docs/clients/cursor.md) · [ChatGPT](docs/clients/chatgpt.md) · [Codex](docs/clients/codex.md) · [VS Code](docs/clients/vscode.md) · [others](docs/clients/others.md) |

## Contributing

Issues and pull requests are welcome — see [CONTRIBUTING.md](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.

```bash
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](LICENSE).