Skip to main content
Glama
clementlemon02

text2flink

by clementlemon02
OverviewSchemaRelated ServersScoreDiscussions
Python
Local

text2flink

CI

Alpha: runnable & tested, APIs may change.

Give your AI coding agent the ability to write Apache Flink SQL that provably works.

AI agents write plausible Flink SQL — and streaming SQL is exactly where plausible isn't correct. A 30-second window where you meant one minute; late records kept when they should be dropped. It compiles, it runs, it returns believable rows, and it silently under-reports in production for weeks. The agent has no way to know it got the watermark or window semantics wrong — a compile check and a golden-file diff both pass.

text2flink closes that loop. Register it once as an MCP server and your agent can verify the SQL it just wrote on a real Flink cluster and self-correct from precise feedback — before you ever see it:

claude mcp add text2flink -- python3 -m text2flink.mcp_server

The agent calls verify_flink_sql with its SQL + a schema + sample data + the intended semantics, and gets back a machine-readable verdict it can act on:

{ "verified": false,
  "violations": [
    { "kind": "value_mismatch",  "detail": "(00:00:00, u1): got 2.0, expected 3.0" },
    { "kind": "spurious_window", "detail": "(00:00:30, u1) emitted but not expected" } ] }

The agent reads the violations ("my window is 30s, should be 60s"), fixes the SQL, and verifies again — a ~1–2s loop — until verified: true. Not "does it parse", not a golden diff: actual streaming correctness, run on real Flink. (The server also exposes generate_flink_job / ground_kafka_topic / deploy_to_kafka for the draft-and-ship flow — see docs/USE_CASES.md.)

Or drive it yourself — CLI + CI

The same engine runs standalone. Write your Flink SELECT and a small test-file next to it (sample data + the semantics you expect); verify grades it on Flink and exits non-zero on failure, so it drops straight into a CI gate:

text2flink verify orders_per_min.test.json
#  ✘ FAIL  orders_per_min
#    [value_mismatch] (00:00:00, u1): got 2.0, expected 3.0   ← 30s window, not 1 min

Fastest taste, no setup: python3 examples/verify_flink_sql.py verifies a correct job, then catches a subtly-broken one.

New here? docs/USE_CASES.md for the AI-agent path, docs/VERIFY.md for the CLI/CI path, DESIGN.md for the architecture.

Status

Alpha — runnable and execution-verified today; APIs may change. The core loop is proven: take a Flink SQL job (hand-written or generated), run it on a real local Flink cluster, and assert streaming-specific properties (windows fire once, counts match a ground truth, event-time and watermarks respected) instead of brittle golden diffs — and, for generated jobs, repair a wrong one automatically from structured verification feedback.

What it does today

  • Verify an AI agent's Flink SQL over MCP — the verify_flink_sql tool runs the SQL the agent wrote on real Flink and returns verified + machine-readable violations, so the agent self-corrects in a ~1–2s loop. This is the headline; see docs/USE_CASES.md.

  • Verify hand-written Flink SQL — text2flink verify job.test.json runs your own SELECT on real Flink and grades it against the streaming semantics you declare, with precise violations on failure and a non-zero exit for CI. See docs/VERIFY.md.

  • Draft a job from natural language — NL → JobSpec extraction via a model-agnostic LLM layer that runs on OpenAI (OPENAI_API_KEY) or Anthropic (ANTHROPIC_API_KEY), with an offline heuristic proposer fallback so everything runs with no API key. A convenience entry point into the same verifier.

  • Windowed aggregates — tumbling, hopping, and session windows (per-key inactivity-gap merging via the Flink SESSION TVF).

  • Updating operators — unbounded GROUP BY (running totals), top-N, and deduplication (ROW_NUMBER()), verified via batch-mode collect and deployed to an upsert-kafka topic keyed by their partition keys.

  • Joins — time-bounded interval joins of two event-time streams, and temporal (versioned-table) joins (FOR SYSTEM_TIME AS OF). Semantics confirmed against real Flink before being encoded in the oracle.

  • Late-data / watermark correctness as a first-class, tested dimension: the oracle drops records whose event-time is at/behind the watermark, and a task verifies the generated job drops exactly those records — the thing batch Text2SQL cannot express.

  • Live Kafka grounding — discover a source schema from a real Kafka topic (sample → infer → wire the connector), graded by running on real Flink against the real topic in bounded mode. Avro grounding is supported via the Flink avro format end-to-end; the Confluent Schema Registry (avro-confluent) path is verified offline against a mock registry.

  • Deployable Kafka pipelines (topic → topic) — a spec compiles to a complete INSERT INTO <sink> SELECT … script; still verified by running to completion and grading the sink topic against the oracle.

  • Multiple codegen targets — Flink SQL (execution-verified), a submittable PyFlink Table API program (flink run -py job.py), and Apache Spark SQL (a portability proof graded against the same oracle; the product stays Flink-first).

StreamBench

A declarative, contributable benchmark scored by execution pass rate, not string match — 22 tasks as JSON files in streambench/tasks/, grouped into core / advanced / adversarial tiers. Add a task or submit a model result without writing engine code (see CONTRIBUTING.md). Live results in LEADERBOARD.md:

Proposer

core

advanced

adversarial

total

openai-gpt-4o

6/6

9/9

7/7

22/22 (100%)

heuristic (offline)

6/6

9/9

0/7

15/22 (68%)

The adversarial tier (misleading phrasing, implicit/negated filters, multi-key grouping, reworded windows, non-second time units) makes the benchmark discriminating, and has surfaced real bugs no prior task exercised. Every task's gold spec is execution-verified on Flink (scripts/check_gold.py), so a broken task can't hide as a proposer failure.

Roadmap

Confluent Schema Registry (avro-confluent) end-to-end, DataStream codegen, K8s deployment manifests, and a bigger adversarial tier.

Requirements

This machine's default Java (26) and Python (3.14) are too new for Flink, so Phase 0 uses:

  • Apache Flink + JDK 17, installed via Homebrew (brew install apache-flink openjdk@17).

  • Python 3.11+ for the orchestrator (no PyFlink dependency — it drives Flink's SQL client as a subprocess, so the system Python is fine).

  • For the live-Kafka example only: brew install kafka, plus the Flink Kafka SQL connector jar in Flink's lib/ (e.g. flink-sql-connector-kafka-5.0.0-2.2.jar from Maven Central).

  • For the Avro example: the Flink Avro format jar in Flink's lib/ (e.g. flink-sql-avro-2.2.1.jar from Maven Central).

  • For the (experimental) cross-engine example only: brew install apache-spark (JDK 17).

Install

pip install "git+https://github.com/clementlemon02/text2flink.git"   # the library + `text2flink` CLI

(A pip install text2flink from PyPI is a tag away — see RELEASING.md.) Contributors work from a checkout instead: pip install -e ".[dev]" (editable, with pytest). No install is even required to try it: python3 -m text2flink.cli verify <file> runs from a checkout, and the example scripts below run directly. Python 3.10+; running a verify needs a local Flink + JDK 17 (see Requirements).

Run it

text2flink verify examples/verify/orders_per_min.test.json  # verify your Flink SQL (see docs/VERIFY.md)
python3 examples/verify_flink_sql.py   # verify a correct job, then catch a subtly-broken one
python3 examples/tumbling_count.py     # generate a job, verify it, then catch & repair a broken one
python3 examples/run_streambench.py    # run StreamBench, report execution pass rate

run_streambench.py uses the offline heuristic proposer by default. To use a real model:

OPENAI_API_KEY=sk-... python3 examples/run_streambench.py       # OpenAI (default gpt-4o)
ANTHROPIC_API_KEY=sk-... python3 examples/run_streambench.py    # Anthropic

Related MCP server: Kafka MCP Server

Layout

text2flink/
  ir.py          # IR: JobSpec (windows/agg) + IntervalJoinSpec + TemporalJoinSpec
  codegen.py     # IR -> Flink SQL: aggregates, interval joins, temporal joins
  data.py        # row-dict -> Flink CSV
  oracle.py      # reference interpreter: ground truth incl. late-drop + session windows
  runtime.py     # local Flink cluster lifecycle + run SQL (sql-client), collect results
  gateway.py     # SQL Gateway harness: submit over REST (~1-2s/case, no per-run JVM boot)
  assertions.py  # property-based streaming correctness checks
  verify.py      # verify(spec) + verify_sql/verify_batch/gateway -> run on Flink -> assert
  testfile.py    # load a verify test-file (SQL + sources + data + expected semantics)
  cli.py         # `text2flink verify` / `text2flink cluster` — the correctness-layer command
  llm.py         # model-agnostic LLM clients: OpenAI + Anthropic (raw HTTPS, no SDK)
  proposer.py    # NL -> JobSpec + repair: LLM / offline-heuristic / scripted proposers
  pipeline.py    # extract -> verify -> repair loop for one task
  kafka.py       # live Kafka grounding: sample a topic, infer schema, build a Source
  schema_registry.py # Avro/Schema-Registry grounding: exact types via avro-confluent
  deploy.py      # deployable INSERT jobs (append + upsert sinks), verify by consuming
  mcp_server.py  # MCP server (stdio) — lets an AI assistant call text2flink as a tool
  mcp_tools.py   # generate-and-run-on-Flink logic behind the MCP tool
  pyflink.py     # third codegen target: same IR -> a submittable PyFlink Table API program
  spark.py       # EXPERIMENTAL portability proof: same IR -> Spark SQL, same oracle
  streambench.py # loader for the StreamBench corpus
streambench/
  tasks/*.json   # the benchmark corpus (declarative, contributable)
examples/
  verify_flink_sql.py  # verify hand-written Flink SQL; catch a subtly-broken job
  verify/              # a verify test-file (.test.json) + the .sql it checks
  tumbling_count.py
  run_streambench.py
  kafka_grounding.py   # discover schema from a live Kafka topic, then verify
  kafka_pipeline.py    # deployable topic -> topic pipeline (INSERT INTO sink), verified
  avro_pipeline.py     # real Avro end-to-end on Flink (no Schema Registry)
  cross_engine.py      # (experimental) one IR compiled + verified on Flink AND Spark
tests/           # fast offline suite (no Flink): corpus, oracle, codegen, parsing
LICENSE          # Apache-2.0
CONTRIBUTING.md  # how to add a StreamBench task
.github/workflows/ci.yml   # runs the offline tests on 3.10–3.12

Development

pip install -e ".[dev]"
pytest                          # fast, offline — no Flink required
python3 scripts/check_gold.py   # verify every task's gold spec on Flink (task soundness)
python3 scripts/leaderboard.py  # score a proposer, update results/ + LEADERBOARD.md

Contributions welcome — the highest-value ones are a new StreamBench task (especially adversarial) and submitting a model to the leaderboard. See CONTRIBUTING.md.

Available Tools

3 tools
deploy_to_kafkaA

Produce a deployable topic->topic Flink job: INSERT INTO a Kafka sink topic SELECT ... The SELECT logic is verified on the sample data by running it on real Flink; the returned deployable_sql is the full pipeline (upsert-kafka sink for updating jobs). Use when the user wants to WRITE results to a Kafka topic, not just query them.

ParametersJSON Schema
NameRequiredDescriptionDefault
requestYes
sourcesYesSource tables. Each: name, columns [[name, flinkType], ...], event_time (column), watermark_delay_seconds, optional primary_key.
bootstrapNo
sink_topicYesDestination Kafka topic.
sample_dataYesMap of source name -> list of row objects. TIMESTAMP columns may be integer seconds (offset from a base) or a 'yyyy-MM-dd HH:mm:ss.SSS' string.

TDQS

A4.3/5.0
Behavior4/5

Does the description disclose side effects, auth requirements, rate limits, or destructive behavior?

No annotations are provided, so the description carries the full burden. It discloses that the SELECT logic is verified on real Flink and that the returned deployable_sql uses an upsert-kafka sink for updating jobs, which is valuable behavioral context beyond the basic purpose.

Agents need to know what a tool does to the world before calling it. Descriptions should go beyond structured annotations to explain consequences.

Conciseness5/5

Is the description appropriately sized, front-loaded, and free of redundancy?

The description is three sentences with key information front-loaded. Every sentence contributes purpose, behavior, or usage context without redundancy.

Shorter descriptions cost fewer tokens and are easier for agents to parse. Every sentence should earn its place.

Completeness3/5

Given the tool's complexity, does the description cover enough for an agent to succeed on first attempt?

The tool has 5 parameters and no output schema, requiring more detail. The description explains the high-level flow but omits the semantics of 'request' and 'bootstrap', and does not describe the structure of the returned deployable_sql beyond noting it is a full pipeline.

Complex tools with many parameters or behaviors need more documentation. Simple tools need less. This dimension scales expectations accordingly.

Parameters3/5

Does the description clarify parameter syntax, constraints, interactions, or defaults beyond what the schema provides?

The schema covers sources, sample_data, and sink_topic (60% coverage), but the description does not explain the 'request' or 'bootstrap' parameters, which also lack schema descriptions. The description adds no parameter-specific semantics beyond what the schema already provides.

Input schemas describe structure but not intent. Descriptions should explain non-obvious parameter relationships and valid value ranges.

Purpose5/5

Does the description clearly state what the tool does and how it differs from similar tools?

The description states a specific action: 'Produce a deployable topic->topic Flink job' with a clear verb and resource. It further distinguishes from siblings by noting 'Use when the user wants to WRITE results to a Kafka topic, not just query them.'

Agents choose between tools based on descriptions. A clear purpose with a specific verb and resource helps agents select the right tool.

Usage Guidelines5/5

Does the description explain when to use this tool, when not to, or what alternatives exist?

Explicitly provides the trigger condition: 'Use when the user wants to WRITE results to a Kafka topic.' It contrasts with query-only use, indicating the alternative path, though it doesn't name sibling tools directly.

Agents often have multiple tools that could apply. Explicit usage guidance like "use X instead of Y when Z" prevents misuse.

ground_kafka_topicA

Discover a Kafka topic's schema by sampling its messages, returning a schema + sample rows ready to pass to generate_flink_job. Use when the user references a real topic and you don't have its schema. Requires a reachable Kafka broker with the topic populated.

ParametersJSON Schema
NameRequiredDescriptionDefault
topicYes
bootstrapNoBroker (default localhost:9092).
event_timeYesName of the event-time column.
sample_sizeNo
watermark_delay_secondsNo

TDQS

A4.1/5.0
Behavior4/5

Does the description disclose side effects, auth requirements, rate limits, or destructive behavior?

No annotations are provided, so the description carries the full burden. It discloses the sampling behavior, the requirement of a reachable broker and populated topic, and what it returns. This goes beyond a simple read assertion, though it does not mention failure modes or potential side effects (which are minimal for a read/sampling tool).

Agents need to know what a tool does to the world before calling it. Descriptions should go beyond structured annotations to explain consequences.

Conciseness5/5

Is the description appropriately sized, front-loaded, and free of redundancy?

The description is two sentences, front-loading the main function and then providing usage and requirements. Every sentence earns its place with actionable information and no fluff.

Shorter descriptions cost fewer tokens and are easier for agents to parse. Every sentence should earn its place.

Completeness4/5

Given the tool's complexity, does the description cover enough for an agent to succeed on first attempt?

With 5 parameters, no output schema, and no annotations, the description covers the tool's purpose, return type, prerequisites, and downstream usage. It lacks details on error handling or parameter-dependent behavior, but for the tool's complexity and available structured fields, it is sufficiently complete. The requirement that 'the topic populated' hints at failure conditions.

Complex tools with many parameters or behaviors need more documentation. Simple tools need less. This dimension scales expectations accordingly.

Parameters2/5

Does the description clarify parameter syntax, constraints, interactions, or defaults beyond what the schema provides?

Schema description coverage is only 40%, and the description does not compensate. Parameters like topic, sample_size, and watermark_delay_seconds lack explanation beyond their raw schema names. The description mentions 'event_time' indirectly but doesn't explain how it is used. No additional meaning is provided over the schema.

Input schemas describe structure but not intent. Descriptions should explain non-obvious parameter relationships and valid value ranges.

Purpose5/5

Does the description clearly state what the tool does and how it differs from similar tools?

The description starts with a specific verb ('Discover') and identifies the exact resource ('Kafka topic's schema') and method ('by sampling its messages'). It clearly distinguishes itself from siblings (generate_flink_job, deploy_to_kafka) by stating its result is 'a schema + sample rows ready to pass to generate_flink_job'.

Agents choose between tools based on descriptions. A clear purpose with a specific verb and resource helps agents select the right tool.

Usage Guidelines4/5

Does the description explain when to use this tool, when not to, or what alternatives exist?

The description explicitly states when to use it: 'Use when the user references a real topic and you don't have its schema.' It does not explicitly mention alternatives or when not to use, but the usage scenario is clear and contextual. The sibling tools are different enough that no exclusion is needed.

Agents often have multiple tools that could apply. Explicit usage guidance like "use X instead of Y when Z" prevents misuse.

Tool Schema Changelog

Recent tool additions, removals, and schema changes observed during successful MCP inspections. Dates show when Glama detected each change.

  1. 3 tool updatesv0.1.0
    • First observeddeploy_to_kafka
    • First observedgenerate_flink_job
    • First observedground_kafka_topic
Install Server
RayrunBuild, deploy, and operate MCP servers without becoming an infrastructure team.ray.runAd

TDQS

A4.4/5.0
Disambiguation5/5

Each tool has a clearly distinct purpose: schema discovery, job generation with verification, and deployment to Kafka. There is no overlap that would confuse an agent.

Naming Consistency5/5

All tool names follow a consistent verb_noun pattern in snake_case (generate_flink_job, ground_kafka_topic, deploy_to_kafka), making the set predictable and easy to navigate.

Tool Count5/5

Three tools is a well-scoped count for a focused pipeline server, covering the essential stages without bloat or deficiency.

Completeness5/5

The tools form a complete workflow: discover a Kafka schema, generate and verify a Flink job, and deploy it to a Kafka sink. No obvious gaps or dead ends for the stated domain.

Maintenance

ActivityMaintained
ResponsivenessSyncing

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Related MCP Connectors

Related MCP Servers

  • KafkaIQ

    ojhaayush03
    F
    license
    Not graded
    quality
    D
    maintenance
    Enables AI assistants to manage and monitor Apache Kafka clusters through natural language, providing real-time operations, health monitoring, consumer lag analysis, and temporal trend detection for intelligent cluster management.
    -

  • Kafka MCP Server

    shivamxtech
    A
    license
    Not graded
    quality
    C
    maintenance
    Enables AI agents to interact with Apache Kafka through natural language, supporting operations like producing/consuming messages, managing topics, and querying brokers, partitions, and consumer group offsets.
    1
    MIT

  • CockroachDB MCP Server

    fastmcp-me
    A
    license
    Not graded
    quality
    D
    maintenance
    A natural language interface for LLMs to manage, monitor, and query CockroachDB. It provides tools for cluster monitoring, database operations, table management, and query execution.
    MIT

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/clementlemon02/text2flink'

If you have feedback or need assistance with the MCP directory API, please join our Discord server