, etc.) in your responses.
+ Tool calls happen automatically in the background - only show the results to the user.
+ Present information naturally as if you performed the action yourself.
+
+ IMPORTANT SAFETY RULE: For any WRITE operations (createSource, createDestination, createPipeline),
+ you MUST ALWAYS call the tool first with userConfirmed=false to get the confirmation details,
+ then present those details to the user and ask for their explicit confirmation,
+ and ONLY after the user confirms should you call the tool again with userConfirmed=true.
+ NEVER execute write operations without user confirmation.
+ """;
+ }
+}
diff --git a/debezium-platform-conductor/src/main/java/io/debezium/platform/ai/model/CustomVertexAiAnthropicStreamingChatModel.java b/debezium-platform-conductor/src/main/java/io/debezium/platform/ai/model/CustomVertexAiAnthropicStreamingChatModel.java
new file mode 100644
index 00000000..399b64be
--- /dev/null
+++ b/debezium-platform-conductor/src/main/java/io/debezium/platform/ai/model/CustomVertexAiAnthropicStreamingChatModel.java
@@ -0,0 +1,774 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.platform.ai.model;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.security.cert.X509Certificate;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import dev.langchain4j.agent.tool.ToolSpecification;
+import dev.langchain4j.data.message.AiMessage;
+import dev.langchain4j.data.message.ChatMessage;
+import dev.langchain4j.data.message.SystemMessage;
+import dev.langchain4j.data.message.ToolExecutionResultMessage;
+import dev.langchain4j.data.message.UserMessage;
+import dev.langchain4j.model.ModelProvider;
+import dev.langchain4j.model.chat.StreamingChatModel;
+import dev.langchain4j.model.chat.listener.ChatModelListener;
+import dev.langchain4j.model.chat.request.ChatRequest;
+import dev.langchain4j.model.chat.request.ChatRequestParameters;
+import dev.langchain4j.model.chat.response.ChatResponse;
+import dev.langchain4j.model.chat.response.StreamingChatResponseHandler;
+import dev.langchain4j.model.output.FinishReason;
+import dev.langchain4j.model.output.TokenUsage;
+
+/**
+ * Custom streaming chat model for internal Vertex AI-compatible Anthropic API.
+ * This implementation works with APIs that use Vertex AI request/response format
+ * but have custom endpoints and bearer token authentication.
+ *
+ *
+ * The following code is mainly generated by Claude since this is just a custom adapter to internal model serving service
+ *
+ * @author Mario Fiore Vitale
+ */
+public class CustomVertexAiAnthropicStreamingChatModel implements StreamingChatModel {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CustomVertexAiAnthropicStreamingChatModel.class);
+ private static final String ANTHROPIC_VERSION = "vertex-2023-10-16";
+
+ // Static initializer to configure SSL to trust all certificates (for development only)
+ static {
+ try {
+ // Create a trust manager that trusts all certificates
+ TrustManager[] trustAllCerts = new TrustManager[]{
+ new X509TrustManager() {
+ public X509Certificate[] getAcceptedIssuers() {
+ return null;
+ }
+
+ public void checkClientTrusted(X509Certificate[] certs, String authType) {
+ }
+
+ public void checkServerTrusted(X509Certificate[] certs, String authType) {
+ }
+ }
+ };
+
+ // Install the all-trusting trust manager
+ SSLContext sc = SSLContext.getInstance("SSL");
+ sc.init(null, trustAllCerts, new java.security.SecureRandom());
+ HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
+
+ // Install the all-trusting host verifier
+ HttpsURLConnection.setDefaultHostnameVerifier((hostname, session) -> true);
+
+ LOG.info("SSL trust-all configured for CustomVertexAiAnthropicStreamingChatModel");
+ }
+ catch (Exception e) {
+ LOG.error("Failed to configure SSL trust-all", e);
+ }
+ }
+
+ private final String baseUrl;
+ private final String apiKey;
+ private final String modelName;
+ private final String modelType; // "sonnet" or "haiku"
+ private final Integer maxTokens;
+ private final Double temperature;
+ private final Boolean logRequests;
+ private final Boolean logResponses;
+ private final String systemPrompt;
+ private final ObjectMapper objectMapper;
+
+ private CustomVertexAiAnthropicStreamingChatModel(Builder builder) {
+ this.baseUrl = builder.baseUrl;
+ this.apiKey = builder.apiKey;
+ this.modelName = builder.modelName;
+ this.modelType = determineModelType(builder.modelName);
+ this.maxTokens = builder.maxTokens != null ? builder.maxTokens : 4096;
+ this.temperature = builder.temperature;
+ this.logRequests = builder.logRequests != null ? builder.logRequests : false;
+ this.logResponses = builder.logResponses != null ? builder.logResponses : false;
+ this.systemPrompt = builder.systemPrompt;
+
+ // Configure ObjectMapper to handle unknown types gracefully
+ this.objectMapper = new ObjectMapper();
+ this.objectMapper.configure(com.fasterxml.jackson.databind.SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+ }
+
+ private String determineModelType(String modelName) {
+ if (modelName.toLowerCase().contains("haiku")) {
+ return "haiku";
+ }
+ else if (modelName.toLowerCase().contains("sonnet")) {
+ return "sonnet";
+ }
+ else {
+ // Default to sonnet
+ return "sonnet";
+ }
+ }
+
+ @Override
+ public void doChat(ChatRequest chatRequest, StreamingChatResponseHandler handler) {
+ try {
+ // Build the endpoint URL
+ String endpoint = String.format("%s/%s/models/%s:streamRawPredict",
+ baseUrl, modelType, modelName);
+
+ if (logRequests) {
+ LOG.info("Calling endpoint: {}", endpoint);
+ }
+
+ // Convert ChatRequest to Vertex AI format
+ Map requestBody = buildRequestBody(chatRequest);
+
+ if (logRequests) {
+ try {
+ LOG.info("Request body: {}", objectMapper.writeValueAsString(requestBody));
+ }
+ catch (Exception e) {
+ LOG.warn("Could not serialize request body for logging: {}", e.getMessage());
+ LOG.info("Request endpoint: {}", endpoint);
+ }
+ }
+
+ // Make HTTP request
+ makeStreamingRequest(endpoint, requestBody, handler);
+
+ }
+ catch (Exception e) {
+ LOG.error("Error during chat", e);
+ try {
+ handler.onError(e);
+ }
+ catch (Exception userException) {
+ LOG.warn("User's onError handler threw an exception, ignoring", userException);
+ }
+ }
+ }
+
+ private Map buildRequestBody(ChatRequest chatRequest) {
+ Map body = new HashMap<>();
+ body.put("anthropic_version", ANTHROPIC_VERSION);
+
+ // Add system message if configured
+ if (systemPrompt != null && !systemPrompt.isEmpty()) {
+ body.put("system", systemPrompt);
+ }
+
+ // Convert messages
+ List> messages = convertMessages(chatRequest.messages());
+ body.put("messages", messages);
+
+ // Set max_tokens
+ ChatRequestParameters params = chatRequest.parameters();
+ Integer tokens = params.maxOutputTokens() != null ? params.maxOutputTokens() : maxTokens;
+ body.put("max_tokens", tokens);
+
+ // Optional parameters
+ if (temperature != null) {
+ body.put("temperature", temperature);
+ }
+
+ // Add tools if present
+ if (params.toolSpecifications() != null && !params.toolSpecifications().isEmpty()) {
+ List> tools = convertTools(params.toolSpecifications());
+ body.put("tools", tools);
+ }
+
+ return body;
+ }
+
+ private List> convertMessages(List messages) {
+ List> result = new ArrayList<>();
+
+ for (ChatMessage message : messages) {
+ Map msg = new HashMap<>();
+
+ // Determine role using instanceof checks
+ String role = switch (message) {
+ case UserMessage userMessage -> "user";
+ case AiMessage aiMessage -> "assistant";
+ case SystemMessage systemMessage -> "system";
+ case ToolExecutionResultMessage toolExecutionResultMessage -> "user"; // Tool results come back as user messages in Anthropic format
+
+ case null, default ->
+ // Default to user for unknown types
+ "user";
+ };
+
+ if ("system".equals(role)) {
+ // Vertex AI handles system messages differently
+ // Skip for now, could be added to request root if needed
+ continue;
+ }
+
+ msg.put("role", role);
+
+ // Build content array
+ List> content = new ArrayList<>();
+
+ if (message instanceof ToolExecutionResultMessage toolResult) {
+ // Tool result - special format for Anthropic
+ Map toolResultContent = new HashMap<>();
+ toolResultContent.put("type", "tool_result");
+ toolResultContent.put("tool_use_id", toolResult.id());
+ toolResultContent.put("content", toolResult.text() != null ? toolResult.text() : "");
+ content.add(toolResultContent);
+ }
+ else if (message instanceof AiMessage aiMessage) {
+ // AI message - might have both text and tool calls
+
+ // Add text content if present
+ if (aiMessage.text() != null && !aiMessage.text().isEmpty()) {
+ Map textContent = new HashMap<>();
+ textContent.put("type", "text");
+ textContent.put("text", aiMessage.text());
+ content.add(textContent);
+ }
+
+ // Add tool calls if present
+ if (aiMessage.hasToolExecutionRequests()) {
+ for (dev.langchain4j.agent.tool.ToolExecutionRequest toolRequest : aiMessage.toolExecutionRequests()) {
+ Map toolUseContent = new HashMap<>();
+ toolUseContent.put("type", "tool_use");
+ toolUseContent.put("id", toolRequest.id());
+ toolUseContent.put("name", toolRequest.name());
+
+ // Parse arguments JSON back to object
+ try {
+ @SuppressWarnings("unchecked")
+ Map inputMap = objectMapper.readValue(toolRequest.arguments(), Map.class);
+ toolUseContent.put("input", inputMap);
+ }
+ catch (Exception e) {
+ LOG.warn("Failed to parse tool arguments: {}", e.getMessage());
+ toolUseContent.put("input", new HashMap<>());
+ }
+
+ content.add(toolUseContent);
+ }
+ }
+ }
+ else {
+ // Regular user message
+ Map textContent = new HashMap<>();
+ textContent.put("type", "text");
+
+ String text = "";
+ if (message instanceof UserMessage) {
+ text = ((UserMessage) message).singleText();
+ }
+
+ textContent.put("text", text != null ? text : "");
+ content.add(textContent);
+ }
+
+ msg.put("content", content);
+ result.add(msg);
+ }
+
+ return result;
+ }
+
+ private List> convertTools(List toolSpecs) {
+ return toolSpecs.stream()
+ .map(tool -> {
+ Map toolMap = new HashMap<>();
+ toolMap.put("name", tool.name());
+ toolMap.put("description", tool.description());
+
+ // Build input_schema
+ Map inputSchema = new HashMap<>();
+ inputSchema.put("type", "object");
+
+ if (tool.parameters() != null) {
+ try {
+ // Use reflection to call properties() method on JsonObjectSchema
+ java.lang.reflect.Method propertiesMethod = tool.parameters().getClass().getMethod("properties");
+ @SuppressWarnings("unchecked")
+ Map schemaProperties = (Map) propertiesMethod.invoke(tool.parameters());
+
+ if (schemaProperties != null && !schemaProperties.isEmpty()) {
+ // Convert each schema element to a map
+ Map convertedProperties = new HashMap<>();
+ for (Map.Entry entry : schemaProperties.entrySet()) {
+ Object schemaElement = entry.getValue();
+ Map propertySchema = convertSchemaElement(schemaElement);
+ convertedProperties.put(entry.getKey(), propertySchema);
+ }
+ inputSchema.put("properties", convertedProperties);
+ }
+ else {
+ inputSchema.put("properties", new HashMap<>());
+ }
+
+ // Try to get required fields using reflection
+ try {
+ java.lang.reflect.Method requiredMethod = tool.parameters().getClass().getMethod("required");
+ Object requiredFields = requiredMethod.invoke(tool.parameters());
+ if (requiredFields != null) {
+ // Handle both List and Set (convert to List for JSON)
+ if (requiredFields instanceof List) {
+ @SuppressWarnings("unchecked")
+ List requiredList = (List) requiredFields;
+ if (!requiredList.isEmpty()) {
+ inputSchema.put("required", requiredList);
+ }
+ }
+ else if (requiredFields instanceof Set) {
+ @SuppressWarnings("unchecked")
+ Set requiredSet = (Set) requiredFields;
+ if (!requiredSet.isEmpty()) {
+ inputSchema.put("required", new ArrayList<>(requiredSet));
+ }
+ }
+ }
+ }
+ catch (NoSuchMethodException e) {
+ // required() method not available, skip
+ }
+
+ }
+ catch (Exception e) {
+ LOG.warn("Failed to convert tool parameters for {}: {}", tool.name(), e.getMessage());
+ // Use basic empty properties
+ inputSchema.put("properties", new HashMap<>());
+ }
+ }
+ else {
+ // No parameters - empty schema
+ inputSchema.put("properties", new HashMap<>());
+ }
+
+ toolMap.put("input_schema", inputSchema);
+ return toolMap;
+ })
+ .collect(Collectors.toList());
+ }
+
+ private Map convertSchemaElement(Object schemaElement) {
+ Map schema = new HashMap<>();
+
+ if (schemaElement == null) {
+ return schema;
+ }
+
+ String typeName = schemaElement.getClass().getSimpleName();
+
+ try {
+ // Try to get type using reflection
+ try {
+ java.lang.reflect.Method typeMethod = schemaElement.getClass().getMethod("type");
+ Object typeObj = typeMethod.invoke(schemaElement);
+ if (typeObj != null) {
+ // Handle enum types (e.g., JsonSchemaType.STRING)
+ String typeStr = typeObj.toString();
+ schema.put("type", typeStr.toLowerCase());
+ }
+ }
+ catch (NoSuchMethodException e) {
+ // Some schema elements might not have a type() method
+ // Infer from class name
+ if (typeName.contains("String")) {
+ schema.put("type", "string");
+ }
+ else if (typeName.contains("Integer")) {
+ schema.put("type", "integer");
+ }
+ else if (typeName.contains("Number")) {
+ schema.put("type", "number");
+ }
+ else if (typeName.contains("Boolean")) {
+ schema.put("type", "boolean");
+ }
+ else if (typeName.contains("Array")) {
+ schema.put("type", "array");
+ }
+ else if (typeName.contains("Object")) {
+ schema.put("type", "object");
+ }
+ }
+
+ // Try to get description
+ try {
+ java.lang.reflect.Method descMethod = schemaElement.getClass().getMethod("description");
+ String description = (String) descMethod.invoke(schemaElement);
+ if (description != null && !description.isEmpty()) {
+ schema.put("description", description);
+ }
+ }
+ catch (NoSuchMethodException e) {
+ // description not available
+ }
+
+ // For nested objects, recursively convert properties
+ if (typeName.equals("JsonObjectSchema")) {
+ try {
+ java.lang.reflect.Method propertiesMethod = schemaElement.getClass().getMethod("properties");
+ @SuppressWarnings("unchecked")
+ Map nestedProperties = (Map) propertiesMethod.invoke(schemaElement);
+
+ if (nestedProperties != null && !nestedProperties.isEmpty()) {
+ Map convertedNested = new HashMap<>();
+ for (Map.Entry entry : nestedProperties.entrySet()) {
+ convertedNested.put(entry.getKey(), convertSchemaElement(entry.getValue()));
+ }
+ schema.put("properties", convertedNested);
+ }
+
+ // Try to get required fields from nested object
+ try {
+ java.lang.reflect.Method requiredMethod = schemaElement.getClass().getMethod("required");
+ Object requiredFields = requiredMethod.invoke(schemaElement);
+ if (requiredFields != null) {
+ if (requiredFields instanceof List) {
+ @SuppressWarnings("unchecked")
+ List requiredList = (List) requiredFields;
+ if (!requiredList.isEmpty()) {
+ schema.put("required", requiredList);
+ }
+ }
+ else if (requiredFields instanceof Set) {
+ @SuppressWarnings("unchecked")
+ Set requiredSet = (Set) requiredFields;
+ if (!requiredSet.isEmpty()) {
+ schema.put("required", new ArrayList<>(requiredSet));
+ }
+ }
+ }
+ }
+ catch (NoSuchMethodException e) {
+ // required() method not available, skip
+ }
+
+ }
+ catch (Exception e) {
+ LOG.debug("Could not extract nested properties from {}: {}", typeName, e.getMessage());
+ }
+ }
+
+ // For arrays, get items schema
+ if (typeName.equals("JsonArraySchema")) {
+ try {
+ java.lang.reflect.Method itemsMethod = schemaElement.getClass().getMethod("items");
+ Object items = itemsMethod.invoke(schemaElement);
+ if (items != null) {
+ schema.put("items", convertSchemaElement(items));
+ }
+ }
+ catch (Exception e) {
+ LOG.debug("Could not extract array items from {}: {}", typeName, e.getMessage());
+ }
+ }
+
+ // For enums, get enum values
+ if (typeName.equals("JsonEnumSchema")) {
+ try {
+ java.lang.reflect.Method enumValuesMethod = schemaElement.getClass().getMethod("enumValues");
+ @SuppressWarnings("unchecked")
+ List enumValues = (List) enumValuesMethod.invoke(schemaElement);
+ if (enumValues != null && !enumValues.isEmpty()) {
+ schema.put("enum", enumValues);
+ }
+ }
+ catch (Exception e) {
+ LOG.debug("Could not extract enum values from {}: {}", typeName, e.getMessage());
+ }
+ }
+
+ }
+ catch (Exception e) {
+ LOG.debug("Error converting schema element {}: {}", typeName, e.getMessage());
+ }
+
+ return schema;
+ }
+
+ private void makeStreamingRequest(String endpoint, Map requestBody,
+ StreamingChatResponseHandler handler)
+ throws IOException {
+
+ URL url = URI.create(endpoint).toURL();
+ HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+
+ try {
+ // Configure connection
+ connection.setRequestMethod("POST");
+ connection.setRequestProperty("Content-Type", "application/json");
+ connection.setRequestProperty("Authorization", "Bearer " + apiKey);
+ connection.setDoOutput(true);
+ connection.setConnectTimeout(30000);
+ connection.setReadTimeout(300000); // 5 minutes for streaming
+
+ // Write request body
+ String jsonBody = objectMapper.writeValueAsString(requestBody);
+ try (OutputStream os = connection.getOutputStream()) {
+ byte[] input = jsonBody.getBytes(StandardCharsets.UTF_8);
+ os.write(input, 0, input.length);
+ }
+
+ // Check response code
+ int responseCode = connection.getResponseCode();
+ if (responseCode != 200) {
+ String errorResponse = readErrorStream(connection);
+ throw new IOException("HTTP " + responseCode + ": " + errorResponse);
+ }
+
+ // Read streaming response
+ handleStreamingResponse(connection, handler);
+
+ }
+ finally {
+ connection.disconnect();
+ }
+ }
+
+ private String readErrorStream(HttpURLConnection connection) {
+ try (BufferedReader br = new BufferedReader(
+ new InputStreamReader(connection.getErrorStream(), StandardCharsets.UTF_8))) {
+ return br.lines().collect(Collectors.joining("\n"));
+ }
+ catch (Exception e) {
+ return "Unable to read error stream";
+ }
+ }
+
+ private void handleStreamingResponse(HttpURLConnection connection, StreamingChatResponseHandler handler)
+ throws IOException {
+ StringBuilder fullText = new StringBuilder();
+ List toolExecutionRequests = new ArrayList<>();
+
+ try (BufferedReader reader = new BufferedReader(
+ new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) {
+
+ // The response might be a single JSON object (non-streaming) or streamed
+ // Read the entire response first
+ String responseText = reader.lines().collect(Collectors.joining("\n"));
+
+ if (logResponses) {
+ LOG.info("Response: {}", responseText);
+ }
+
+ // Parse response
+ JsonNode responseNode = objectMapper.readTree(responseText);
+
+ // Extract content
+ if (responseNode.has("content")) {
+ JsonNode contentArray = responseNode.get("content");
+ for (JsonNode contentItem : contentArray) {
+ String contentType = contentItem.has("type") ? contentItem.get("type").asText() : "";
+
+ if ("text".equals(contentType)) {
+ String text = contentItem.get("text").asText();
+ fullText.append(text);
+
+ // Send as partial response (simulate streaming by sending chunks)
+ sendInChunks(text, handler);
+ }
+ else if ("tool_use".equals(contentType)) {
+ // Extract tool call
+ String toolId = contentItem.has("id") ? contentItem.get("id").asText() : "";
+ String toolName = contentItem.has("name") ? contentItem.get("name").asText() : "";
+ JsonNode inputNode = contentItem.has("input") ? contentItem.get("input") : null;
+
+ // Convert input to JSON string
+ String arguments = inputNode != null ? objectMapper.writeValueAsString(inputNode) : "{}";
+
+ dev.langchain4j.agent.tool.ToolExecutionRequest toolRequest = dev.langchain4j.agent.tool.ToolExecutionRequest.builder()
+ .id(toolId)
+ .name(toolName)
+ .arguments(arguments)
+ .build();
+
+ toolExecutionRequests.add(toolRequest);
+ LOG.info("Tool execution requested: {} ({})", toolName, toolId);
+ }
+ }
+ }
+
+ // Send tool execution requests if present
+ if (!toolExecutionRequests.isEmpty()) {
+ for (int i = 0; i < toolExecutionRequests.size(); i++) {
+ dev.langchain4j.agent.tool.ToolExecutionRequest toolRequest = toolExecutionRequests.get(i);
+ dev.langchain4j.model.chat.response.CompleteToolCall completeToolCall = new dev.langchain4j.model.chat.response.CompleteToolCall(i, toolRequest);
+ handler.onCompleteToolCall(completeToolCall);
+ }
+ }
+
+ // Build complete response
+ TokenUsage tokenUsage = extractTokenUsage(responseNode);
+
+ // Build AI message with text and tool execution requests
+ AiMessage.Builder aiMessageBuilder = AiMessage.builder();
+ if (!fullText.isEmpty()) {
+ aiMessageBuilder.text(fullText.toString());
+ }
+ if (!toolExecutionRequests.isEmpty()) {
+ aiMessageBuilder.toolExecutionRequests(toolExecutionRequests);
+ }
+
+ ChatResponse chatResponse = ChatResponse.builder()
+ .aiMessage(aiMessageBuilder.build())
+ .tokenUsage(tokenUsage)
+ .finishReason(extractFinishReason(responseNode))
+ .build();
+
+ handler.onCompleteResponse(chatResponse);
+
+ }
+ catch (Exception e) {
+ LOG.error("Error processing streaming response", e);
+ handler.onError(e);
+ }
+ }
+
+ private void sendInChunks(String text, StreamingChatResponseHandler handler) {
+ // Split text into small chunks to simulate streaming
+ int chunkSize = 5;
+ for (int i = 0; i < text.length(); i += chunkSize) {
+ int end = Math.min(i + chunkSize, text.length());
+ String chunk = text.substring(i, end);
+ try {
+ handler.onPartialResponse(chunk);
+ Thread.sleep(10); // Small delay to simulate streaming
+ }
+ catch (Exception e) {
+ LOG.warn("Error sending chunk", e);
+ }
+ }
+ }
+
+ private TokenUsage extractTokenUsage(JsonNode responseNode) {
+ if (responseNode.has("usage")) {
+ JsonNode usage = responseNode.get("usage");
+ int inputTokens = usage.has("input_tokens") ? usage.get("input_tokens").asInt() : 0;
+ int outputTokens = usage.has("output_tokens") ? usage.get("output_tokens").asInt() : 0;
+ return new TokenUsage(inputTokens, outputTokens);
+ }
+ return null;
+ }
+
+ private FinishReason extractFinishReason(JsonNode responseNode) {
+ if (responseNode.has("stop_reason")) {
+ String stopReason = responseNode.get("stop_reason").asText();
+ return switch (stopReason) {
+ case "end_turn" -> FinishReason.STOP;
+ case "max_tokens" -> FinishReason.LENGTH;
+ case "stop_sequence" -> FinishReason.STOP;
+ case "tool_use" -> FinishReason.TOOL_EXECUTION;
+ case null, default -> FinishReason.OTHER;
+ };
+ }
+ return FinishReason.STOP;
+ }
+
+ @Override
+ public List listeners() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public ModelProvider provider() {
+ return ModelProvider.GOOGLE_VERTEX_AI_ANTHROPIC;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ private String baseUrl;
+ private String apiKey;
+ private String modelName;
+ private Integer maxTokens;
+ private Double temperature;
+ private Boolean logRequests;
+ private Boolean logResponses;
+ private String systemPrompt;
+
+ public Builder baseUrl(String baseUrl) {
+ this.baseUrl = baseUrl;
+ return this;
+ }
+
+ public Builder apiKey(String apiKey) {
+ this.apiKey = apiKey;
+ return this;
+ }
+
+ public Builder modelName(String modelName) {
+ this.modelName = modelName;
+ return this;
+ }
+
+ public Builder maxTokens(Integer maxTokens) {
+ this.maxTokens = maxTokens;
+ return this;
+ }
+
+ public Builder temperature(Double temperature) {
+ this.temperature = temperature;
+ return this;
+ }
+
+ public Builder logRequests(Boolean logRequests) {
+ this.logRequests = logRequests;
+ return this;
+ }
+
+ public Builder logResponses(Boolean logResponses) {
+ this.logResponses = logResponses;
+ return this;
+ }
+
+ /**
+ * Set a system prompt to control Claude's behavior and output format.
+ * Common use cases:
+ * - HTML formatting: "Format responses using HTML tags (strong, em, ul, li, p). No markdown."
+ * - Plain text: "Reply in plain text without any formatting or markdown."
+ * - Custom instructions: "You are a helpful assistant specialized in..."
+ */
+ public Builder systemPrompt(String systemPrompt) {
+ this.systemPrompt = systemPrompt;
+ return this;
+ }
+
+ public CustomVertexAiAnthropicStreamingChatModel build() {
+ if (baseUrl == null || baseUrl.trim().isEmpty()) {
+ throw new IllegalArgumentException("baseUrl is required");
+ }
+ if (apiKey == null || apiKey.trim().isEmpty()) {
+ throw new IllegalArgumentException("apiKey is required");
+ }
+ if (modelName == null || modelName.trim().isEmpty()) {
+ throw new IllegalArgumentException("modelName is required");
+ }
+ return new CustomVertexAiAnthropicStreamingChatModel(this);
+ }
+ }
+}
diff --git a/debezium-platform-conductor/src/main/resources/application.yml b/debezium-platform-conductor/src/main/resources/application.yml
index 33c2746f..e73f8ba1 100644
--- a/debezium-platform-conductor/src/main/resources/application.yml
+++ b/debezium-platform-conductor/src/main/resources/application.yml
@@ -45,6 +45,10 @@ destinations:
connection:
timeout: 60
+# Context7 MCP integration
+context7:
+ api-key: ${CONTEXT7_API_KEY:fake}
+
quarkus:
hibernate-orm:
mapping:
@@ -73,6 +77,36 @@ quarkus:
migrate-at-start: true
swagger-ui:
always-include: true
+ # LangChain4j Anthropic configuration
+ langchain4j:
+ anthropic:
+ api-key: ${CHAT_OPENAI_API_KEY:fake}
+ base-url: ${CHAT_OPENAI_BASE_URL:fake}
+ chat-model:
+ model-name: ${CHAT_OPENAI_MODEL_NAME:fake}
+ log-requests: ${CHAT_LOG_REQUESTS:false}
+ log-responses: ${CHAT_LOG_RESPONSES:false}
+ # MCP client configurations
+ mcp:
+ debezium-platform:
+ transport-type: streamable-http
+ url: http://localhost:8081/mcp
+ log-requests: ${MCP_LOG_REQUESTS:false}
+ log-responses: ${MCP_LOG_RESPONSES:false}
+ context7:
+ transport-type: streamable-http
+ url: https://mcp.context7.com/mcp
+ log-requests: ${MCP_LOG_REQUESTS:false}
+ log-responses: ${MCP_LOG_RESPONSES:false}
+ # MCP server schema generator configuration
+ mcp:
+ server:
+ schema-generator:
+ jakarta-validation:
+ enabled: true
+ not-nullable-field-is-required: true
+ not-nullable-method-is-required: true
+ include-pattern-expressions: true
log:
min-level: TRACE
level: INFO
diff --git a/debezium-platform-stage/package.json b/debezium-platform-stage/package.json
index 91e28e07..34b462f7 100644
--- a/debezium-platform-stage/package.json
+++ b/debezium-platform-stage/package.json
@@ -23,6 +23,7 @@
},
"dependencies": {
"@hookform/resolvers": "^5.2.2",
+ "@patternfly/chatbot": "^6.4.1",
"@patternfly/patternfly": "^6.3.0",
"@patternfly/react-charts": "^8.4.0",
"@patternfly/react-code-editor": "^6.4.0",
@@ -84,13 +85,13 @@
"chromatic": "^13.3.3",
"cypress": "^15.7.1",
"eslint": "^8.57.0",
- "start-server-and-test": "^2.1.3",
"eslint-plugin-react": "^7.35.0",
"eslint-plugin-react-hooks": "^5.2.0",
"eslint-plugin-react-refresh": "^0.4.6",
"eslint-plugin-storybook": "^9.0.7",
"jsdom": "^24.1.0",
"msw": "^2.11.6",
+ "start-server-and-test": "^2.1.3",
"storybook": "^8.0.0",
"typescript": "^5.2.2",
"vite": "^5.4.21",
diff --git a/debezium-platform-stage/src/App.tsx b/debezium-platform-stage/src/App.tsx
index f66bae29..08462485 100644
--- a/debezium-platform-stage/src/App.tsx
+++ b/debezium-platform-stage/src/App.tsx
@@ -5,16 +5,19 @@ import { AppLayout } from "./appLayout/AppLayout";
import { AppRoutes } from "./AppRoutes";
import { AppContextProvider } from "./appLayout/AppContext";
import { NotificationProvider } from "./appLayout/AppNotificationContext";
+import { ChatbotProvider } from "./components/AIChatbot";
function App() {
return (
-
-
-
-
+
+
+
+
+
+
);
diff --git a/debezium-platform-stage/src/appLayout/AppLayout.tsx b/debezium-platform-stage/src/appLayout/AppLayout.tsx
index ddb2c0c0..0a3313fa 100644
--- a/debezium-platform-stage/src/appLayout/AppLayout.tsx
+++ b/debezium-platform-stage/src/appLayout/AppLayout.tsx
@@ -19,6 +19,7 @@ import {
selectedDestinationAtom,
selectedTransformAtom,
} from "../pages/Pipeline/PipelineDesigner";
+import { AIChatbot } from "../components/AIChatbot";
interface IAppLayout {
children: React.ReactNode;
@@ -159,6 +160,7 @@ const AppLayout: React.FunctionComponent = ({ children }) => {
>
{alerts.slice(0, 3)}
+
);
};
diff --git a/debezium-platform-stage/src/components/AIChatbot/AIChatbot.css b/debezium-platform-stage/src/components/AIChatbot/AIChatbot.css
new file mode 100644
index 00000000..ed3a0e97
--- /dev/null
+++ b/debezium-platform-stage/src/components/AIChatbot/AIChatbot.css
@@ -0,0 +1,328 @@
+/* Chatbot Toggle Button */
+.chatbot-toggle-button {
+ position: fixed;
+ bottom: 24px;
+ right: 24px;
+ z-index: 9999;
+}
+
+.chatbot-fab {
+ box-shadow: 0 4px 8px rgba(0, 0, 0, 0.2);
+ border-radius: 24px;
+ padding: 12px 20px;
+ font-size: 14px;
+ font-weight: 600;
+}
+
+.chatbot-fab:hover {
+ box-shadow: 0 6px 12px rgba(0, 0, 0, 0.3);
+ transform: translateY(-2px);
+ transition: all 0.2s ease;
+}
+
+/* Chatbot Container */
+.chatbot-container {
+ position: fixed;
+ bottom: 24px;
+ right: 24px;
+ z-index: 9999;
+ width: 380px;
+ max-height: 600px;
+ animation: slideUp 0.3s ease-out;
+}
+
+@keyframes slideUp {
+ from {
+ opacity: 0;
+ transform: translateY(20px);
+ }
+ to {
+ opacity: 1;
+ transform: translateY(0);
+ }
+}
+
+.chatbot-card {
+ box-shadow: 0 8px 24px rgba(0, 0, 0, 0.15);
+ height: 600px;
+ display: flex;
+ flex-direction: column;
+}
+
+.chatbot-card-body {
+ padding: 0;
+ height: 100%;
+ display: flex;
+ flex-direction: column;
+}
+
+/* Header */
+.chatbot-header {
+ padding: 16px;
+ border-bottom: 1px solid var(--pf-v6-global--BorderColor--100);
+ background: linear-gradient(135deg, #0066cc 0%, #004080 100%);
+ color: black;
+ border-radius: 8px 8px 0 0;
+}
+
+.chatbot-header .pf-v6-c-title {
+ color: black;
+ margin: 0;
+}
+
+.chatbot-subtitle-container {
+ display: flex;
+ align-items: center;
+ gap: 12px;
+}
+
+.chatbot-subtitle {
+ color: rgba(255, 255, 255, 0.8);
+ font-size: 12px;
+}
+
+.chatbot-status {
+ display: inline-flex;
+ align-items: center;
+ gap: 6px;
+ font-size: 11px;
+ padding: 2px 8px;
+ border-radius: 10px;
+ font-weight: 600;
+ text-transform: uppercase;
+ letter-spacing: 0.5px;
+}
+
+.chatbot-status::before {
+ content: '';
+ display: inline-block;
+ width: 6px;
+ height: 6px;
+ border-radius: 50%;
+}
+
+.chatbot-status-connected {
+ background-color: rgba(146, 212, 0, 0.2);
+ color: #92d400;
+}
+
+.chatbot-status-connected::before {
+ background-color: #92d400;
+ animation: pulse 2s ease-in-out infinite;
+}
+
+.chatbot-status-connecting {
+ background-color: rgba(240, 171, 0, 0.2);
+ color: #f0ab00;
+}
+
+.chatbot-status-connecting::before {
+ background-color: #f0ab00;
+ animation: blink 1s ease-in-out infinite;
+}
+
+.chatbot-status-error {
+ background-color: rgba(201, 25, 11, 0.2);
+ color: #c9190b;
+}
+
+.chatbot-status-error::before {
+ background-color: #c9190b;
+}
+
+@keyframes pulse {
+ 0%, 100% {
+ opacity: 1;
+ }
+ 50% {
+ opacity: 0.5;
+ }
+}
+
+@keyframes blink {
+ 0%, 100% {
+ opacity: 1;
+ }
+ 50% {
+ opacity: 0.3;
+ }
+}
+
+.chatbot-icon {
+ color: white;
+}
+
+.chatbot-header .pf-v6-c-button {
+ color: black;
+}
+
+.chatbot-header .pf-v6-c-button:hover {
+ background-color: rgba(255, 255, 255, 0.1);
+}
+
+/* Messages Container */
+.chatbot-messages {
+ padding: 16px;
+ overflow-y: auto;
+ flex: 1;
+ background-color: var(--pf-v6-global--BackgroundColor--200);
+ min-height: 0;
+}
+
+.chatbot-messages::-webkit-scrollbar {
+ width: 6px;
+}
+
+.chatbot-messages::-webkit-scrollbar-track {
+ background: transparent;
+}
+
+.chatbot-messages::-webkit-scrollbar-thumb {
+ background: var(--pf-v6-global--BorderColor--200);
+ border-radius: 3px;
+}
+
+.chatbot-messages::-webkit-scrollbar-thumb:hover {
+ background: var(--pf-v6-global--BorderColor--300);
+}
+
+/* Message Bubbles */
+.chatbot-message {
+ padding: 12px;
+ border-radius: 12px;
+ max-width: 85%;
+ animation: fadeIn 0.3s ease-out;
+}
+
+@keyframes fadeIn {
+ from {
+ opacity: 0;
+ transform: translateY(10px);
+ }
+ to {
+ opacity: 1;
+ transform: translateY(0);
+ }
+}
+
+.chatbot-message-bot {
+ background-color: white;
+ margin-right: auto;
+ border: 1px solid var(--pf-v6-global--BorderColor--100);
+ box-shadow: 0 1px 2px rgba(0, 0, 0, 0.05);
+}
+
+.chatbot-message-user {
+ background-color: var(--pf-v6-global--primary-color--100);
+ color: black;
+ margin-left: auto;
+ box-shadow: 0 1px 3px rgba(0, 102, 204, 0.3);
+}
+
+.chatbot-message-header {
+ display: flex;
+ justify-content: space-between;
+ margin-bottom: 6px;
+ font-size: 11px;
+ opacity: 0.8;
+}
+
+.chatbot-message-role {
+ font-weight: 600;
+ text-transform: uppercase;
+ letter-spacing: 0.5px;
+}
+
+.chatbot-message-user .chatbot-message-role,
+.chatbot-message-user .chatbot-message-time {
+ color: rgba(255, 255, 255, 0.9);
+}
+
+.chatbot-message-content {
+ line-height: 1.5;
+ white-space: pre-wrap;
+ word-wrap: break-word;
+ font-size: 14px;
+}
+
+/* Typing Indicator */
+.chatbot-typing-indicator {
+ display: flex;
+ align-items: center;
+ gap: 8px;
+ padding: 4px 0;
+}
+
+/* Input Container */
+.chatbot-input-container {
+ padding: 16px;
+ border-top: 1px solid var(--pf-v6-global--BorderColor--100);
+ background-color: white;
+ border-radius: 0 0 8px 8px;
+}
+
+.chatbot-input-container .pf-v6-c-text-input-group {
+ width: 100%;
+}
+
+.chatbot-input-container .pf-v6-c-form-control,
+.chatbot-input-container .pf-v6-c-text-input-group__text-input,
+.chatbot-input-container input[type="text"] {
+ color: var(--pf-v6-global--Color--100) !important;
+ background-color: white !important;
+ border: 1px solid var(--pf-v6-global--BorderColor--100);
+ font-size: 14px;
+ padding: 8px 12px;
+}
+
+.chatbot-input-container .pf-v6-c-form-control:focus,
+.chatbot-input-container .pf-v6-c-text-input-group__text-input:focus,
+.chatbot-input-container input[type="text"]:focus {
+ color: var(--pf-v6-global--Color--100) !important;
+ border-color: var(--pf-v6-global--primary-color--100);
+ outline: none;
+ box-shadow: 0 0 0 1px var(--pf-v6-global--primary-color--100);
+}
+
+.chatbot-input-container .pf-v6-c-form-control::placeholder,
+.chatbot-input-container input[type="text"]::placeholder {
+ color: var(--pf-v6-global--Color--200);
+ opacity: 1;
+}
+
+.chatbot-input-container .pf-v6-c-button {
+ height: 40px;
+ min-width: 40px;
+}
+
+/* Responsive Design */
+@media (max-width: 768px) {
+ .chatbot-container {
+ width: calc(100vw - 32px);
+ right: 16px;
+ bottom: 16px;
+ max-height: calc(100vh - 100px);
+ }
+
+ .chatbot-card {
+ height: calc(100vh - 100px);
+ }
+
+ .chatbot-toggle-button {
+ right: 16px;
+ bottom: 16px;
+ }
+}
+
+/* Accessibility */
+.chatbot-message:focus {
+ outline: 2px solid var(--pf-v6-global--primary-color--100);
+ outline-offset: 2px;
+}
+
+/* Smooth Transitions */
+.chatbot-card,
+.chatbot-message,
+.chatbot-fab {
+ transition: all 0.2s ease;
+}
diff --git a/debezium-platform-stage/src/components/AIChatbot/AIChatbot.tsx b/debezium-platform-stage/src/components/AIChatbot/AIChatbot.tsx
new file mode 100644
index 00000000..f9cd19e7
--- /dev/null
+++ b/debezium-platform-stage/src/components/AIChatbot/AIChatbot.tsx
@@ -0,0 +1,214 @@
+import React, { useState, useRef, useEffect } from 'react';
+import {
+ Button,
+ TextInput,
+ Stack,
+ StackItem,
+ Card,
+ CardBody,
+ Title,
+ Split,
+ SplitItem,
+ Icon,
+ Spinner,
+} from '@patternfly/react-core';
+import {
+ CommentIcon,
+ TimesIcon,
+ PaperPlaneIcon,
+ RobotIcon,
+ TrashIcon,
+} from '@patternfly/react-icons';
+import { useChatbot } from './AIChatbotContext';
+import './AIChatbot.css';
+
+const AIChatbot: React.FC = () => {
+ const { isOpen, messages, isLoading, connectionStatus, toggleChatbot, sendMessage, clearMessages } = useChatbot();
+ const [inputValue, setInputValue] = useState('');
+ const messagesEndRef = useRef(null);
+
+ const scrollToBottom = () => {
+ messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
+ };
+
+ useEffect(() => {
+ scrollToBottom();
+ }, [messages]);
+
+ const handleSendMessage = async () => {
+ if (inputValue.trim() && connectionStatus === 'connected') {
+ await sendMessage(inputValue);
+ setInputValue('');
+ }
+ };
+
+ const getConnectionStatusText = () => {
+ switch (connectionStatus) {
+ case 'connecting':
+ return 'Connecting...';
+ case 'connected':
+ return 'Connected';
+ case 'disconnected':
+ return 'Disconnected';
+ case 'error':
+ return 'Connection Error';
+ default:
+ return '';
+ }
+ };
+
+ const getConnectionStatusClass = () => {
+ switch (connectionStatus) {
+ case 'connecting':
+ return 'chatbot-status-connecting';
+ case 'connected':
+ return 'chatbot-status-connected';
+ case 'disconnected':
+ case 'error':
+ return 'chatbot-status-error';
+ default:
+ return '';
+ }
+ };
+
+ const handleKeyPress = (e: React.KeyboardEvent) => {
+ if (e.key === 'Enter' && !e.shiftKey) {
+ e.preventDefault();
+ handleSendMessage();
+ }
+ };
+
+ if (!isOpen) {
+ return (
+
+ }
+ aria-label="Open AI Assistant"
+ className="chatbot-fab"
+ >
+ AI Assistant
+
+
+ );
+ }
+
+ return (
+
+
+
+
+ {/* Header */}
+
+
+
+
+
+
+
+
+
+ AI Assistant
+
+
+
+ Powered by Debezium Platform
+
+
+ {getConnectionStatusText()}
+
+
+
+
+ }
+ size="sm"
+ />
+
+
+ }
+ />
+
+
+
+
+ {/* Messages */}
+
+
+ {messages.map((message) => (
+
+
+
+
+ {message.role === 'user' ? 'You' : 'AI Assistant'}
+
+
+ {message.timestamp.toLocaleTimeString([], {
+ hour: '2-digit',
+ minute: '2-digit',
+ })}
+
+
+
{message.content}
+
+
+
+ AI is thinking...
+
+
+