GlassFlow Server

Model Context Protocol server for managing and diagnosing GlassFlow streaming pipelines. Exposes pipeline CRUD, metrics queries, log search, and a composite diagnostic tool as MCP tools that AI agents (Claude Code, etc.) can call over SSE transport.

Features

• Multi-cluster — connect to multiple GlassFlow deployments and switch between them at runtime
• Pipeline management — create, list, get, edit, stop, resume, delete pipelines
• Diagnostics — query throughput, latency, DLQ state, and error logs
• NATS JetStream diagnostics (Enterprise) — inspect stream message counts and consumer health (get_pipeline_streams, get_stream_report, get_consumer_report) to spot stuck consumers, backlogs, and subject mismatches
• diagnose_pipeline — single-call diagnostic snapshot combining health, metrics, DLQ, recent errors, and NATS stream health
• V3 config reference — MCP resource with the complete pipeline configuration format
• Uses the official GlassFlow Python SDK

Quick start

Local development

pip install -e .

# Option A: auto-connect a default cluster via env var
export GLASSFLOW_API_URL="http://localhost:8081"
python -m glassflow_mcp.server

# Option B: start with no cluster, connect at runtime via tools
python -m glassflow_mcp.server

Connect Claude Code

claude mcp add --transport sse glassflow http://localhost:8080/sse

Start a new Claude Code session — the GlassFlow tools will appear automatically.

Available tools

Cluster management

Connect to one or more GlassFlow clusters and switch between them. All pipeline and diagnostic tools operate against the active cluster.

Tool	Description
connect_cluster	Register a GlassFlow cluster by name + API URL (+ optional VM/VL URLs)
list_clusters	Show all connected clusters with active indicator
switch_cluster	Change the active cluster
disconnect_cluster	Remove a cluster connection

Example flow:

You: "Connect to my staging cluster at http://staging-api:8081"
  → Agent calls: connect_cluster(name="staging", api_url="http://staging-api:8081")

You: "List my pipelines"
  → Agent calls: list_pipelines()  (uses staging)

You: "Now connect to production at http://prod-api:8081"
  → Agent calls: connect_cluster(name="production", api_url="http://prod-api:8081")

You: "Switch to production"
  → Agent calls: switch_cluster("production")

You: "List pipelines"
  → Agent calls: list_pipelines()  (now uses production)

If GLASSFLOW_API_URL is set as an env var, the server auto-connects a default cluster on startup for backwards compatibility.

Pipeline management

Tool	Description
list_pipelines	List all pipelines with status
get_pipeline	Get full V3 pipeline configuration
get_pipeline_health	Get pipeline health and status
create_pipeline	Create a new pipeline (V3 JSON config)
edit_pipeline	Edit a stopped pipeline
stop_pipeline	Stop a running pipeline
resume_pipeline	Resume a stopped pipeline
delete_pipeline	Delete a pipeline

Diagnostics

Tool	Description
diagnose_pipeline	Complete diagnostic snapshot (health + metrics + DLQ + errors)
query_pipeline_metrics	Query specific metrics (throughput, latency, DLQ rate, bytes)
query_custom_metric	Custom PromQL query (restricted to gfm_* metrics)
query_pipeline_logs	Search logs by pipeline, severity, and component
get_pipeline_errors	Recent ERROR/WARN logs for a pipeline
get_dlq_state	Dead-letter queue message count

Resources

URI	Description
glassflow://docs/pipeline-v3-format	Complete V3 pipeline configuration reference with examples

Configuration

All configuration is via environment variables. These configure the default cluster that auto-connects on startup. Additional clusters can be added at runtime via connect_cluster.

Variable	Default	Description
GLASSFLOW_API_URL	http://glassflow-api....:8081	GlassFlow REST API URL (default cluster)
VICTORIAMETRICS_URL	http://victoria-metrics....:8428	VictoriaMetrics URL (default cluster)
VICTORIALOGS_URL	http://victoria-logs....:9428	VictoriaLogs URL (default cluster)
MCP_PORT	8080	Port the SSE server listens on

VictoriaMetrics and VictoriaLogs URLs are optional — metrics and log tools gracefully degrade when not configured for a cluster.

Deployment

Docker

docker build -t glassflow-mcp-server .
docker run -p 8080:8080 \
  -e GLASSFLOW_API_URL=http://your-glassflow-api:8081 \
  glassflow-mcp-server

Kubernetes

Example manifests are provided in k8s/examples/. Copy them, edit the CHANGEME values, and apply:

kubectl apply -f k8s/examples/deployment.yaml -f k8s/examples/service.yaml

Then connect via port-forward:

kubectl port-forward -n <namespace> svc/glassflow-mcp 8080:8080
claude mcp add --transport sse glassflow http://localhost:8080/sse

See k8s/README.md for full details including optional Ingress setup.

PyPI

pip install mcp-server-glassflow
mcp-server-glassflow

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest -v

# Lint
ruff check src/ tests/
ruff format --check src/ tests/

License

Apache 2.0