MCP Server for Confluent Manager for Apache Flink

Confluent Manager for Apache Flink (CMF) includes a Model Context Protocol (MCP) server that lets AI agents and coding assistants inspect and manage Apache Flink® resources directly, alongside the existing REST APIs and CLI Operations.

MCP is an open protocol that lets AI applications connect to external tools and data sources. When enabled, the CMF MCP server exposes environments, applications, statements, compute pools, secrets, savepoints, Kafka catalogs, and Kubernetes clusters as a set of tools that an MCP client can call.

Important

The CMF MCP server is disabled by default, because it’s a preview feature. The endpoint is versioned as v1alpha1 and might change in future releases.

Enable the MCP server

The MCP server is off by default and is configured through Helm values.

Helm value	Default	Description
`cmf.mcp.enabled`	`false`	Master switch for the MCP server. Must be `true` for any MCP tool to be available.
`cmf.mcp.writeTools.enabled`	`false`	Enables the mutating tools that create, update, or delete resources.
`cmf.mcp.logs.enabled`	`true`	Enables the `get_flinkapplicationlogs` tool. Pod logs can contain query payloads, credentials, and personally identifiable information (PII), so set this to `false` to drop the tool while keeping the rest of the read-only surface.

To enable read-only access, upgrade your CMF Helm release with:

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
  --set cmf.mcp.enabled=true

To also allow AI agents to create, update, or delete resources, add:

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
  --set cmf.mcp.enabled=true \
  --set cmf.mcp.writeTools.enabled=true

For general installation steps, see Installation with Helm.

Connect an MCP client

The MCP server uses the Streamable HTTP transport and is served at /cmf/mcp/v1alpha1 on the same host and port as the CMF REST API. Point your MCP client at this URL, for example:

http://<cmf-host>:8080/cmf/mcp/v1alpha1

Most MCP clients accept a URL-based server configuration. For example, using the Claude Code CLI:

claude mcp add --transport http cmf http://<cmf-host>:8080/cmf/mcp/v1alpha1

Consult your MCP client’s documentation for the exact configuration format it expects.

Authentication and authorization

Every MCP tool call is dispatched through the checks used by the REST APIs, so the same authentication and role-based access control (RBAC) authorization apply. A caller only sees and can act on resources it’s already authorized to access; MCP grants no additional privileges.

Because write tools can modify or delete cluster resources, only give CMF credentials that have write-tool access to agents you trust to take those actions.

Available tools

Read-only tools

The MCP server exposes 19 read-only tools, plus get_flinkapplicationlogs, which is gated separately by cmf.mcp.logs.enabled.

Category	Tool	Description
System	`get_cmfoverview`	Returns CMF version and resource counts. Use this first to orient an agent session.
Environments	`list_flinkenvironments`	Lists environments with their Kubernetes namespaces.
Environments	`get_flinkenvironment`	Gets full configuration for one environment, including defaults and secret mappings.
Applications	`list_flinkapplications`	Lists applications in an environment with job status.
Applications	`get_flinkapplication`	Gets an application’s spec, status, and resource usage.
Applications	`list_flinkapplicationevents`	Gets an application’s event history (deployments, status transitions, errors).
Applications	`list_flinkapplicationinstances`	Lists recent job instances for an application.
Applications	`get_flinkapplicationinstance`	Gets details of a single job instance, including its spec snapshot.
Applications	`get_flinkapplicationlogs`	Retrieves recent JobManager or TaskManager pod logs.
Statements	`list_flinkstatements`	Lists SQL statements in an environment, filterable by phase or compute pool.
Statements	`get_flinkstatement`	Gets a statement’s SQL text, phase, and status.
Statements	`list_flinkstatementexceptions`	Gets recent exceptions for a failing statement.
Statements	`get_flinkstatementresults`	Gets result rows for a running or completed `SELECT` statement.
Compute pools	`list_flinkcomputepools`	Lists compute pools with phase and resource usage.
Compute pools	`get_flinkcomputepool`	Gets a compute pool’s spec, status, and resource usage.
Savepoints	`list_flinksavepoints`	Lists savepoints for an application, statement, or detached savepoints.
Savepoints	`get_flinksavepoint`	Gets full details of a single savepoint.
Secrets	`list_flinksecrets`	Lists registered secrets (values masked) and the environments they’re attached to.
Catalogs	`list_kafkacatalogs`	Lists Kafka catalogs, or a specific catalog’s databases.
Kubernetes clusters	`list_kubernetesclusters`	Lists registered Kubernetes clusters with connectivity and lifecycle state.

Write tools

Write tools are only registered when cmf.mcp.writeTools.enabled is set to true. Clients are typically prompted to confirm before an agent calls one of these tools.

Category	Tool	Description
Environments	`apply_flinkenvironment`	Creates or updates an environment (idempotent).
Environments	`delete_flinkenvironment`	Deletes an environment. Fails if applications, statements, or compute pools still exist.
Applications	`apply_flinkapplication`	Creates or updates an application (idempotent).
Applications	`start_flinkapplication`	Starts a suspended application, optionally from a savepoint.
Applications	`suspend_flinkapplication`	Suspends a running application.
Applications	`delete_flinkapplication`	Deletes an application.
Statements	`create_flinkstatement`	Submits a SQL statement for execution in a compute pool.
Statements	`start_flinkstatement`	Resumes a stopped statement, optionally from a savepoint.
Statements	`stop_flinkstatement`	Stops a running statement.
Statements	`update_flinkstatement`	Changes a statement’s parallelism, the only spec field mutable after creation.
Statements	`delete_flinkstatement`	Deletes a statement.
Compute pools	`create_flinkcomputepool`	Creates a compute pool.
Compute pools	`update_flinkcomputepool`	Updates a compute pool’s spec. Rejected while statements are running on it.
Compute pools	`delete_flinkcomputepool`	Deletes a compute pool.
Secrets	`attach_flinksecret`	Attaches a secret to an environment through a named mapping.
Secrets	`detach_flinksecret`	Detaches a secret mapping from an environment.
Secrets	`delete_flinksecret`	Deletes a secret.
Savepoints	`trigger_flinksavepoint`	Triggers a savepoint for a running application or statement.
Savepoints	`pin_flinksavepoint`	Pins or unpins a savepoint to exempt it from automatic retention cleanup.
Savepoints	`detach_flinksavepoint`	Detaches an application savepoint into a standalone detached savepoint.
Savepoints	`bulk_delete_flinksavepoints`	Bulk-deletes savepoints by source and age. Pinned savepoints are never deleted.
Savepoints	`create_flinkdetachedsavepoint`	Registers a standalone savepoint that isn’t attached to any resource.
Savepoints	`delete_flinkdetachedsavepoint`	Deletes a detached savepoint record, not the underlying storage data.
Savepoints	`delete_flinksavepoint`	Deletes a savepoint record attached to an application or statement.

Resources and prompts

The MCP server also exposes a static resource and three guided prompts:

Resource	Description
`cmf://api-reference`	Static reference text covering the object hierarchy, status and phase enums, and filter syntax.

Prompt	Arguments	Description
`debug_failing_application`	environment, application	Chains `get_flinkapplication`, `list_flinkapplicationevents`, `list_flinkapplicationinstances`, and `get_flinkapplicationlogs` to investigate a failing application.
`debug_failing_statement`	environment, statement	Chains `get_flinkstatement` and `list_flinkstatementexceptions` to investigate a failing SQL statement.
`environment_health_check`	environment	Surveys all resources in an environment and their statuses for a health report.