yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-11-03 12:08:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move MCP into separate maven module (#22043)

Browse Source 
* Move MCP into separate maven module

* Fix styling

* Add Tests, upgrade to 0.10 mcp-sdk version

* Fix tests

* revert npm package and related files

---------

Co-authored-by: Chirag Madlani <12962843+chirag-madlani@users.noreply.github.com>
This commit is contained in:
Sriharsha Chintalapani 2025-07-01 12:02:29 -07:00 committed by GitHub
parent e5e5017393
commit b461eeb881
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
37 changed files with 1608 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
openmetadata-dist/pom.xml
View File

@ -34,6 +34,11 @@
<artifactId>openmetadata-service</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.open-metadata</groupId>
<artifactId>openmetadata-mcp</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
<properties>

114
openmetadata-mcp/LICENSE Normal file
View File

@ -0,0 +1,114 @@
Collate Community License Agreement
Version 1.0
This Collate Community License Agreement Version 1.0 (the “Agreement”) sets
forth the terms on which Collate, Inc. (“Collate”) makes available certain
software made available by Collate under this Agreement (the “Software”). BY
INSTALLING, DOWNLOADING, ACCESSING, USING OR DISTRIBUTING ANY OF THE SOFTWARE,
YOU AGREE TO THE TERMS AND CONDITIONS OF THIS AGREEMENT. IF YOU DO NOT AGREE TO
SUCH TERMS AND CONDITIONS, YOU MUST NOT USE THE SOFTWARE. IF YOU ARE RECEIVING
THE SOFTWARE ON BEHALF OF A LEGAL ENTITY, YOU REPRESENT AND WARRANT THAT YOU
HAVE THE ACTUAL AUTHORITY TO AGREE TO THE TERMS AND CONDITIONS OF THIS
AGREEMENT ON BEHALF OF SUCH ENTITY. “Licensee” means you, an individual, or
the entity on whose behalf you are receiving the Software.
1. LICENSE GRANT AND CONDITIONS.
1.1 License. Subject to the terms and conditions of this Agreement,
Collate hereby grants to Licensee a non-exclusive, royalty-free,
worldwide, non-transferable, non-sublicenseable license during the term
of this Agreement to: (a) use the Software; (b) prepare modifications and
derivative works of the Software; (c) distribute the Software (including
without limitation in source code or object code form); and (d) reproduce
copies of the Software (the “License”). Licensee is not granted the
right to, and Licensee shall not, exercise the License for an Excluded
Purpose. For purposes of this Agreement, “Excluded Purpose” means making
available any software-as-a-service, platform-as-a-service,
infrastructure-as-a-service or other similar online service that competes
with Collate products or services that provide the Software.
1.2 Conditions. In consideration of the License, Licensee’s distribution
of the Software is subject to the following conditions:
(a) Licensee must cause any Software modified by Licensee to carry
prominent notices stating that Licensee modified the Software.
(b) On each Software copy, Licensee shall reproduce and not remove or
alter all Collate or third party copyright or other proprietary
notices contained in the Software, and Licensee must provide the
notice below with each copy.
“This software is made available by Collate, Inc., under the
terms of the Collate Community License Agreement, Version 1.0
located at http://www.getcollate.io/collate-community-license. BY
INSTALLING, DOWNLOADING, ACCESSING, USING OR DISTRIBUTING ANY OF
THE SOFTWARE, YOU AGREE TO THE TERMS OF SUCH LICENSE AGREEMENT.”
1.3 Licensee Modifications. Licensee may add its own copyright notices
to modifications made by Licensee and may provide additional or different
license terms and conditions for use, reproduction, or distribution of
Licensee’s modifications. While redistributing the Software or
modifications thereof, Licensee may choose to offer, for a fee or free of
charge, support, warranty, indemnity, or other obligations. Licensee, and
not Collate, will be responsible for any such obligations.
1.4 No Sublicensing. The License does not include the right to
sublicense the Software, however, each recipient to which Licensee
provides the Software may exercise the Licenses so long as such recipient
agrees to the terms and conditions of this Agreement.
2. TERM AND TERMINATION. This Agreement will continue unless and until
earlier terminated as set forth herein. If Licensee breaches any of its
conditions or obligations under this Agreement, this Agreement will
terminate automatically and the License will terminate automatically and
permanently.
3. INTELLECTUAL PROPERTY. As between the parties, Collate will retain all
right, title, and interest in the Software, and all intellectual property
rights therein. Collate hereby reserves all rights not expressly granted
to Licensee in this Agreement. Collate hereby reserves all rights in its
trademarks and service marks, and no licenses therein are granted in this
Agreement.
4. DISCLAIMER. COLLATE HEREBY DISCLAIMS ANY AND ALL WARRANTIES AND
CONDITIONS, EXPRESS, IMPLIED, STATUTORY, OR OTHERWISE, AND SPECIFICALLY
DISCLAIMS ANY WARRANTY OF MERCHANTABILITY OR FITNESS FOR A PARTICULAR
PURPOSE, WITH RESPECT TO THE SOFTWARE.
5. LIMITATION OF LIABILITY. COLLATE WILL NOT BE LIABLE FOR ANY DAMAGES OF
ANY KIND, INCLUDING BUT NOT LIMITED TO, LOST PROFITS OR ANY CONSEQUENTIAL,
SPECIAL, INCIDENTAL, INDIRECT, OR DIRECT DAMAGES, HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, ARISING OUT OF THIS AGREEMENT. THE FOREGOING SHALL
APPLY TO THE EXTENT PERMITTED BY APPLICABLE LAW.
6.GENERAL.
6.1 Governing Law. This Agreement will be governed by and interpreted in
accordance with the laws of the state of California, without reference to
its conflict of laws principles. If Licensee is located within the
United States, all disputes arising out of this Agreement are subject to
the exclusive jurisdiction of courts located in Santa Clara County,
California. USA. If Licensee is located outside of the United States,
any dispute, controversy or claim arising out of or relating to this
Agreement will be referred to and finally determined by arbitration in
accordance with the JAMS International Arbitration Rules. The tribunal
will consist of one arbitrator. The place of arbitration will be Palo
Alto, California. The language to be used in the arbitral proceedings
will be English. Judgment upon the award rendered by the arbitrator may
be entered in any court having jurisdiction thereof.
6.2 Assignment. Licensee is not authorized to assign its rights under
this Agreement to any third party. Collate may freely assign its rights
under this Agreement to any third party.
6.3 Other. This Agreement is the entire agreement between the parties
regarding the subject matter hereof. No amendment or modification of
this Agreement will be valid or binding upon the parties unless made in
writing and signed by the duly authorized representatives of both
parties. In the event that any provision, including without limitation
any condition, of this Agreement is held to be unenforceable, this
Agreement and all licenses and rights granted hereunder will immediately
terminate. Waiver by Collate of a breach of any provision of this
Agreement or the failure by Collate to exercise any right hereunder
will not be construed as a waiver of any subsequent breach of that right
or as a waiver of any other right.

2
openmetadata-mcp/lombok.config Normal file
View File

@ -0,0 +1,2 @@
config.stopBubbling = true
lombok.log.fieldName = LOG

205
openmetadata-mcp/pom.xml Normal file
View File

@ -0,0 +1,205 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.open-metadata</groupId>
<artifactId>platform</artifactId>
<version>1.8.0-SNAPSHOT</version>
</parent>
<artifactId>openmetadata-mcp</artifactId>
<name>OpenMetadata MCP</name>
<description>Model Context Protocol implementation for OpenMetadata</description>
<properties>
<assertj-core.version>3.26.3</assertj-core.version>
<okhttp.version>4.12.0</okhttp.version>
<awaitility.version>4.2.2</awaitility.version>
<mcp-sdk.version>0.10.0</mcp-sdk.version>
<testcontainers.version>1.20.4</testcontainers.version>
<jetty.version>11.0.25</jetty.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-bom</artifactId>
<version>${mcp-sdk.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-bom</artifactId>
<version>${testcontainers.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
<version>${mcp-sdk.version}</version>
</dependency>
<dependency>
<groupId>org.open-metadata</groupId>
<artifactId>openmetadata-service</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>io.dropwizard</groupId>
<artifactId>dropwizard-core</artifactId>
<version>${dropwizard.version}</version>
</dependency>
<dependency>
<groupId>io.dropwizard</groupId>
<artifactId>dropwizard-jetty</artifactId>
<version>${dropwizard.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-servlet</artifactId>
<version>${jetty.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
<version>${jetty.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-util</artifactId>
<version>${jetty.version}</version>
</dependency>
<dependency>
<groupId>org.open-metadata</groupId>
<artifactId>openmetadata-service</artifactId>
<version>${project.parent.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>${org.junit.jupiter.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.dropwizard</groupId>
<artifactId>dropwizard-testing</artifactId>
<version>${dropwizard.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.dropwizard</groupId>
<artifactId>dropwizard-client</artifactId>
<version>${dropwizard.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-client</artifactId>
<version>${jetty.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.glassfish.jersey.connectors</groupId>
<artifactId>jersey-jetty-connector</artifactId>
<version>3.1.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-http</artifactId>
<version>${jetty.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-io</artifactId>
<version>${jetty.version}</version>
</dependency>
<dependency>
<groupId>io.dropwizard</groupId>
<artifactId>dropwizard-jersey</artifactId>
<version>${dropwizard.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>${assertj-core.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
<version>${okhttp.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp-sse</artifactId>
<version>${okhttp.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>${awaitility.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mysql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>elasticsearch</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemPropertyVariables>
<jdbcContainerClassName>org.testcontainers.containers.MySQLContainer</jdbcContainerClassName>
<jdbcContainerImage>mysql:8</jdbcContainerImage>
<elasticSearchContainerClassName>docker.elastic.co/elasticsearch/elasticsearch:8.11.4</elasticSearchContainerClassName>
</systemPropertyVariables>
</configuration>
</plugin>
</plugins>
</build>
</project>

98
openmetadata-service/src/main/java/org/openmetadata/service/mcp/HttpServletSseServerTransportProvider.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/HttpServletSseServerTransportProvider.java
View File

@ -1,6 +1,6 @@
package org.openmetadata.service.mcp;
package org.openmetadata.mcp;
import static org.openmetadata.service.mcp.McpUtils.getJsonRpcMessageWithAuthorizationParam;
import static org.openmetadata.mcp.McpUtils.getJsonRpcMessageWithAuthorizationParam;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -26,16 +26,14 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@Slf4j
@WebServlet(asyncSupported = true)
public class HttpServletSseServerTransportProvider extends HttpServlet
implements McpServerTransportProvider {
private static final Logger logger =
LoggerFactory.getLogger(HttpServletSseServerTransportProvider.class);
public static final String UTF_8 = "UTF-8";
public static final String APPLICATION_JSON = "application/json";
public static final String FAILED_TO_SEND_ERROR_RESPONSE = "Failed to send error response: {}";
@ -81,13 +79,13 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
}
@Override
public Mono<Void> notifyClients(String method, Map<String, Object> params) {
public Mono<Void> notifyClients(String method, Object params) {
if (sessions.isEmpty()) {
logger.debug("No active sessions to broadcast message to");
LOG.debug("No active sessions to broadcast message to");
return Mono.empty();
}
logger.debug("Attempting to broadcast message to {} active sessions", sessions.size());
LOG.debug("Attempting to broadcast message to {} active sessions", sessions.size());
return Flux.fromIterable(sessions.values())
.flatMap(
session ->
@ -95,7 +93,7 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
.sendNotification(method, params)
.doOnError(
e ->
logger.error(
LOG.error(
"Failed to send message to session {}: {}",
session.getId(),
e.getMessage()))
@ -109,6 +107,28 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
handleSseEvent(request, response);
}
public Mono<Void> notifyClients(String method, Map<String, Object> params) {
if (sessions.isEmpty()) {
LOG.debug("No active sessions to broadcast message to");
return Mono.empty();
}
LOG.debug("Attempting to broadcast message to {} active sessions", sessions.size());
return Flux.fromIterable(sessions.values())
.flatMap(
session ->
session
.sendNotification(method, params)
.doOnError(
e ->
LOG.error(
"Failed to send message to session {}: {}",
session.getId(),
e.getMessage()))
.onErrorComplete())
.then();
}
private void handleSseEvent(HttpServletRequest request, HttpServletResponse response)
throws IOException {
String requestURI = request.getRequestURI();
@ -144,24 +164,37 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
executorService.submit(
() -> {
// TODO: Handle session lifecycle and keepalive
// Handle session lifecycle and keepalive
try {
while (sessions.containsKey(sessionId)) {
while (sessions.containsKey(sessionId) && !isClosing.get()) {
// Send keepalive every 30 seconds
Thread.sleep(30000);
writer.write(": keep-alive\n\n");
writer.flush();
if (!isClosing.get() && sessions.containsKey(sessionId)) {
writer.write(": keep-alive\n\n");
writer.flush();
// Check if client is still connected
if (writer.checkError()) {
LOG.debug("Client disconnected for session: {}", sessionId);
break;
}
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.debug("Keepalive thread interrupted for session: {}", sessionId);
} catch (Exception e) {
log("SSE error", e);
LOG.error("SSE error for session: {}", sessionId, e);
} finally {
sessions.remove(sessionId);
try {
session.closeGracefully();
asyncContext.complete();
} catch (IllegalStateException e) {
// AsyncContext already completed
LOG.debug("AsyncContext already completed for session: {}", sessionId);
} catch (Exception e) {
log("Error closing long-lived SSE connection", e);
LOG.error("Error closing long-lived SSE connection for session: {}", sessionId, e);
}
}
});
@ -232,7 +265,7 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
response.setStatus(HttpServletResponse.SC_OK);
} catch (Exception e) {
logger.error("Error processing message: {}", e.getMessage());
LOG.error("Error processing message: {}", e.getMessage());
try {
McpError mcpError = new McpError(e.getMessage());
response.setContentType(APPLICATION_JSON);
@ -243,7 +276,7 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
writer.write(jsonError);
writer.flush();
} catch (IOException ex) {
logger.error(FAILED_TO_SEND_ERROR_RESPONSE, ex.getMessage());
LOG.error(FAILED_TO_SEND_ERROR_RESPONSE, ex.getMessage());
response.sendError(
HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Error processing message");
}
@ -253,7 +286,7 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
@Override
public Mono<Void> closeGracefully() {
isClosing.set(true);
logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
LOG.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
return Flux.fromIterable(sessions.values()).flatMap(McpServerSession::closeGracefully).then();
}
@ -269,12 +302,16 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
@Override
public void destroy() {
isClosing.set(true);
LOG.info("Shutting down HttpServletSseServerTransportProvider");
if (executorService != null) {
executorService.shutdownNow();
}
closeGracefully().block();
if (executorService != null) {
executorService.shutdown();
try {
if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
executorService.shutdownNow();
if (!executorService.awaitTermination(2, TimeUnit.SECONDS)) {
LOG.warn("Executor service did not terminate within timeout");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@ -293,7 +330,7 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
this.sessionId = sessionId;
this.asyncContext = asyncContext;
this.writer = writer;
logger.debug("Session transport {} initialized with SSE writer", sessionId);
LOG.debug("Session transport {} initialized with SSE writer", sessionId);
}
@Override
@ -303,9 +340,9 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
try {
String jsonText = objectMapper.writeValueAsString(message);
sendEvent(writer, MESSAGE_EVENT_TYPE, jsonText);
logger.debug("Message sent to session {}", sessionId);
LOG.debug("Message sent to session {}", sessionId);
} catch (Exception e) {
logger.error("Failed to send message to session {}: {}", sessionId, e.getMessage());
LOG.error("Failed to send message to session {}: {}", sessionId, e.getMessage());
sessions.remove(sessionId);
asyncContext.complete();
}
@ -321,13 +358,13 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(
() -> {
logger.debug("Closing session transport: {}", sessionId);
LOG.debug("Closing session transport: {}", sessionId);
try {
sessions.remove(sessionId);
asyncContext.complete();
logger.debug("Successfully completed async context for session {}", sessionId);
LOG.debug("Successfully completed async context for session {}", sessionId);
} catch (Exception e) {
logger.warn(
LOG.warn(
"Failed to complete async context for session {}: {}", sessionId, e.getMessage());
}
});
@ -338,10 +375,9 @@ public class HttpServletSseServerTransportProvider extends HttpServlet
try {
sessions.remove(sessionId);
asyncContext.complete();
logger.debug("Successfully completed async context for session {}", sessionId);
LOG.debug("Successfully completed async context for session {}", sessionId);
} catch (Exception e) {
logger.warn(
"Failed to complete async context for session {}: {}", sessionId, e.getMessage());
LOG.warn("Failed to complete async context for session {}: {}", sessionId, e.getMessage());
}
}
}

51
openmetadata-service/src/main/java/org/openmetadata/service/mcp/MCPStreamableHttpServlet.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/MCPStreamableHttpServlet.java
View File

@ -1,7 +1,7 @@
package org.openmetadata.service.mcp;
package org.openmetadata.mcp;
import static org.openmetadata.service.mcp.McpUtils.getJsonRpcMessageWithAuthorizationParam;
import static org.openmetadata.service.mcp.McpUtils.readRequestBody;
import static org.openmetadata.mcp.McpUtils.getJsonRpcMessageWithAuthorizationParam;
import static org.openmetadata.mcp.McpUtils.readRequestBody;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -36,6 +36,8 @@ import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.openmetadata.mcp.prompts.DefaultPromptsContext;
import org.openmetadata.mcp.tools.DefaultToolContext;
import org.openmetadata.schema.api.configuration.OpenMetadataBaseUrlConfiguration;
import org.openmetadata.schema.entity.app.App;
import org.openmetadata.schema.settings.SettingsType;
@ -45,8 +47,6 @@ import org.openmetadata.service.Entity;
import org.openmetadata.service.config.MCPConfiguration;
import org.openmetadata.service.exception.BadRequestException;
import org.openmetadata.service.limits.Limits;
import org.openmetadata.service.mcp.prompts.DefaultPromptsContext;
import org.openmetadata.service.mcp.tools.DefaultToolContext;
import org.openmetadata.service.resources.settings.SettingsCache;
import org.openmetadata.service.security.Authorizer;
import org.openmetadata.service.security.JwtFilter;
@ -110,7 +110,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
@Override
public void init() throws ServletException {
super.init();
log("MCP Streamable HTTP Servlet initialized");
LOG.info("MCP Streamable HTTP Servlet initialized");
}
@Override
@ -131,7 +131,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
try {
connection.close();
} catch (IOException e) {
log("Error closing SSE connection", e);
LOG.error("Error closing SSE connection", e);
}
}
@ -173,7 +173,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
handleSingleMessage(request, response, jsonNode, sessionId);
}
} catch (Exception e) {
log("Error handling POST request", e);
LOG.error("Error handling POST request", e);
sendError(response, HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Internal server error");
}
}
@ -225,17 +225,17 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
try {
if (isResponse) {
processClientResponse(message, sessionId);
log("Processed client response for session: " + sessionId);
LOG.debug("Processed client response for session: " + sessionId);
} else {
processNotification(message, sessionId);
log("Processed client notification for session: " + sessionId);
LOG.debug("Processed client notification for session: " + sessionId);
}
response.setStatus(HttpServletResponse.SC_ACCEPTED);
response.setContentLength(0);
} catch (Exception e) {
log("Error processing response/notification", e);
LOG.error("Error processing response/notification", e);
sendError(response, HttpServletResponse.SC_BAD_REQUEST, "Failed to process message");
}
}
@ -333,7 +333,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
connection.close();
} catch (Exception e) {
log("Error in SSE stream processing for request", e);
LOG.error("Error in SSE stream processing for request", e);
try {
// Send error response before closing
Map<String, Object> errorResponse =
@ -342,13 +342,13 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
connection.sendEvent(objectMapper.writeValueAsString(errorResponse));
connection.close();
} catch (Exception ex) {
log("Error sending error response in SSE stream", ex);
LOG.error("Error sending error response in SSE stream", ex);
}
} finally {
try {
asyncContext.complete();
} catch (Exception e) {
log("Error completing async context", e);
LOG.error("Error completing async context", e);
}
sseConnections.remove(connectionId);
}
@ -433,6 +433,9 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
// TODO: This is good for now, but we can enhance this logic later, like tools/call can be long
// running for our use case should be fine
// Use SSE for requests that are streaming operations or have specific methods
if (sessionId == null) {
return false;
}
MCPSession session = sessions.get(sessionId);
return session != null;
}
@ -540,20 +543,20 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
connection.close();
} catch (Exception e) {
log("Error in SSE stream processing for batch", e);
LOG.error("Error in SSE stream processing for batch", e);
try {
Map<String, Object> errorResponse =
createErrorResponse(null, -32603, "Internal error: " + e.getMessage());
connection.sendEvent(objectMapper.writeValueAsString(errorResponse));
connection.close();
} catch (Exception ex) {
log("Error sending error response in batch SSE stream", ex);
LOG.error("Error sending error response in batch SSE stream", ex);
}
} finally {
try {
asyncContext.complete();
} catch (Exception e) {
log("Error completing async context for batch", e);
LOG.error("Error completing async context for batch", e);
}
sseConnections.remove(connectionId);
}
@ -687,7 +690,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
response.setStatus(HttpServletResponse.SC_OK);
response.getWriter().write(responseJson);
log("New session initialized: " + sessionId);
LOG.info("New session initialized: " + sessionId);
}
@Override
@ -696,7 +699,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
}
@Override
public Mono<Void> notifyClients(String method, Map<String, Object> params) {
public Mono<Void> notifyClients(String method, Object params) {
if (sessions.isEmpty()) {
LOG.debug("No active sessions to broadcast message to");
return Mono.empty();
@ -751,7 +754,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
return state;
}
public Mono<Void> sendNotification(String method, Map<String, Object> params) {
public Mono<Void> sendNotification(String method, Object params) {
McpSchema.JSONRPCNotification jsonrpcNotification =
new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION, method, params);
return Mono.fromRunnable(
@ -836,7 +839,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
response.put("error", createError(-32601, "Method not found: " + method));
}
} catch (Exception e) {
log("Error processing request: " + method, e);
LOG.error("Error processing request: " + method, e);
response.put("error", createError(-32603, "Internal error: " + e.getMessage()));
}
@ -855,7 +858,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
private void processNotification(JsonNode notification, String sessionId) {
String method = notification.get("method").asText();
log("Received notification: " + method + " (session: " + sessionId + ")");
LOG.debug("Received notification: " + method + " (session: " + sessionId + ")");
// Handle specific notifications
switch (method) {
@ -868,7 +871,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
removeMCPSession(sessionId);
break;
default:
log("Unknown notification: " + method);
LOG.debug("Unknown notification: " + method);
}
}
@ -944,7 +947,7 @@ public class MCPStreamableHttpServlet extends HttpServlet implements McpServerTr
connection.sendEvent(objectMapper.writeValueAsString(notification));
} catch (IOException e) {
log("Error sending notification to session: " + sessionId, e);
LOG.error("Error sending notification to session: " + sessionId, e);
}
}
}

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/McpAuthFilter.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/McpAuthFilter.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp;
package org.openmetadata.mcp;
import static org.openmetadata.service.socket.SocketAddressFilter.validatePrefixedTokenRequest;

16
openmetadata-service/src/main/java/org/openmetadata/service/mcp/McpServer.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/McpServer.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp;
package org.openmetadata.mcp;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.dropwizard.core.setup.Environment;
@ -12,27 +12,35 @@ import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.jetty.servlet.FilterHolder;
import org.eclipse.jetty.servlet.ServletHolder;
import org.openmetadata.mcp.prompts.DefaultPromptsContext;
import org.openmetadata.mcp.tools.DefaultToolContext;
import org.openmetadata.schema.utils.JsonUtils;
import org.openmetadata.service.OpenMetadataApplicationConfig;
import org.openmetadata.service.apps.McpServerProvider;
import org.openmetadata.service.limits.Limits;
import org.openmetadata.service.mcp.prompts.DefaultPromptsContext;
import org.openmetadata.service.mcp.tools.DefaultToolContext;
import org.openmetadata.service.security.Authorizer;
import org.openmetadata.service.security.JwtFilter;
@Slf4j
public class McpServer {
public class McpServer implements McpServerProvider {
private JwtFilter jwtFilter;
private Authorizer authorizer;
private Limits limits;
protected DefaultToolContext toolContext;
protected DefaultPromptsContext promptsContext;
// Default constructor for dynamic loading
public McpServer() {
this.toolContext = new DefaultToolContext();
this.promptsContext = new DefaultPromptsContext();
}
public McpServer(DefaultToolContext toolContext, DefaultPromptsContext promptsContext) {
this.toolContext = toolContext;
this.promptsContext = promptsContext;
}
@Override
public void initializeMcpServer(
Environment environment,
Authorizer authorizer,

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/McpUtils.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/McpUtils.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp;
package org.openmetadata.mcp;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

4
openmetadata-service/src/main/java/org/openmetadata/service/mcp/prompts/DefaultPromptsContext.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/prompts/DefaultPromptsContext.java
View File

@ -1,6 +1,6 @@
package org.openmetadata.service.mcp.prompts;
package org.openmetadata.mcp.prompts;
import static org.openmetadata.service.mcp.McpUtils.getPrompts;
import static org.openmetadata.mcp.McpUtils.getPrompts;
import io.modelcontextprotocol.spec.McpSchema;
import java.util.ArrayList;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/prompts/GreetingsPrompt.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/prompts/GreetingsPrompt.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.prompts;
package org.openmetadata.mcp.prompts;
import io.modelcontextprotocol.spec.McpSchema;
import java.util.List;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/prompts/McpPrompt.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/prompts/McpPrompt.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.prompts;
package org.openmetadata.mcp.prompts;
import io.modelcontextprotocol.spec.McpSchema;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/prompts/SearchPrompt.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/prompts/SearchPrompt.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.prompts;
package org.openmetadata.mcp.prompts;
import io.modelcontextprotocol.spec.McpSchema;
import java.util.List;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/prompts/WrappedGetPromptResult.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/prompts/WrappedGetPromptResult.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.prompts;
package org.openmetadata.mcp.prompts;
import io.modelcontextprotocol.spec.McpSchema;
import lombok.Getter;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/CommonUtils.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/CommonUtils.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import java.util.List;
import java.util.Map;

4
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/DefaultToolContext.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/DefaultToolContext.java
View File

@ -1,6 +1,6 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import static org.openmetadata.service.mcp.McpUtils.getToolProperties;
import static org.openmetadata.mcp.McpUtils.getToolProperties;
import io.modelcontextprotocol.spec.McpSchema;
import java.util.List;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/GetEntityTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/GetEntityTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import static org.openmetadata.schema.type.MetadataOperation.VIEW_ALL;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/GetLineageTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/GetLineageTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import java.io.IOException;
import java.util.Map;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/GlossaryTermTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/GlossaryTermTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import static org.openmetadata.common.utils.CommonUtil.nullOrEmpty;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/GlossaryTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/GlossaryTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import java.util.List;
import java.util.Map;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/McpTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/McpTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import java.io.IOException;
import java.util.Map;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/PatchEntityTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/PatchEntityTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import jakarta.json.JsonPatch;
import java.util.Map;

2
openmetadata-service/src/main/java/org/openmetadata/service/mcp/tools/SearchMetadataTool.java → openmetadata-mcp/src/main/java/org/openmetadata/mcp/tools/SearchMetadataTool.java
View File

@ -1,4 +1,4 @@
package org.openmetadata.service.mcp.tools;
package org.openmetadata.mcp.tools;
import static org.openmetadata.service.search.SearchUtil.mapEntityTypesToIndexNames;
import static org.openmetadata.service.security.DefaultAuthorizer.getSubjectContext;

38
openmetadata-mcp/src/main/resources/json/data/mcp/prompts.json Normal file
View File

@ -0,0 +1,38 @@
{
"tools": [
{
"name": "create-greeting",
"description": "Generate a customized greeting message",
"arguments": [
{
"name": "name",
"description": "Name of the person to greet",
"required": true
},
{
"name": "style",
"description": "The style of greeting, such a formal, excited, or casual. If not specified casual will be used"
}
]
},
{
"name": "search_metadata",
"description": "Creates a prompt for Searching metadata in OpenMetadata.",
"arguments": [
{
"name": "query",
"description": "Keywords to use for searching.",
"required": true
},
{
"name": "entity_type",
"description": "Entity Type to Filter Report."
},
{
"name": "limit",
"description": "Maximum number of results to return. Default is 10. Try to keep this number low unless the user asks for more."
}
]
}
]
}

187
openmetadata-mcp/src/main/resources/json/data/mcp/tools.json Normal file
View File

@ -0,0 +1,187 @@
{
"tools": [
{
"name": "search_metadata",
"description": "Find your data and business terms in OpenMetadata. For example if the user asks to 'find tables that contain customers information', then 'customers' should be the query, and the entity_type should be 'table'. Here make sure to use 'Href' is available in result to create a hyperlink to the entity in OpenMetadata.",
"parameters": {
"description": "The search query to find metadata in the OpenMetadata catalog, entity type could be table, topic etc. Limit can be used to paginate on the data.",
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Keywords to use for searching."
},
"entity_type": {
"type": "string",
"description": "Optional entity type to filter results. The OpenMetadata entities are categorized as follows: Service Entities include databaseService, messagingService, apiService, dashboardService, pipelineService, storageService, mlmodelService, metadataService, and searchService; Data Asset Entities include apiCollection, apiEndpoint, table, storedProcedure, database, databaseSchema, dashboard, dashboardDataModel, pipeline, chart, topic, searchIndex, mlmodel, and container; User Entities include user and team; Domain entities include domain and dataProduct; and Governance entities include metric, glossary, and glossaryTerm."
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return. Default is 10. Try to keep this number low unless the user asks for more."
}
},
"required": [
"query"
]
}
},
{
"name": "get_entity_details",
"description": "Get detailed information about a specific entity",
"parameters": {
"description": "Fqn is the fully qualified name of the entity. Entity type could be table, topic etc.",
"type": "object",
"properties": {
"entity_type": {
"type": "string",
"description": "Type of entity"
},
"fqn": {
"type": "string",
"description": "Fully qualified name of the entity"
}
},
"required": [
"entity_type",
"fqn"
]
}
},
{
"name": "create_glossary_term",
"description": "Creates a new Glossary Term. Note that a glossary term must be part of a Glossary, so the glossary must be specified in the parameters. If you can't find the right glossary to use, respond back to the user to create a new Glossary first. Note that you can help the user to create the Glossary as well. If you don't find any Glossary that could be related, please list to the user the available Glossaries so users can choose if they want to create or reuse something. Also, note that glossary terms can be hierarchical: for example, a glossary term 'Accounts' can have a child term 'Credit Account', 'Savings Account', etc. So if you find any terms that can be related, it might make sense to create a new term as a child of an existing term.",
"parameters": {
"type": "object",
"properties": {
"glossary": {
"type": "string",
"description": "Glossary in which the term belongs. This should be its fully qualified name."
},
"parentTerm": {
"type": "string",
"description": "Optional parent term for the new term. This should be its fully qualified name defined as <glossary>.<term>. If the Glossary Term has other parents, the Fully Qualified Name will be <glossary>.<parent>...<term>. If not provided, the term will be created at the root level of the glossary."
},
"name": {
"type": "string",
"description": "Glossary Term name."
},
"description": {
"type": "string",
"description": "Glossary Term description."
},
"owners": {
"type": "array",
"description": "Glossary Term owner. This could be an OpenMetadata User or Team. If you don't know who the owner is, you can leave this empty, but let the user know that they can add owners later.",
"items": {
"type": "string"
}
}
},
"required": [
"glossary",
"name",
"description"
]
}
},
{
"name": "create_glossary",
"description": "Creates a new Glossary. A Glossary is a collection of terms that are used to define the business vocabulary of an organization. Typically, similar terms are grouped together in a Glossary. For example, a Glossary names 'Marketing' could contain terms like 'Campaign', 'Lead', 'Opportunity', etc.",
"parameters": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Glossary Term name."
},
"description": {
"type": "string",
"description": "Glossary Term description."
},
"owners": {
"type": "array",
"description": "Glossary Term owner. This could be an OpenMetadata User or Team. If you don't know who the owner is, you can leave this empty, but let the user know that they can add owners later.",
"items": {
"type": "string"
}
},
"reviewers": {
"type": "array",
"description": "Glossary Term owner. This could be an OpenMetadata User or Team. If you don't know who the owner is, you can leave this empty, but let the user know that they can add owners later.",
"items": {
"type": "string"
}
},
"mutuallyExclusive": {
"type": "boolean",
"description": "Glossary terms that are direct children in this glossary are mutually exclusive. When mutually exclusive is `true` only one term can be used to tag an entity. When mutually exclusive is `false`, multiple terms from this group can be used to tag an entity. This is an important setting. If you are not sure, ask the user to clarify. If the user doesn't know, set it to `false`.",
"default": false
}
},
"required": [
"name",
"description",
"mutuallyExclusive"
]
}
},
{
"name": "patch_entity",
"description": "Patches an Entity based on a JSONPatch. Beforehand the Entity should be validated by finding it and creating a proper patch.",
"parameters": {
"type": "object",
"properties": {
"entityType": {
"type": "string",
"description": "Entity Type to patch."
},
"entityFqn": {
"type": "string",
"description": "Fully Qualified Name of the Entity to be patched."
},
"patch": {
"type": "string",
"description": "JSONPatch as String format."
}
},
"required": [
"entityType",
"entityFqn",
"patch"
]
}
},
{
"name": "get_entity_lineage",
"description": "Get detailed information about lineage (upstream/downstream dependencies) of a specific entity. Use this for root cause (upstream entities) or impact (downstream entities) analysis and explaining dependencies between entities.",
"parameters": {
"description": "Fqn is the fully qualified name of the entity. Entity type could be table, topic etc.",
"type": "object",
"properties": {
"entity_type": {
"type": "string",
"description": "Type of entity"
},
"fqn": {
"type": "string",
"description": "Fully qualified name of the entity"
},
"upstream_depth": {
"type": "integer",
"description": "Depth for reaching upstream entities. Default is 5."
},
"downstream_depth": {
"type": "integer",
"description": "Depth for reaching downstream entities. Default is 5."
}
},
"required": [
"entity_type",
"fqn",
"upstream_depth",
"downstream_depth"
]
}
}
]
}

427
openmetadata-mcp/src/test/java/org/openmetadata/mcp/McpIntegrationTest.java Normal file
View File

@ -0,0 +1,427 @@
package org.openmetadata.mcp;
import static org.assertj.core.api.Assertions.assertThat;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.dropwizard.testing.ConfigOverride;
import io.dropwizard.testing.ResourceHelpers;
import io.dropwizard.testing.junit5.DropwizardAppExtension;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import okhttp3.sse.EventSources;
import org.jetbrains.annotations.NotNull;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.openmetadata.schema.auth.JWTAuthMechanism;
import org.openmetadata.schema.auth.ServiceTokenType;
import org.openmetadata.schema.entity.app.AppSchedule;
import org.openmetadata.schema.entity.app.CreateApp;
import org.openmetadata.schema.entity.app.ScheduleTimeline;
import org.openmetadata.schema.entity.teams.User;
import org.openmetadata.service.Entity;
import org.openmetadata.service.OpenMetadataApplication;
import org.openmetadata.service.OpenMetadataApplicationConfig;
import org.openmetadata.service.OpenMetadataApplicationTest;
import org.openmetadata.service.jdbi3.AppRepository;
import org.openmetadata.service.security.jwt.JWTTokenGenerator;
public class McpIntegrationTest extends OpenMetadataApplicationTest {
private static String CONFIG_PATH_OVERRIDE =
ResourceHelpers.resourceFilePath("test-config-mcp.yaml");
private OkHttpClient client;
private ObjectMapper objectMapper;
private String authToken;
@Override
@NotNull
protected DropwizardAppExtension<OpenMetadataApplicationConfig> getApp(
ConfigOverride[] configOverridesArray) {
return new DropwizardAppExtension<>(
OpenMetadataApplication.class, CONFIG_PATH_OVERRIDE, configOverridesArray);
}
@BeforeAll
void setUp() throws Exception {
client =
new OkHttpClient.Builder()
.connectTimeout(10, TimeUnit.SECONDS)
.readTimeout(10, TimeUnit.SECONDS)
.writeTimeout(10, TimeUnit.SECONDS)
.build();
objectMapper = new ObjectMapper();
OpenMetadataApplicationConfig config = APP.getConfiguration();
try {
User adminUser =
new User().withName("admin").withEmail("admin@open-metadata.org").withIsAdmin(true);
JWTTokenGenerator.getInstance()
.init(
config.getAuthenticationConfiguration().getTokenValidationAlgorithm(),
config.getJwtTokenConfiguration());
JWTAuthMechanism jwtAuthMechanism =
JWTTokenGenerator.getInstance()
.generateJWTToken(
adminUser.getName(),
null,
adminUser.getIsAdmin(),
adminUser.getEmail(),
3600,
false,
ServiceTokenType.OM_USER);
authToken = "Bearer " + jwtAuthMechanism.getJWTToken();
} catch (Exception e) {
throw new RuntimeException("Failed to generate auth token", e);
}
installMcpApplication();
}
@AfterAll
void tearDown() throws Exception {
if (client != null) {
client.dispatcher().executorService().shutdown();
client.connectionPool().evictAll();
if (client.cache() != null) {
client.cache().close();
}
}
// Give some time for resources to clean up
Thread.sleep(500);
}
private String getMcpUrl(String path) {
return String.format("http://localhost:%d%s", APP.getLocalPort(), path);
}
private void installMcpApplication() throws Exception {
// Check if McpApplication already exists
AppRepository appRepository = (AppRepository) Entity.getEntityRepository(Entity.APPLICATION);
try {
appRepository.getByName(null, "McpApplication", appRepository.getFields("id"));
} catch (Exception e) {
Map<String, Object> appConfig = new HashMap<>();
appConfig.put("originValidationEnabled", false);
appConfig.put("originHeaderUri", "http://localhost:" + APP.getLocalPort());
CreateApp createApp =
new CreateApp()
.withName("McpApplication")
.withAppConfiguration(appConfig)
.withAppSchedule(new AppSchedule().withScheduleTimeline(ScheduleTimeline.HOURLY));
WebTarget installTarget = getResource("apps");
Response createResponse =
installTarget
.request(MediaType.APPLICATION_JSON)
.header("Authorization", authToken)
.post(jakarta.ws.rs.client.Entity.json(createApp));
if (createResponse.getStatus() != 201 && createResponse.getStatus() != 409) {
throw new RuntimeException(
"Failed to create McpApplication: " + createResponse.getStatus());
}
Thread.sleep(2000);
}
}
@Test
void testMcpInitialization() throws Exception {
Map<String, Object> initRequest = McpTestUtils.createInitializeRequest();
String requestBody = objectMapper.writeValueAsString(initRequest);
okhttp3.RequestBody body =
okhttp3.RequestBody.create(requestBody, okhttp3.MediaType.parse("application/json"));
Request request =
new Request.Builder()
.url(getMcpUrl("/mcp"))
.header("Accept", "application/json, text/event-stream")
.header("Authorization", authToken)
.post(body)
.build();
try (okhttp3.Response response = client.newCall(request).execute()) {
assertThat(response.code()).isEqualTo(200);
assertThat(response.body()).isNotNull();
String responseBody = response.body().string();
JsonNode responseJson = objectMapper.readTree(responseBody);
assertThat(responseJson.has("jsonrpc")).isTrue();
assertThat(responseJson.get("jsonrpc").asText()).isEqualTo("2.0");
assertThat(responseJson.has("result")).isTrue();
JsonNode result = responseJson.get("result");
assertThat(result.has("protocolVersion")).isTrue();
assertThat(result.get("protocolVersion").asText()).isEqualTo("2024-11-05");
assertThat(result.has("capabilities")).isTrue();
assertThat(result.has("serverInfo")).isTrue();
String sessionId = response.header("Mcp-Session-Id");
assertThat(sessionId).isNotNull();
}
}
@Test
void testMcpToolsList() throws Exception {
// Given - Initialize session first
String sessionId = initializeMcpSession();
Map<String, Object> toolsListRequest = new HashMap<>();
toolsListRequest.put("jsonrpc", "2.0");
toolsListRequest.put("id", UUID.randomUUID().toString());
toolsListRequest.put("method", "tools/list");
String requestBody = objectMapper.writeValueAsString(toolsListRequest);
okhttp3.RequestBody body =
okhttp3.RequestBody.create(requestBody, okhttp3.MediaType.parse("application/json"));
Request request =
new Request.Builder()
.url(getMcpUrl("/mcp"))
.header("Accept", "application/json, text/event-stream")
.header("Authorization", authToken)
.header("Mcp-Session-Id", sessionId)
.post(body)
.build();
try (okhttp3.Response response = client.newCall(request).execute()) {
assertThat(response.code()).isEqualTo(200);
assert response.body() != null;
String responseBody = response.body().string();
// Handle SSE response format if present
String jsonContent = responseBody;
if (responseBody.startsWith("id:") || responseBody.startsWith("data:")) {
// Extract JSON from SSE format
String[] lines = responseBody.split("\n");
for (String line : lines) {
if (line.startsWith("data:")) {
jsonContent = line.substring(5).trim();
break;
}
}
}
JsonNode responseJson = objectMapper.readTree(jsonContent);
assertThat(responseJson.has("result")).isTrue();
JsonNode result = responseJson.get("result");
assertThat(result.has("tools")).isTrue();
assertThat(result.get("tools").isArray()).isTrue();
assertThat(result.get("tools").size()).isGreaterThan(0);
boolean hasSearchTool = false;
boolean hasGetEntityTool = false;
for (JsonNode tool : result.get("tools")) {
String toolName = tool.get("name").asText();
if ("search_metadata".equals(toolName)) {
hasSearchTool = true;
} else if ("get_entity_details".equals(toolName)) {
hasGetEntityTool = true;
}
}
assertThat(hasSearchTool).isTrue();
assertThat(hasGetEntityTool).isTrue();
}
}
@Test
void testMcpPromptsList() throws Exception {
String sessionId = initializeMcpSession();
Map<String, Object> promptsListRequest = new HashMap<>();
promptsListRequest.put("jsonrpc", "2.0");
promptsListRequest.put("id", UUID.randomUUID().toString());
promptsListRequest.put("method", "prompts/list");
String requestBody = objectMapper.writeValueAsString(promptsListRequest);
okhttp3.RequestBody body =
okhttp3.RequestBody.create(requestBody, okhttp3.MediaType.parse("application/json"));
Request request =
new Request.Builder()
.url(getMcpUrl("/mcp"))
.header("Accept", "application/json, text/event-stream")
.header("Authorization", authToken)
.header("Mcp-Session-Id", sessionId)
.post(body)
.build();
try (okhttp3.Response response = client.newCall(request).execute()) {
assertThat(response.code()).isEqualTo(200);
assert response.body() != null;
String responseBody = response.body().string();
// Handle SSE response format if present
String jsonContent = responseBody;
if (responseBody.startsWith("id:") || responseBody.startsWith("data:")) {
// Extract JSON from SSE format
String[] lines = responseBody.split("\n");
for (String line : lines) {
if (line.startsWith("data:")) {
jsonContent = line.substring(5).trim();
break;
}
}
}
JsonNode responseJson = objectMapper.readTree(jsonContent);
assertThat(responseJson.has("result")).isTrue();
JsonNode result = responseJson.get("result");
assertThat(result.has("prompts")).isTrue();
assertThat(result.get("prompts").isArray()).isTrue();
}
}
// test passes but server doesn't close cleanly
void testMcpSseConnection() throws Exception {
CountDownLatch connectionLatch = new CountDownLatch(1);
CountDownLatch closeLatch = new CountDownLatch(1);
AtomicReference<String> endpointUrl = new AtomicReference<>();
AtomicReference<EventSource> eventSourceRef = new AtomicReference<>();
Request request =
new Request.Builder()
.url(getMcpUrl("/mcp/sse"))
.header("Accept", "text/event-stream")
.header("Authorization", authToken)
.build();
EventSourceListener listener =
new EventSourceListener() {
@Override
public void onOpen(@NotNull EventSource eventSource, @NotNull okhttp3.Response response) {
eventSourceRef.set(eventSource);
}
@Override
public void onEvent(EventSource eventSource, String id, String type, String data) {
if ("endpoint".equals(type)) {
endpointUrl.set(data);
connectionLatch.countDown();
eventSource.cancel();
}
}
@Override
public void onFailure(
@NotNull EventSource eventSource, Throwable t, okhttp3.Response response) {
connectionLatch.countDown();
closeLatch.countDown();
}
@Override
public void onClosed(@NotNull EventSource eventSource) {
closeLatch.countDown();
}
};
EventSource eventSource = EventSources.createFactory(client).newEventSource(request, listener);
try {
assertThat(connectionLatch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(endpointUrl.get()).isNotNull();
assertThat(endpointUrl.get()).contains("/mcp/messages?sessionId=");
EventSource es = eventSourceRef.get();
if (es != null) {
es.cancel();
}
} finally {
eventSource.cancel();
}
}
@Test
void testMcpToolCall() throws Exception {
// Given - Initialize session first
String sessionId = initializeMcpSession();
// Create a search metadata tool call
Map<String, Object> toolCallRequest = McpTestUtils.createSearchMetadataToolCall("test", 5);
String requestBody = objectMapper.writeValueAsString(toolCallRequest);
okhttp3.RequestBody body =
okhttp3.RequestBody.create(requestBody, okhttp3.MediaType.parse("application/json"));
Request request =
new Request.Builder()
.url(getMcpUrl("/mcp"))
.header("Accept", "application/json, text/event-stream")
.header("Authorization", authToken)
.header("Mcp-Session-Id", sessionId)
.post(body)
.build();
// When
try (okhttp3.Response response = client.newCall(request).execute()) {
// Then
assertThat(response.code()).isEqualTo(200);
String responseBody = response.body().string();
// Handle SSE response format if present
String jsonContent = responseBody;
if (responseBody.startsWith("id:") || responseBody.startsWith("data:")) {
// Extract JSON from SSE format
String[] lines = responseBody.split("\n");
for (String line : lines) {
if (line.startsWith("data:")) {
jsonContent = line.substring(5).trim();
break;
}
}
}
JsonNode responseJson = objectMapper.readTree(jsonContent);
assertThat(responseJson.has("result")).isTrue();
JsonNode result = responseJson.get("result");
assertThat(result.has("content")).isTrue();
assertThat(result.get("content").isArray()).isTrue();
// The search result should contain at least one content item
JsonNode content = result.get("content");
assertThat(content.size()).isGreaterThanOrEqualTo(1);
assertThat(content.get(0).has("type")).isTrue();
assertThat(content.get(0).get("type").asText()).isEqualTo("text");
}
}
private String initializeMcpSession() throws Exception {
Map<String, Object> initRequest = McpTestUtils.createInitializeRequest();
String requestBody = objectMapper.writeValueAsString(initRequest);
okhttp3.RequestBody body =
okhttp3.RequestBody.create(requestBody, okhttp3.MediaType.parse("application/json"));
Request request =
new Request.Builder()
.url(getMcpUrl("/mcp"))
.header("Accept", "application/json, text/event-stream")
.header("Authorization", authToken)
.post(body)
.build();
try (okhttp3.Response response = client.newCall(request).execute()) {
assertThat(response.code()).isEqualTo(200);
return response.header("Mcp-Session-Id");
}
}
}

163
openmetadata-mcp/src/test/java/org/openmetadata/mcp/McpTestUtils.java Normal file
View File

@ -0,0 +1,163 @@
package org.openmetadata.mcp;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
/**
* Utility class for MCP testing.
* Provides helper methods to create valid MCP protocol messages.
*/
public class McpTestUtils {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
/**
* Creates a valid JSON-RPC 2.0 request.
*/
public static Map<String, Object> createJsonRpcRequest(
String method, Map<String, Object> params) {
Map<String, Object> request = new HashMap<>();
request.put("jsonrpc", "2.0");
request.put("id", UUID.randomUUID().toString());
request.put("method", method);
if (params != null) {
request.put("params", params);
}
return request;
}
/**
* Creates a valid JSON-RPC 2.0 notification (no id).
*/
public static Map<String, Object> createJsonRpcNotification(
String method, Map<String, Object> params) {
Map<String, Object> notification = new HashMap<>();
notification.put("jsonrpc", "2.0");
notification.put("method", method);
if (params != null) {
notification.put("params", params);
}
return notification;
}
/**
* Creates an initialize request.
*/
public static Map<String, Object> createInitializeRequest() {
Map<String, Object> params = new HashMap<>();
params.put("protocolVersion", "2024-11-05");
params.put(
"capabilities",
Map.of(
"tools", Map.of("listSupported", true),
"prompts", Map.of("listSupported", true)));
params.put(
"clientInfo",
Map.of(
"name", "test-client",
"version", "1.0.0"));
return createJsonRpcRequest("initialize", params);
}
/**
* Creates a tool call request.
*/
public static Map<String, Object> createToolCallRequest(
String toolName, Map<String, Object> arguments) {
Map<String, Object> params = new HashMap<>();
params.put("name", toolName);
params.put("arguments", arguments != null ? arguments : new HashMap<>());
return createJsonRpcRequest("tools/call", params);
}
/**
* Creates a prompt get request.
*/
public static Map<String, Object> createPromptGetRequest(
String promptName, Map<String, Object> arguments) {
Map<String, Object> params = new HashMap<>();
params.put("name", promptName);
params.put("arguments", arguments != null ? arguments : new HashMap<>());
return createJsonRpcRequest("prompts/get", params);
}
/**
* Validates a JSON-RPC response.
*/
public static void validateJsonRpcResponse(JsonNode response) {
if (response == null) {
throw new AssertionError("Response is null");
}
if (!response.has("jsonrpc") || !"2.0".equals(response.get("jsonrpc").asText())) {
throw new AssertionError("Invalid JSON-RPC version");
}
if (!response.has("id")) {
throw new AssertionError("Response missing id");
}
boolean hasResult = response.has("result");
boolean hasError = response.has("error");
if (hasResult && hasError) {
throw new AssertionError("Response cannot have both result and error");
}
if (!hasResult && !hasError) {
throw new AssertionError("Response must have either result or error");
}
if (hasError) {
JsonNode error = response.get("error");
if (!error.has("code") || !error.has("message")) {
throw new AssertionError("Error response missing required fields");
}
}
}
/**
* Creates test authorization header.
*/
public static String createAuthorizationHeader(String token) {
return "Bearer " + token;
}
/**
* Extracts session ID from response headers.
*/
public static String extractSessionId(okhttp3.Response response) {
return response.header("Mcp-Session-Id");
}
/**
* Creates a search metadata tool call.
*/
public static Map<String, Object> createSearchMetadataToolCall(String query, int limit) {
Map<String, Object> arguments = new HashMap<>();
arguments.put("query", query);
arguments.put("limit", limit);
arguments.put("entity_type", "table");
arguments.put("Authorization", createAuthorizationHeader("test-token"));
return createToolCallRequest("search_metadata", arguments);
}
/**
* Creates a get entity details tool call.
*/
public static Map<String, Object> createGetEntityToolCall(String entityType, String fqn) {
Map<String, Object> arguments = new HashMap<>();
arguments.put("entity_type", entityType);
arguments.put("fqn", fqn);
arguments.put("Authorization", createAuthorizationHeader("test-token"));
return createToolCallRequest("get_entity_details", arguments);
}
}

BIN
openmetadata-mcp/src/test/resources/private_key.der Normal file
View File

Binary file not shown.

BIN
openmetadata-mcp/src/test/resources/public_key.der Normal file
View File

Binary file not shown.

222
openmetadata-mcp/src/test/resources/test-config-mcp.yaml Normal file
View File

@ -0,0 +1,222 @@
# Copyright 2021 Collate
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
clusterName: openmetadata
swagger:
resourcePackage: org.openmetadata.service.webservice.resources
server:
rootPath: '/api/*'
applicationConnectors:
- type: http
port: 0
adminConnectors:
- type: http
port: 0
# Above configuration for running http is fine for dev and testing.
# For production setup, where UI app will hit apis through DPS it
# is strongly recommended running https instead. Note that only
# keyStorePath and keyStorePassword are mandatory properties. Values
# for other properties are defaults
#server:
#applicationConnectors:
# - type: https
# port: 8585
# keyStorePath: ./conf/keystore.jks
# keyStorePassword: changeit
# keyStoreType: JKS
# keyStoreProvider:
# trustStorePath: /path/to/file
# trustStorePassword: changeit
# trustStoreType: JKS
# trustStoreProvider:
# keyManagerPassword: changeit
# needClientAuth: false
# wantClientAuth:
# certAlias: <alias>
# crlPath: /path/to/file
# enableCRLDP: false
# enableOCSP: false
# maxCertPathLength: (unlimited)
# ocspResponderUrl: (none)
# jceProvider: (none)
# validateCerts: true
# validatePeers: true
# supportedProtocols: SSLv3
# supportedCipherSuites: TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256
# allowRenegotiation: true
# endpointIdentificationAlgorithm: (none)
#adminConnectors:
# - type: https
# port: 8586
# keyStorePath: ./conf/keystore.jks
# keyStorePassword: changeit
# keyStoreType: JKS
# keyStoreProvider:
# trustStorePath: /path/to/file
# trustStorePassword: changeit
# trustStoreType: JKS
# trustStoreProvider:
# keyManagerPassword: changeit
# needClientAuth: false
# wantClientAuth:
# certAlias: <alias>
# crlPath: /path/to/file
# enableCRLDP: false
# enableOCSP: false
# maxCertPathLength: (unlimited)
# ocspResponderUrl: (none)
# jceProvider: (none)
# validateCerts: true
# validatePeers: true
# supportedProtocols: SSLv3
# supportedCipherSuites: TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256
# allowRenegotiation: true
# endpointIdentificationAlgorithm: (none)
# Logging settings.
logging:
level: INFO
appenders:
- type: console
logFormat: "%level %logger{5} - %msg%n"
database:
# the name of the JDBC driver, h2 is used for testing
driverClass: org.postgresql.Driver
# the username and password
user: test
password:
# the JDBC URL; the database is called openmetadata_test_db
url: jdbc:postgresql://localhost:3307/openmetadata_test_db?useSSL=false&serverTimezone=UTC
migrationConfiguration:
flywayPath: "../bootstrap/sql/migrations/flyway"
nativePath: "../bootstrap/sql/migrations/native"
extensionPath: ""
#elasticsearch:
# host: localhost
# port: 0
secretsManagerConfiguration:
secretsManager: db
health:
delayedShutdownHandlerEnabled: true
shutdownWaitPeriod: 5s
healthChecks:
- name: OpenMetadataServerHealthCheck
critical: true
# Authorizer Configuration
authorizerConfiguration:
className: "org.openmetadata.service.security.DefaultAuthorizer"
# JWT Filter
containerRequestFilter: "org.openmetadata.service.security.CatalogOpenIdAuthorizationRequestFilter"
adminPrincipals:
- "admin"
- "hello.world"
# Added only for test purposes and not for production setup
testPrincipals:
- "test"
principalDomain: "open-metadata.org"
enforcePrincipalDomain: false
enableSecureSocketConnection: false
allowedEmailRegistrationDomains:
- "all"
authenticationConfiguration:
clientType: "public"
provider: "basic"
providerName: ""
publicKeyUrls:
- "https://www.googleapis.com/oauth2/v3/certs"
authority: "https://accounts.google.com"
clientId: "261867039324-neb92r2147i6upchb78tv29idk079bps.apps.googleusercontent.com"
callbackUrl: "http://localhost:8585/callback"
jwtPrincipalClaims:
- "email"
- "preferred_username"
- "sub"
enableSelfSignup : true
samlConfiguration:
debugMode: false
idp:
entityId: ""
ssoLoginUrl: ""
idpX509Certificate: ""
authorityUrl: ""
nameId: ""
sp:
entityId: ""
acs: ""
spX509Certificate: ""
callback: ""
security:
strictMode: false
validateXml: false
tokenValidity: 3600
sendEncryptedNameId: false
sendSignedAuthRequest: false
signSpMetadata: false
wantMessagesSigned: false
wantAssertionsSigned: false
wantAssertionEncrypted: false
jwtTokenConfiguration:
rsapublicKeyFilePath: "src/test/resources/public_key.der"
rsaprivateKeyFilePath: "src/test/resources/private_key.der"
jwtissuer: "open-metadata.org"
keyId: "Gb389a-9f76-gdjs-a92j-0242bk94356"
eventMonitoringConfiguration:
eventMonitor: "prometheus"
batchSize: 10
pathPattern: ["/*"]
latency: [0.90]
eventHandlerConfiguration:
eventHandlerClassNames:
- "org.openmetadata.service.events.AuditEventHandler"
- "org.openmetadata.service.events.ChangeEventHandler"
pipelineServiceClientConfiguration:
className: "org.openmetadata.service.clients.pipeline.airflow.AirflowRESTClient"
metadataApiEndpoint: http://localhost:8585/api
apiEndpoint: http://localhost:8080
hostIp: ""
healthCheckInterval: 300
verifySSL: "no-ssl"
authProvider: "openmetadata"
parameters:
username: admin
password: admin
timeout: 10
truststorePath: ""
truststorePassword: ""
fernetConfiguration:
fernetKey: ${FERNET_KEY:-ihZpp5gmmDvVsgoOG6OVivKWwC9vd5JQ}
dataQualityConfiguration:
severityIncidentClassifier: "org.openmetadata.service.util.incidentSeverityClassifier.LogisticRegressionIncidentSeverityClassifier"
objectStorage:
enabled: false
provider: NOOP
maxFileSize: 5242880

24
openmetadata-service/src/main/java/org/openmetadata/service/OpenMetadataApplication.java
View File

@ -37,7 +37,6 @@ import io.swagger.v3.oas.annotations.info.Info;
import io.swagger.v3.oas.annotations.info.License;
import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.security.SecurityScheme;
import io.swagger.v3.oas.annotations.security.SecuritySchemes;
import io.swagger.v3.oas.annotations.servers.Server;
import jakarta.servlet.DispatcherType;
import jakarta.servlet.FilterRegistration;
@ -75,6 +74,7 @@ import org.openmetadata.schema.services.connections.metadata.AuthProvider;
import org.openmetadata.search.IndexMappingLoader;
import org.openmetadata.service.apps.ApplicationContext;
import org.openmetadata.service.apps.ApplicationHandler;
import org.openmetadata.service.apps.McpServerProvider;
import org.openmetadata.service.apps.scheduler.AppScheduler;
import org.openmetadata.service.config.OMWebBundle;
import org.openmetadata.service.config.OMWebConfiguration;
@ -99,9 +99,6 @@ import org.openmetadata.service.jobs.JobDAO;
import org.openmetadata.service.jobs.JobHandlerRegistry;
import org.openmetadata.service.limits.DefaultLimits;
import org.openmetadata.service.limits.Limits;
import org.openmetadata.service.mcp.McpServer;
import org.openmetadata.service.mcp.prompts.DefaultPromptsContext;
import org.openmetadata.service.mcp.tools.DefaultToolContext;
import org.openmetadata.service.migration.Migration;
import org.openmetadata.service.migration.MigrationValidationClient;
import org.openmetadata.service.migration.api.MigrationWorkflow;
@ -165,13 +162,11 @@ import org.quartz.SchedulerException;
@Server(url = "http://localhost:8585/api", description = "Endpoint URL")
},
security = @SecurityRequirement(name = "BearerAuth"))
@SecuritySchemes({
@SecurityScheme(
name = "BearerAuth",
type = SecuritySchemeType.HTTP,
scheme = "bearer",
bearerFormat = "JWT")
})
@SecurityScheme(
name = "BearerAuth",
type = SecuritySchemeType.HTTP,
scheme = "bearer",
bearerFormat = "JWT")
public class OpenMetadataApplication extends Application<OpenMetadataApplicationConfig> {
protected Authorizer authorizer;
private AuthenticatorHandler authenticatorHandler;
@ -328,9 +323,14 @@ public class OpenMetadataApplication extends Application<OpenMetadataApplication
OpenMetadataApplicationConfig catalogConfig, Environment environment) {
try {
if (ApplicationContext.getInstance().getAppIfExists("McpApplication") != null) {
McpServer mcpServer = new McpServer(new DefaultToolContext(), new DefaultPromptsContext());
Class<?> mcpServerClass = Class.forName("org.openmetadata.mcp.McpServer");
McpServerProvider mcpServer =
(McpServerProvider) mcpServerClass.getDeclaredConstructor().newInstance();
mcpServer.initializeMcpServer(environment, authorizer, limits, catalogConfig);
LOG.info("MCP Server registered successfully");
}
} catch (ClassNotFoundException ex) {
LOG.info("MCP module not found in classpath, skipping MCP server initialization");
} catch (Exception ex) {
LOG.error("Error initializing MCP server", ex);
}

26
openmetadata-service/src/main/java/org/openmetadata/service/apps/McpServerProvider.java Normal file
View File

@ -0,0 +1,26 @@
package org.openmetadata.service.apps;
import io.dropwizard.core.setup.Environment;
import org.openmetadata.service.OpenMetadataApplicationConfig;
import org.openmetadata.service.limits.Limits;
import org.openmetadata.service.security.Authorizer;
/**
* Interface for MCP Server Provider to avoid circular dependency.
* The actual implementation will be in openmetadata-mcp module.
*/
public interface McpServerProvider {
/**
* Initialize and register the MCP server with the application.
*
* @param environment Dropwizard environment
* @param authorizer Security authorizer
* @param limits Request limits
* @param config Application configuration
*/
void initializeMcpServer(
Environment environment,
Authorizer authorizer,
Limits limits,
OpenMetadataApplicationConfig config);
}

16
openmetadata-service/src/main/java/org/openmetadata/service/security/JwtFilter.java
View File

@ -39,9 +39,9 @@ import jakarta.ws.rs.core.MultivaluedMap;
import jakarta.ws.rs.core.SecurityContext;
import jakarta.ws.rs.core.UriInfo;
import jakarta.ws.rs.ext.Provider;
import java.net.URI;
import java.net.URL;
import java.security.interfaces.RSAPublicKey;
import java.util.Arrays;
import java.util.Calendar;
import java.util.HashSet;
import java.util.List;
@ -82,10 +82,6 @@ public class JwtFilter implements ContainerRequestFilter {
private boolean useRolesFromProvider = false;
private AuthenticationConfiguration.TokenValidationAlgorithm tokenValidationAlgorithm;
private static final List<String> DEFAULT_PUBLIC_KEY_URLS =
Arrays.asList(
"http://localhost:8585/api/v1/system/config/jwks",
"http://host.docker.internal:8585/api/v1/system/config/jwks");
public static final List<String> EXCLUDED_ENDPOINTS =
List.of(
"v1/system/config/jwks",
@ -120,16 +116,10 @@ public class JwtFilter implements ContainerRequestFilter {
ImmutableList.Builder<URL> publicKeyUrlsBuilder = ImmutableList.builder();
for (String publicKeyUrlStr : authenticationConfiguration.getPublicKeyUrls()) {
publicKeyUrlsBuilder.add(new URL(publicKeyUrlStr));
publicKeyUrlsBuilder.add(URI.create(publicKeyUrlStr).toURL());
}
// avoid users misconfiguration and add default publicKeyUrls
for (String publicKeyUrl : DEFAULT_PUBLIC_KEY_URLS) {
if (!authenticationConfiguration.getPublicKeyUrls().contains(publicKeyUrl)) {
publicKeyUrlsBuilder.add(new URL(publicKeyUrl));
}
}
this.jwkProvider = new MultiUrlJwkProvider(publicKeyUrlsBuilder.build());
this.principalDomain = authorizerConfiguration.getPrincipalDomain();
this.allowedDomains = authorizerConfiguration.getAllowedDomains();
this.enforcePrincipalDomain = authorizerConfiguration.getEnforcePrincipalDomain();

27
openmetadata-service/src/main/java/org/openmetadata/service/security/LocalJwkProvider.java Normal file
View File

@ -0,0 +1,27 @@
package org.openmetadata.service.security;
import com.auth0.jwk.Jwk;
import com.auth0.jwk.JwkProvider;
import java.util.Map;
import org.openmetadata.service.security.jwt.JWTTokenGenerator;
public class LocalJwkProvider implements JwkProvider {
private final Jwk self;
LocalJwkProvider() {
var jwtGen = JWTTokenGenerator.getInstance();
var key = jwtGen.getJWKSResponse().getJwsKeys().getFirst();
self =
Jwk.fromValues(
Map.of(
"kid", key.getKid(),
"kty", key.getKty(),
"n", key.getN(),
"e", key.getE()));
}
@Override
public Jwk get(String kid) {
return self;
}
}

18
openmetadata-service/src/main/java/org/openmetadata/service/security/MultiUrlJwkProvider.java
View File

@ -24,10 +24,12 @@ import com.google.common.cache.LoadingCache;
import java.net.URL;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.jetbrains.annotations.NotNull;
import org.openmetadata.service.exception.UnhandledServerException;
final class MultiUrlJwkProvider implements JwkProvider {
private final List<UrlJwkProvider> urlJwkProviders;
private final List<JwkProvider> jwkProviders;
private final LocalJwkProvider localJwkProvider = new LocalJwkProvider();
private final LoadingCache<String, Jwk> CACHE =
CacheBuilder.newBuilder()
.maximumSize(10)
@ -35,13 +37,18 @@ final class MultiUrlJwkProvider implements JwkProvider {
.build(
new CacheLoader<>() {
@Override
public Jwk load(String key) throws Exception {
public @NotNull Jwk load(@NotNull String key) throws Exception {
try {
return localJwkProvider.get(key);
} catch (Exception ignored) {
// ignore exception
}
JwkException lastException =
new SigningKeyNotFoundException(
"JWT Token keyID doesn't match the configured keyID. This usually happens if you didn't configure "
+ "proper publicKeyUrls under authentication configuration.",
null);
for (UrlJwkProvider jwkProvider : urlJwkProviders) {
for (JwkProvider jwkProvider : jwkProviders) {
try {
return jwkProvider.get(key);
} catch (JwkException e) {
@ -53,7 +60,10 @@ final class MultiUrlJwkProvider implements JwkProvider {
});
public MultiUrlJwkProvider(List<URL> publicKeyUris) {
this.urlJwkProviders = publicKeyUris.stream().map(UrlJwkProvider::new).toList();
this.jwkProviders =
publicKeyUris.stream()
.map(UrlJwkProvider::new)
.collect(java.util.stream.Collectors.toList());
}
@Override

1
pom.xml
View File

@ -33,6 +33,7 @@
<module>common</module>
<module>openmetadata-shaded-deps</module>
<module>openmetadata-service</module>
<module>openmetadata-mcp</module>
<module>openmetadata-ui</module>
<module>openmetadata-dist</module>
<module>openmetadata-clients</module>

76
yarn.lock
View File

@ -15,9 +15,9 @@
integrity sha512-tLjfhinr6doUBcWi7BWnkT2zT6G5UhiZftsiIH6xVvykeXE+FU7Wr0MyqwmqideWlDD5rG+VjVLptLviGo04CA==
"@glideapps/ts-necessities@^2.1.2":
version "2.3.2"
resolved "https://registry.yarnpkg.com/@glideapps/ts-necessities/-/ts-necessities-2.3.2.tgz#3e7a07f41c8c07527757631f25599a7b67d39d8c"
integrity sha512-tOXo3SrEeLu+4X2q6O2iNPXdGI1qoXEz/KrbkElTsWiWb69tFH4GzWz2K++0nBD6O3qO2Ft1C4L4ZvUfE2QDlQ==
version "2.4.0"
resolved "https://registry.yarnpkg.com/@glideapps/ts-necessities/-/ts-necessities-2.4.0.tgz#4f00312cabe0038eec6b82332916590724c7a679"
integrity sha512-mDC+qosuNa4lxR3ioMBb6CD0XLRsQBplU+zRPUYiMLXKeVPZ6UYphdNG/EGReig0YyfnVlBKZEXl1wzTotYmPA==
"@jridgewell/resolve-uri@^3.0.3":
version "3.1.2"
@ -25,9 +25,9 @@
integrity sha512-bRISgCIjP20/tbWSPWMEi54QVPRZExkuD9lJL+UIxUKtwVJA8wW1Trb1jMs1RFXo1CBTNZ/5hpC9QvmKWdopKw==
"@jridgewell/sourcemap-codec@^1.4.10":
version "1.5.0"
resolved "https://registry.yarnpkg.com/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.5.0.tgz#3188bcb273a414b0d215fd22a58540b989b9409a"
integrity sha512-gv3ZRaISU3fjPAgNsriBRqGWQL6quFx04YMPW/zD8XMLsU32mhCCbfbO6KZFLjvYpCZ8zyDEgqsgf+PwPaM7GQ==
version "1.5.3"
resolved "https://registry.yarnpkg.com/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.5.3.tgz#2d7bbc63fdfe4e4fa6b3478f651ec3844e284c1b"
integrity sha512-AiR5uKpFxP3PjO4R19kQGIMwxyRyPuXmKEEy301V1C0+1rVjS94EZQXf1QKZYN8Q0YM+estSPhmx5JwNftv6nw==
"@jridgewell/trace-mapping@0.3.9":
version "0.3.9"
@ -77,9 +77,9 @@
integrity sha512-5+fP8P8MFNC+AyZCDxrB2pkZFPGzqQWUzpSeuuVLvm8VMcorNYavBqoFcxK8bQz4Qsbn4oUEEem4wDLfcysGHA==
"@types/node@^16.9.2":
version "16.18.108"
resolved "https://registry.yarnpkg.com/@types/node/-/node-16.18.108.tgz#b794e2b2a85b4c12935ea7d0f18641be68b352f9"
integrity sha512-fj42LD82fSv6yN9C6Q4dzS+hujHj+pTv0IpRR3kI20fnYeS0ytBpjFO9OjmDowSPPt4lNKN46JLaKbCyP+BW2A==
version "16.18.126"
resolved "https://registry.yarnpkg.com/@types/node/-/node-16.18.126.tgz#27875faa2926c0f475b39a8bb1e546c0176f8d4b"
integrity sha512-OTcgaiwfGFBKacvfwuHzzn1KLxH/er8mluiy8/uM3sGXHaRe73RrSIj01jow9t4kJEW633Ov+cOexXeiApTyAw==
"@types/urijs@^1.19.19":
version "1.19.25"
@ -101,9 +101,9 @@ acorn-walk@^8.1.1:
acorn "^8.11.0"
acorn@^8.11.0, acorn@^8.4.1:
version "8.12.1"
resolved "https://registry.yarnpkg.com/acorn/-/acorn-8.12.1.tgz#71616bdccbe25e27a54439e0046e89ca76df2248"
integrity sha512-tcpGyI9zbizT9JbV6oYE477V6mTlXvvi0T0G3SNIYE2apm/G5huBa1+K89VGeovbg+jycCrfhl3ADxErOuO6Jg==
version "8.15.0"
resolved "https://registry.yarnpkg.com/acorn/-/acorn-8.15.0.tgz#a360898bc415edaac46c8241f6383975b930b816"
integrity sha512-NZyJarBfL7nWwIq+FDL6Zp/yHEhePMNnnJ0y3qfieCrmNvYct8uvtiV41UvlSe6apAfk0fY1FbWx+NwfmpvtTg==
ansi-regex@^5.0.1:
version "5.0.1"
@ -150,9 +150,9 @@ base64-js@^1.3.0, base64-js@^1.3.1:
integrity sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA==
brace-expansion@^1.1.7:
version "1.1.11"
resolved "https://registry.yarnpkg.com/brace-expansion/-/brace-expansion-1.1.11.tgz#3c7fcbf529d87226f3d2f52b966ff5271eb441dd"
integrity sha512-iCuPHDFgrHX7H2vEI/5xpz07zSHB00TpugqhmYtVmMO6518mCuRMoOYFldEBl0g187ufozdaHgWKcYFb61qGiA==
version "1.1.12"
resolved "https://registry.yarnpkg.com/brace-expansion/-/brace-expansion-1.1.12.tgz#ab9b454466e5a8cc3a187beaad580412a9c5b843"
integrity sha512-9T9UjW3r0UW5c1Q7GTwllptXwhvYmEzFhzMfZ9H7FQWt+uZePjZPjBP/W1ZEyZ1twGWom5/56TF4lPcqjnDHcg==
dependencies:
balanced-match "^1.0.0"
concat-map "0.0.1"
@ -248,7 +248,7 @@ command-line-usage@^6.1.3:
concat-map@0.0.1:
version "0.0.1"
resolved "https://registry.yarnpkg.com/concat-map/-/concat-map-0.0.1.tgz#d8a96bd77fd68df7793a73036a3ba0d5405d477b"
integrity sha1-2Klr13/Wjfd5OnMDajug1UBdR3s=
integrity sha512-/Srv4dswyQNBfohGpz9o6Yb3Gz3SrUDqBH5rTuhGR7ahtlbYKnVxw2bCFMRljaA7EXHaXZ8wsHdodFvbkhKmqg==
create-require@^1.1.0:
version "1.1.1"
@ -256,11 +256,11 @@ create-require@^1.1.0:
integrity sha512-dcKFX3jn0MpIaXjisoRvexIJVEKzaq7z2rZKxf+MSr9TkdmHmsU4m2lcLojrj/FHl8mk5VxMmYA+ftRkP/3oKQ==
cross-fetch@^3.1.5:
version "3.1.8"
resolved "https://registry.yarnpkg.com/cross-fetch/-/cross-fetch-3.1.8.tgz#0327eba65fd68a7d119f8fb2bf9334a1a7956f82"
integrity sha512-cvA+JwZoU0Xq+h6WkMvAUqPEYy92Obet6UdKLfW60qn99ftItKjB5T+BkyWOFWe2pUyfQ+IJHmpOTznqk1M6Kg==
version "3.2.0"
resolved "https://registry.yarnpkg.com/cross-fetch/-/cross-fetch-3.2.0.tgz#34e9192f53bc757d6614304d9e5e6fb4edb782e3"
integrity sha512-Q+xVJLoGOeIMXZmbUK4HYk+69cQH6LudR0Vu/pRm2YlU/hDV9CiS0gKUMaWY5f2NeUH9C1nV3bsTlCo0FsTV1Q==
dependencies:
node-fetch "^2.6.12"
node-fetch "^2.7.0"
deep-extend@~0.6.0:
version "0.6.0"
@ -307,7 +307,7 @@ find-replace@^3.0.0:
fs.realpath@^1.0.0:
version "1.0.0"
resolved "https://registry.yarnpkg.com/fs.realpath/-/fs.realpath-1.0.0.tgz#1504ad2523158caa40db4a2787cb01411994ea4f"
integrity sha1-FQStJSMVjKpA20onh8sBQRmU6k8=
integrity sha512-OO0pH2lK6a0hZnAdau5ItzHPI6pUlvI7jMVnxUQRtw4owF2wk8lOSabtGDCTP4Ggrg2MbGnWO9X8K1t4+fGMDw==
get-caller-file@^2.0.5:
version "2.0.5"
@ -351,7 +351,7 @@ ieee754@^1.2.1:
inflight@^1.0.4:
version "1.0.6"
resolved "https://registry.yarnpkg.com/inflight/-/inflight-1.0.6.tgz#49bd6331d7d02d0c09bc910a1075ba8165b56df9"
integrity sha1-Sb1jMdfQLQwJvJEKEHW6gWW1bfk=
integrity sha512-k92I/b08q4wvFscXCLvqfsHCrjrF7yiXsQuIVvVE7N82W3+aqpzuUdBbfhWcy/FZR3/4IgflMgKLOsvPDrGCJA==
dependencies:
once "^1.3.0"
wrappy "1"
@ -408,7 +408,7 @@ moment@^2.29.4:
resolved "https://registry.yarnpkg.com/moment/-/moment-2.30.1.tgz#f8c91c07b7a786e30c59926df530b4eac96974ae"
integrity sha512-uEmtNhbDOrWPFS+hdjFCBfy9f2YoyzRpwcl+DqpC6taX21FzsTLQVbMV/W7PzNSX6x/bhC1zA3c2UQ5NzH6how==
node-fetch@^2.6.12:
node-fetch@^2.7.0:
version "2.7.0"
resolved "https://registry.yarnpkg.com/node-fetch/-/node-fetch-2.7.0.tgz#d0f0fa6e3e2dc1d27efcd8ad99d550bda94d187d"
integrity sha512-c4FRfUm/dbcWZ7U+1Wq0AwCyFL+3nt2bEw05wfxSz+DWpWsitgmSgYmy2dQdWyKC1694ELPqMs/YzUSNozLt8A==
@ -418,14 +418,14 @@ node-fetch@^2.6.12:
once@^1.3.0:
version "1.4.0"
resolved "https://registry.yarnpkg.com/once/-/once-1.4.0.tgz#583b1aa775961d4b113ac17d9c50baef9dd76bd1"
integrity sha1-WDsap3WWHUsROsF9nFC6753Xa9E=
integrity sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==
dependencies:
wrappy "1"
pako@^0.2.5:
version "0.2.9"
resolved "https://registry.yarnpkg.com/pako/-/pako-0.2.9.tgz#f3f7522f4ef782348da8161bad9ecfd51bf83a75"
integrity sha1-8/dSL073gjSNqBYbrZ7P1Rv4OnU=
integrity sha512-NUcwaKxUxWrZLpDG+z/xZaCgQITkA/Dv4V/T6bw7VON6l1Xz/VnrBqrYjZQ12TamKHzITTfOEIYUj48y2KXImA==
pako@^1.0.6:
version "1.0.11"
@ -440,7 +440,7 @@ path-equal@^1.1.2:
path-is-absolute@^1.0.0:
version "1.0.1"
resolved "https://registry.yarnpkg.com/path-is-absolute/-/path-is-absolute-1.0.1.tgz#174b9268735534ffbc7ace6bf53a5a9e1b5c5f5f"
integrity sha1-F0uSaHNVNP+8es5r9TpanhtcX18=
integrity sha512-AVbw3UJ2e9bq64vSaS9Am0fje1Pa8pbGqTTsmXfaIiMpnr5DlDhfJOuLj9Sf95ZPVDAUerDfEk88MPmPe7UCQg==
pluralize@^8.0.0:
version "8.0.0"
@ -532,9 +532,9 @@ readable-stream@^3.4.0:
util-deprecate "^1.0.1"
readable-stream@^4.3.0:
version "4.5.2"
resolved "https://registry.yarnpkg.com/readable-stream/-/readable-stream-4.5.2.tgz#9e7fc4c45099baeed934bff6eb97ba6cf2729e09"
integrity sha512-yjavECdqeZ3GLXNgRXgeQEdz9fvDDkNKyHnbHRFtOr7/LcfgBcmct7t/ET+HaCTqfh06OzoAxrkN/IfjJBVe+g==
version "4.7.0"
resolved "https://registry.yarnpkg.com/readable-stream/-/readable-stream-4.7.0.tgz#cedbd8a1146c13dfff8dab14068028d58c15ac91"
integrity sha512-oIGGmcpTLwPga8Bn6/Z75SVaH1z5dUut2ibSyAMVhmUggWpmDn2dapB0n7f8nwaSiRtepAsfJyfXIO5DCVAODg==
dependencies:
abort-controller "^3.0.0"
buffer "^6.0.3"
@ -550,7 +550,7 @@ reduce-flatten@^2.0.0:
require-directory@^2.1.1:
version "2.1.1"
resolved "https://registry.yarnpkg.com/require-directory/-/require-directory-2.1.1.tgz#8c64ad5fd30dab1c976e2344ffe7f792a6a6df42"
integrity sha1-jGStX9MNqxyXbiNE/+f3kqam30I=
integrity sha512-fGxEI7+wsG9xrvdjsrlmL22OMTTiHRwAMroiEeMgq8gzoLC/PQr7RsRDSTLUg/bZAZtF+TVIkHc6/4RIKrui+Q==
safe-buffer@~5.2.0:
version "5.2.1"
@ -636,7 +636,7 @@ tiny-inflate@^1.0.0:
tr46@~0.0.3:
version "0.0.3"
resolved "https://registry.yarnpkg.com/tr46/-/tr46-0.0.3.tgz#8184fd347dac9cdc185992f3a6622e14b9d9ab6a"
integrity sha1-gYT9NH2snNwYWZLzpmIuFLnZq2o=
integrity sha512-N3WMsuqV66lT30CrXNbEjx4GEwlow3v6rr4mCcv6prnfwhS01rkgyFdjPNBYd9br7LpXV1+Emh01fHnq2Gdgrw==
ts-node@^10.9.1:
version "10.9.2"
@ -706,12 +706,12 @@ v8-compile-cache-lib@^3.0.1:
webidl-conversions@^3.0.0:
version "3.0.1"
resolved "https://registry.yarnpkg.com/webidl-conversions/-/webidl-conversions-3.0.1.tgz#24534275e2a7bc6be7bc86611cc16ae0a5654871"
integrity sha1-JFNCdeKnvGvnvIZhHMFq4KVlSHE=
integrity sha512-2JAn3z8AR6rjK8Sm8orRC0h/bcl/DqL7tRPdGZ4I1CjdF+EaMLmYxBHyXuKL849eucPFhvBoxMsflfOb8kxaeQ==
whatwg-url@^5.0.0:
version "5.0.0"
resolved "https://registry.yarnpkg.com/whatwg-url/-/whatwg-url-5.0.0.tgz#966454e8765462e37644d3626f6742ce8b70965d"
integrity sha1-lmRU6HZUYuN2RNNib2dCzotwll0=
integrity sha512-saE57nupxk6v3HY35+jzBwYa0rKSy0XR8JSxZPwgLr7ys0IBzhGviA1/TUGJLmSVqs8pb9AnvICXEuOHLprYTw==
dependencies:
tr46 "~0.0.3"
webidl-conversions "^3.0.0"
@ -719,7 +719,7 @@ whatwg-url@^5.0.0:
wordwrap@^1.0.0:
version "1.0.0"
resolved "https://registry.yarnpkg.com/wordwrap/-/wordwrap-1.0.0.tgz#27584810891456a4171c8d0226441ade90cbcaeb"
integrity sha1-J1hIEIkUVqQXHI0CJkQa3pDLyus=
integrity sha512-gvVzJFlPycKc5dZN4yPkP8w7Dc37BtP1yczEneOb4uq34pXZcvrtRTmWV8W+Ume+XCxKgbjM+nevkyFPMybd4Q==
wordwrapjs@^4.0.0:
version "4.0.1"
@ -741,7 +741,7 @@ wrap-ansi@^7.0.0:
wrappy@1:
version "1.0.2"
resolved "https://registry.yarnpkg.com/wrappy/-/wrappy-1.0.2.tgz#b5243d8f3ec1aa35f1364605bc0d1036e30ab69f"
integrity sha1-tSQ9jz7BqjXxNkYFvA0QNuMKtp8=
integrity sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==
y18n@^5.0.5:
version "5.0.8"
@ -749,9 +749,9 @@ y18n@^5.0.5:
integrity sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA==
yaml@^2.2.1:
version "2.5.1"
resolved "https://registry.yarnpkg.com/yaml/-/yaml-2.5.1.tgz#c9772aacf62cb7494a95b0c4f1fb065b563db130"
integrity sha512-bLQOjaX/ADgQ20isPJRvF0iRUHIxVhYvr53Of7wGcWlO2jvtUlH5m87DsmulFVxRpNLOnI4tB6p/oh8D7kpn9Q==
version "2.8.0"
resolved "https://registry.yarnpkg.com/yaml/-/yaml-2.8.0.tgz#15f8c9866211bdc2d3781a0890e44d4fa1a5fff6"
integrity sha512-4lLa/EcQCB0cJkyts+FpIRx5G/llPxfP6VQU5KByHEhLxY3IJCH0f0Hy1MHI8sClTvsIb8qwRJ6R/ZdlDJ/leQ==
yargs-parser@^21.1.1:
version "21.1.1"