delete-flink-statements

Instructions

Implementation Reference

• src/confluent/tools/handlers/flink/delete-flink-statement-handler.ts:46-106 (handler)

  The DeleteFlinkStatementHandler class implements the core logic for the 'delete-flink-statements' tool. It parses input arguments, ensures required parameters, makes a DELETE request to the Confluent Cloud Flink SQL API to delete the specified statement, and returns the status.

  export class DeleteFlinkStatementHandler extends BaseToolHandler {
    async handle(
      clientManager: ClientManager,
      toolArguments: Record<string, unknown> | undefined,
    ): Promise<CallToolResult> {
      const { statementName, environmentId, organizationId, baseUrl } =
        deleteFlinkStatementArguments.parse(toolArguments);
      const organization_id = getEnsuredParam(
        "FLINK_ORG_ID",
        "Organization ID is required",
        organizationId,
      );
      const environment_id = getEnsuredParam(
        "FLINK_ENV_ID",
        "Environment ID is required",
        environmentId,
      );
  
      if (baseUrl !== undefined && baseUrl !== "") {
        clientManager.setConfluentCloudFlinkEndpoint(baseUrl);
      }
      const pathBasedClient = wrapAsPathBasedClient(
        clientManager.getConfluentCloudFlinkRestClient(),
      );
      const { response, error } = await pathBasedClient[
        "/sql/v1/organizations/{organization_id}/environments/{environment_id}/statements/{statement_name}"
      ].DELETE({
        params: {
          path: {
            organization_id: organization_id,
            environment_id: environment_id,
            statement_name: statementName,
          },
        },
      });
      if (error) {
        return this.createResponse(
          `Failed to delete Flink SQL statement: ${JSON.stringify(error)}`,
          true,
        );
      }
      return this.createResponse(
        `Flink SQL Statement Deletion Status Code: ${response?.status}`,
      );
    }
    getToolConfig(): ToolConfig {
      return {
        name: ToolName.DELETE_FLINK_STATEMENTS,
        description: "Make a request to delete a statement.",
        inputSchema: deleteFlinkStatementArguments.shape,
      };
    }
  
    getRequiredEnvVars(): EnvVar[] {
      return ["FLINK_API_KEY", "FLINK_API_SECRET"];
    }
  
    isConfluentCloudOnly(): boolean {
      return true;
    }
  }

• src/confluent/tools/handlers/flink/delete-flink-statement-handler.ts:14-44 (schema)

  Zod input schema defining the parameters for the delete-flink-statements tool: baseUrl (optional Flink REST endpoint), organizationId (optional), environmentId (optional), and statementName (required, with regex validation).

  const deleteFlinkStatementArguments = z.object({
    baseUrl: z
      .string()
      .trim()
      .describe("The base URL of the Flink REST API.")
      .url()
      .default(() => env.FLINK_REST_ENDPOINT ?? "")
      .optional(),
    organizationId: z
      .string()
      .trim()
      .optional()
      .describe("The unique identifier for the organization."),
    environmentId: z
      .string()
      .trim()
      .optional()
      .describe("The unique identifier for the environment."),
    statementName: z
      .string()
      .regex(
        new RegExp(
          "[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*",
        ),
      )
      .nonempty()
      .max(100)
      .describe(
        "The user provided name of the resource, unique within this environment.",
      ),
  });

• src/confluent/tools/tool-factory.ts:50-50 (registration)

  Registration of the DeleteFlinkStatementHandler in the ToolFactory's handlers Map under the key ToolName.DELETE_FLINK_STATEMENTS.

  [ToolName.DELETE_FLINK_STATEMENTS, new DeleteFlinkStatementHandler()],

• src/confluent/tools/tool-factory.ts:15-15 (registration)

  Import of the DeleteFlinkStatementHandler class in tool-factory.ts for registration.

  import { DeleteFlinkStatementHandler } from "@src/confluent/tools/handlers/flink/delete-flink-statement-handler.js";

• src/confluent/tools/tool-name.ts:10-10 (helper)

  Enum constant defining the tool name 'delete-flink-statements'.

  DELETE_FLINK_STATEMENTS = "delete-flink-statements",