Add MCP server

.github/workflows/integration-tests.yml

@@ -16,10 +16,10 @@ jobs:
 
     steps:
     - uses: actions/checkout@v3
-    - name: Set up JDK 11
+    - name: Set up JDK 17
       uses: actions/setup-java@v3
       with:
-        java-version: '11'
+        java-version: '17'
         distribution: 'temurin'
     - name: Build
       run: make build

.github/workflows/release.yml

@@ -10,7 +10,7 @@ jobs:
       - name: Set up Java
         uses: actions/setup-java@v3
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'adopt'
       - name: Validate Gradle wrapper
         uses: gradle/wrapper-validation-action@ccb4328a959376b642e027874838f60f8e596de3

README.md

@@ -127,6 +127,22 @@ The CLI includes some additional commands. See `!intro`.
 
 To use Hoptimator from Java code, or from anything that supports JDBC, use the `jdbc:hoptimator://` JDBC driver.
 
+## The MCP Server
+
+To use Hoptimator from an AI chat bot, agent, IDE, etc, you can use the Model Context Protocol Server. Just point your MCP configs at the server path:
+
+```
+{
+  "mcpServers": {
+    "Hoptimator": {
+      "command": "./hoptimator-mcp-server/start"
+    }
+}
+```
+
+You may need additional configuration, e.g. `JAVA_HOME`, depending on your environment.
+
+
 ## The Operator
 
 `hoptimator-operator` turns materialized views into real data pipelines. The name operator comes from the Kubernetes Operator pattern.

gradle/libs.versions.toml

@@ -20,12 +20,14 @@ flink-table-planner = "org.apache.flink:flink-table-planner_2.12:1.18.1"
 flink-table-runtime = "org.apache.flink:flink-table-runtime:1.18.1"
 gson = "com.google.code.gson:gson:2.9.0"
 jackson = "com.fasterxml.jackson.core:jackson-core:2.15.0"
+jackson-databind = "com.fasterxml.jackson.core:jackson-databind:2.15.0"
 jackson-dataformat-yaml = "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.15.0"
 javax-annotation-api = "javax.annotation:javax.annotation-api:1.3.2"
 junit = "junit:junit:4.12"
 kafka-clients = "org.apache.kafka:kafka-clients:3.2.0"
 kubernetes-client = "io.kubernetes:client-java:18.0.0"
 kubernetes-extended-client = "io.kubernetes:client-java-extended:18.0.0"
+mcp-bom = "io.modelcontextprotocol.sdk:mcp-bom:0.10.0"
 slf4j-simple = "org.slf4j:slf4j-simple:2.0.11"
 slf4j-api = "org.slf4j:slf4j-api:2.0.11"
 sqlline = "sqlline:sqlline:1.12.0"

hoptimator-mcp-server/build.gradle

@@ -0,0 +1,30 @@
+plugins {
+  id 'java'
+  id 'application'
+  id 'idea'
+}
+
+dependencies {
+  implementation project(':hoptimator-jdbc-driver')
+  implementation libs.slf4j.simple
+  implementation project(':hoptimator-demodb')
+  implementation project(':hoptimator-kafka')
+  implementation project(':hoptimator-venice')
+
+  implementation libs.gson
+  implementation libs.jackson.databind
+  implementation platform(libs.mcp.bom)
+  implementation 'io.modelcontextprotocol.sdk:mcp'
+}
+
+application {
+  mainClassName  = 'com.linkedin.hoptimator.mcp.server.HoptimatorMcpServer'
+}
+
+tasks.withType(JavaCompile).configureEach {
+  options.release = 17
+  options.compilerArgs << '-Xlint:deprecation'
+  options.compilerArgs << '-Xlint:unchecked'
+}
+
+

hoptimator-mcp-server/src/main/java/com/linkedin/hoptimator/mcp/server/HoptimatorMcpServer.java

@@ -0,0 +1,87 @@
+package com.linkedin.hoptimator.mcp.server;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Statement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.List;
+
+import com.google.gson.Gson;
+import io.modelcontextprotocol.server.McpServer;
+import io.modelcontextprotocol.server.McpSyncServer;
+import io.modelcontextprotocol.server.transport.StdioServerTransportProvider;
+
+import static io.modelcontextprotocol.server.McpServerFeatures.SyncToolSpecification;
+import static io.modelcontextprotocol.spec.McpSchema.CallToolResult;
+import static io.modelcontextprotocol.spec.McpSchema.ServerCapabilities;
+import static io.modelcontextprotocol.spec.McpSchema.Tool;
+
+
+public final class HoptimatorMcpServer {
+
+  private HoptimatorMcpServer() {
+  }
+
+  public static void main(String[] args) throws Exception {
+    Connection conn = DriverManager.getConnection("jdbc:hoptimator://fun=mysql");
+    Gson gson = new Gson();
+
+    String sqlSchema = "{\"type\" : \"object\", \"id\" : \"urn:jsonschema:Sql\","
+        + "\"properties\" : {\"sql\" : {\"type\" : \"string\"}}}";
+    StdioServerTransportProvider transportProvider = new StdioServerTransportProvider();
+    SyncToolSpecification query = new SyncToolSpecification(
+      new Tool("query", "SQL Query", sqlSchema), (x, args2) -> {
+        try (Statement stmt = conn.createStatement()) {
+          String sql = rewriteCommands((String) args2.get("sql"));
+          ResultSet rs = stmt.executeQuery(sql);
+          return new CallToolResult(gson.toJson(collect(rs)), false);
+        } catch (Exception e) {
+          return new CallToolResult("ERROR: " + e.toString(), true);
+        }
+      });
+
+    McpSyncServer server = McpServer.sync(transportProvider)
+        .serverInfo("hoptimator", "0.0.0")
+        .capabilities(ServerCapabilities.builder()
+        .tools(true)         // Enable tool support
+        .build())
+        .build();
+
+    server.addTool(query);
+    while (true) {
+      Thread.sleep(1000L);
+    }
+  }
+
+  private static String rewriteCommands(String sql) {
+    if ("show tables".equalsIgnoreCase(sql)) {
+      return "select * from \"metadata\".tables";
+    }
+    if ("show databases".equalsIgnoreCase(sql)) {
+      return "select * from \"k8s\".databases";
+    }
+    if ("show pipelines".equalsIgnoreCase(sql)) {
+      return "select * from \"k8s\".pipelines";
+    }
+    return sql;
+  }
+
+  private static List<Map<String, String>> collect(ResultSet rs) throws SQLException {
+    ResultSetMetaData meta = rs.getMetaData();
+    int n = meta.getColumnCount();
+    List<Map<String, String>> results = new ArrayList<>();
+    while (rs.next()) {
+      Map<String, String> row = new HashMap<>();
+      for (int j = 1; j <= n; j++) {
+        row.put(meta.getColumnName(j), rs.getString(j)); 
+      }
+      results.add(row);
+    }
+    return results;
+  }
+}

hoptimator-mcp-server/start

@@ -0,0 +1,9 @@
+#!/bin/sh
+
+BASEDIR="$( cd "$( dirname "$0" )" && pwd )"
+
+$BASEDIR/build/install/hoptimator-mcp-server/bin/hoptimator-mcp-server \
+  -Dorg.slf4j.simpleLogger.showThreadName=false \
+  -Dorg.slf4j.simpleLogger.showLogName=false \
+  com.linkedin.hoptimator.mcp.server.HoptimatorMcpServer
+

settings.gradle

@@ -13,6 +13,7 @@ include 'hoptimator-jdbc-driver-int'
 include 'hoptimator-k8s'
 include 'hoptimator-kafka-controller'
 include 'hoptimator-kafka'
+include 'hoptimator-mcp-server'
 include 'hoptimator-models'           // <-- marked for deletion
 include 'hoptimator-operator'
 include 'hoptimator-operator-integration'