MCP 服务器

fluxpy

An MCP server that enables real-time data streaming for AI agents, with persistent connections to live feeds, reactive operator pipelines, and consumption via push, pull, or block modes.

README

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

Reactive, real-time data streaming for AI agents — as a first-class citizen of MCP.

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

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.

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:


Anyone	15 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
Developers	13 source types, 19 operators, a sandboxed filter language, cursor semantics, and a CLI that reproduces the whole engine outside any client. → Concepts
Organisations	Read-only mode, bearer auth, Prometheus metrics, JSONL audit logging, Kubernetes manifests. All off by default. → For organisations

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:

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

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>Claude Desktop</summary>

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. </details>

<details> <summary>Cursor</summary>

Add to ~/.cursor/mcp.json:

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

</details>

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

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"]
    }
  }
}

</details>

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

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

</details>

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

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. </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, 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


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

// 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

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` · `random_walk` · `system_metrics`	Zero-setup demos. No network, no credentials.

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 →


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

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.

fluxpy

README

The problem

What fluxpy does

Quickstart

Presets

What it looks like

Sources

Operators

Try it without an agent

Configuration

Security

Running it for a team

Documentation

Contributing

License

推荐服务器