Provides tools to interact with Apache Flink SQL Gateway, enabling session management, SQL query execution, streaming query results, and job control.
Click on "Deploy Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type @ followed by the MCP server name and your instructions, e.g., "@flink-mcp Run a SQL query to select from the 'orders' table and show the first 5 rows."
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
This project provides an MCP server that connects to Apache Flink SQL Gateway.
A running Apache Flink cluster and SQL Gateway
Start cluster: ./bin/start-cluster.sh
Start gateway: ./bin/sql-gateway.sh start -Dsql-gateway.endpoint.rest.address=localhost
Verify: curl http://localhost:8083/v3/info
Configure environment:
Set SQL_GATEWAY_API_BASE_URL (default http://localhost:8083). You can use a .env file at repo root.
Install and run via the console script:
pip install -e .
flink-mcpMCP clients should launch the server over stdio with command: flink-mcp.
Ensure SQL_GATEWAY_API_BASE_URL is set in your environment or .env.
flink_info (resource): returns cluster info from /v3/info.
open_new_session(properties?: dict) -> { sessionHandle, ... }.
get_config(sessionHandle: str): returns session configuration.
configure_session(sessionHandle: str, statement: str): apply session-scoped DDL/config (CREATE/USE/SET/RESET/LOAD/UNLOAD/ADD JAR).
run_query_collect_and_stop(sessionHandle: str, query: str, max_rows: int=5, max_seconds: float=15.0): execute, fetch up to N rows within T seconds, then STOP the job if a jobID is present; closes the operation.
run_query_stream_start(sessionHandle: str, query: str): execute a streaming query and return { jobID, operationHandle }; the job is left running.
fetch_result_page(sessionHandle: str, operationHandle: str, token: int): fetch a single page; returns { page, nextToken, isEnd }.
cancel_job(sessionHandle: str, jobId: str): issue STOP JOB '<jobId>', wait until DESCRIBE JOB status is not RUNNING; returns { jobID, status, jobGone, jobStatus }.
Tools are stateless; clients manage and pass session/operation handles explicitly.
run_query_stream_start returns both jobID and operationHandle; use fetch_result_page to stream results.
cancel_job issues STOP and waits using DESCRIBE JOB; close_operation is invoked internally where appropriate.
Endpoints target SQL Gateway v3-style paths.
This server cannot be deployed
Governed data discovery, exact queries, decisions, simulations, and runtime utilities over MCP.
Governed MCP gateway: one endpoint for your tools, with credential custody and audit log.
Query, join, profile, clean and convert CSV/JSON/Parquet with server-side DuckDB over MCP.
Streamkap CLI & MCP server - manage CDC pipelines, sources, destinations, and transforms
