Skip to content

Ean Schuessler / mo-mcp

Go to a project
Toggle navigation
Toggle navigation pinning
959cd392 by Ean Schuessler
Options

Consolidate MCP implementation with shared JSON-RPC classes 

- Extract JsonRpcMessage classes to separate file for better code organization
- Remove deprecated McpSessionManager (unused, replaced by Visit-based sessions)
- Remove problematic ServiceBasedMcpServlet (async limitations, service invocation bugs)
- Enhance EnhancedMcpServlet with configuration parameters and improved monitoring
- Add broadcast success/failure counting and helper methods
- Fix variable scope issue with requestBody in JSON-RPC handler
- Consolidate to single, working MCP servlet implementation

Working features:
- Authentication with Basic auth
- SSE connections with proper session management
- JSON-RPC protocol (ping, initialize, tools/list)
- Visit-based session persistence
- Service delegation to McpServices.xml
1 parent 339ee0ef
Expand all
Inline Side-by-side
Showing 4 changed files with 116 additions and 326 deletions
......@@ -32,46 +32,7 @@ import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicBoolean
import java.util.UUID
/**
* Simple JSON-RPC Message classes for MCP compatibility
*/
class JsonRpcMessage {
String jsonrpc = "2.0"
}
class JsonRpcResponse extends JsonRpcMessage {
Object id
Object result
Map error
JsonRpcResponse(Object result, Object id) {
this.result = result
this.id = id
}
JsonRpcResponse(Map error, Object id) {
this.error = error
this.id = id
}
String toJson() {
return JsonOutput.toJson(this)
}
}
class JsonRpcNotification extends JsonRpcMessage {
String method
Object params
JsonRpcNotification(String method, Object params = null) {
this.method = method
this.params = params
}
String toJson() {
return JsonOutput.toJson(this)
}
}
/**
* Enhanced MCP Servlet with proper SSE handling inspired by HttpServletSseServerTransportProvider
......@@ -88,12 +49,28 @@ class EnhancedMcpServlet extends HttpServlet {
// Session management using Moqui's Visit system directly
// No need for separate session manager - Visit entity handles persistence
// Configuration parameters
private String sseEndpoint = "/sse"
private String messageEndpoint = "/message"
private int keepAliveIntervalSeconds = 30
private int maxConnections = 100
@Override
void init(ServletConfig config) throws ServletException {
super.init(config)
// Read configuration from servlet init parameters
sseEndpoint = config.getInitParameter("sseEndpoint") ?: sseEndpoint
messageEndpoint = config.getInitParameter("messageEndpoint") ?: messageEndpoint
keepAliveIntervalSeconds = config.getInitParameter("keepAliveIntervalSeconds")?.toInteger() ?: keepAliveIntervalSeconds
maxConnections = config.getInitParameter("maxConnections")?.toInteger() ?: maxConnections
String webappName = config.getInitParameter("moqui-name") ?:
config.getServletContext().getInitParameter("moqui-name")
logger.info("EnhancedMcpServlet initialized for webapp ${webappName}")
logger.info("SSE endpoint: ${sseEndpoint}, Message endpoint: ${messageEndpoint}")
logger.info("Keep-alive interval: ${keepAliveIntervalSeconds}s, Max connections: ${maxConnections}")
}
@Override
......@@ -542,11 +519,6 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
logger.info("Enhanced MCP JSON-RPC Request: ${method} ${request.requestURI} - Accept: ${acceptHeader}, Content-Type: ${contentType}")
// Log request body for debugging (be careful with this in production)
if (requestBody?.length() > 0) {
logger.info("MCP JSON-RPC request body: ${requestBody}")
}
// Handle POST requests for JSON-RPC
if (!"POST".equals(method)) {
response.setStatus(HttpServletResponse.SC_METHOD_NOT_ALLOWED)
......@@ -593,6 +565,11 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
return
}
// Log request body for debugging (be careful with this in production)
if (requestBody.length() > 0) {
logger.info("MCP JSON-RPC request body: ${requestBody}")
}
def rpcRequest
try {
rpcRequest = jsonSlurper.parseText(requestBody)
......@@ -777,25 +754,55 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
logger.info("Broadcasting to ${mcpVisits.size()} MCP visits, ${activeConnections.size()} active connections")
int successCount = 0
int failureCount = 0
// Send to active connections (transient)
mcpVisits.each { visit ->
PrintWriter writer = activeConnections.get(visit.visitId)
if (writer && !writer.checkError()) {
try {
sendSseEvent(writer, "broadcast", message.toJson())
successCount++
} catch (Exception e) {
logger.warn("Failed to send broadcast to ${visit.visitId}: ${e.message}")
// Remove broken connection
activeConnections.remove(visit.visitId)
failureCount++
}
} else {
// No active connection for this visit
failureCount++
}
}
logger.info("Broadcast completed: ${successCount} successful, ${failureCount} failed")
} catch (Exception e) {
logger.error("Error broadcasting to all sessions: ${e.message}", e)
}
}
/**
* Send SSE event to specific session (helper method)
*/
void sendToSession(String sessionId, JsonRpcMessage message) {
try {
PrintWriter writer = activeConnections.get(sessionId)
if (writer && !writer.checkError()) {
sendSseEvent(writer, "message", message.toJson())
logger.debug("Sent message to session ${sessionId}")
} else {
logger.warn("No active connection for session ${sessionId}")
}
} catch (Exception e) {
logger.error("Error sending message to session ${sessionId}: ${e.message}", e)
// Remove broken connection
activeConnections.remove(sessionId)
}
}
/**
* Get session statistics for monitoring
*/
Map getSessionStatistics() {
......@@ -808,12 +815,22 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
return [
totalMcpVisits: mcpVisits.size(),
activeConnections: activeConnections.size(),
maxConnections: maxConnections,
architecture: "Visit-based sessions with connection registry",
message: "Enhanced MCP with session tracking"
message: "Enhanced MCP with session tracking",
endpoints: [
sse: sseEndpoint,
message: messageEndpoint
],
keepAliveInterval: keepAliveIntervalSeconds
]
} catch (Exception e) {
logger.error("Error getting session statistics: ${e.message}", e)
return [activeSessions: activeConnections.size(), error: e.message]
return [
activeConnections: activeConnections.size(),
maxConnections: maxConnections,
error: e.message
]
}
}
}
\ No newline at end of file
......
/*
* This software is in the public domain under CC0 1.0 Universal plus a
* Grant of Patent License.
*
* To the extent possible under law, author(s) have dedicated all
* copyright and related and neighboring rights to this software to the
* public domain worldwide. This software is distributed without any
* warranty.
*
* You should have received a copy of the CC0 Public Domain Dedication
* along with this software (see the LICENSE.md file). If not, see
* <http://creativecommons.org/publicdomain/zero/1.0/>.
*/
package org.moqui.mcp
import groovy.json.JsonOutput
/**
* Simple JSON-RPC Message classes for MCP compatibility
*/
class JsonRpcMessage {
String jsonrpc = "2.0"
String toJson() {
return JsonOutput.toJson(this)
}
}
class JsonRpcResponse extends JsonRpcMessage {
Object id
Object result
Map error
JsonRpcResponse(Object result, Object id) {
this.result = result
this.id = id
}
JsonRpcResponse(Map error, Object id) {
this.error = error
this.id = id
}
}
class JsonRpcNotification extends JsonRpcMessage {
String method
Object params
JsonRpcNotification(String method, Object params = null) {
this.method = method
this.params = params
}
}
\ No newline at end of file
/*
* This software is in the public domain under CC0 1.0 Universal plus a
* Grant of Patent License.
*
* To the extent possible under law, author(s) have dedicated all
* copyright and related and neighboring rights to this software to the
* public domain worldwide. This software is distributed without any
* warranty.
*
* You should have received a copy of the CC0 Public Domain Dedication
* along with this software (see the LICENSE.md file). If not, see
* <http://creativecommons.org/publicdomain/zero/1.0/>.
*/
package org.moqui.mcp
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
/**
* MCP Session Manager with SDK-style capabilities
*
* @deprecated This class is deprecated. Use Moqui's Visit entity directly for session management.
* See VisitBasedMcpSession for the new Visit-based approach.
*
* Provides centralized session management, broadcasting, and graceful shutdown
*/
@Deprecated
class McpSessionManager {
protected final static Logger logger = LoggerFactory.getLogger(McpSessionManager.class)
private final Map<String, VisitBasedMcpSession> sessions = new ConcurrentHashMap<>()
private final AtomicBoolean isShuttingDown = new AtomicBoolean(false)
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2)
// Session cleanup and monitoring
private final long sessionTimeoutMs = 30 * 60 * 1000 // 30 minutes
private final long cleanupIntervalMs = 5 * 60 * 1000 // 5 minutes
McpSessionManager() {
// Start periodic cleanup task
scheduler.scheduleAtFixedRate(this::cleanupInactiveSessions,
cleanupIntervalMs, cleanupIntervalMs, TimeUnit.MILLISECONDS)
logger.info("MCP Session Manager initialized")
}
/**
* Register a new session
*/
void registerSession(VisitBasedMcpSession session) {
if (isShuttingDown.get()) {
logger.warn("Rejecting session registration during shutdown: ${session.sessionId}")
return
}
sessions.put(session.sessionId, session)
logger.info("Registered MCP session ${session.sessionId} (total: ${sessions.size()})")
// Send welcome message to new session
def welcomeMessage = new JsonRpcNotification("welcome", [
sessionId: session.sessionId,
totalSessions: sessions.size(),
timestamp: System.currentTimeMillis()
])
session.sendMessage(welcomeMessage)
}
/**
* Unregister a session
*/
void unregisterSession(String sessionId) {
def session = sessions.remove(sessionId)
if (session) {
logger.info("Unregistered MCP session ${sessionId} (remaining: ${sessions.size()})")
}
}
/**
* Get session by ID
*/
VisitBasedMcpSession getSession(String sessionId) {
return sessions.get(sessionId)
}
/**
* Broadcast message to all active sessions
*/
void broadcast(JsonRpcMessage message) {
if (isShuttingDown.get()) {
logger.warn("Rejecting broadcast during shutdown")
return
}
def inactiveSessions = []
def activeCount = 0
sessions.values().each { session ->
try {
if (session.isActive()) {
session.sendMessage(message)
activeCount++
} else {
inactiveSessions << session.sessionId
}
} catch (Exception e) {
logger.warn("Error broadcasting to session ${session.sessionId}: ${e.message}")
inactiveSessions << session.sessionId
}
}
// Clean up inactive sessions
inactiveSessions.each { sessionId ->
unregisterSession(sessionId)
}
logger.info("Broadcast message to ${activeCount} active sessions (removed ${inactiveSessions.size()} inactive)")
}
/**
* Send message to specific session
*/
boolean sendToSession(String sessionId, JsonRpcMessage message) {
def session = sessions.get(sessionId)
if (!session) {
return false
}
try {
if (session.isActive()) {
session.sendMessage(message)
return true
} else {
unregisterSession(sessionId)
return false
}
} catch (Exception e) {
logger.warn("Error sending to session ${sessionId}: ${e.message}")
unregisterSession(sessionId)
return false
}
}
/**
* Get session statistics
*/
Map getSessionStatistics() {
def stats = [
totalSessions: sessions.size(),
activeSessions: 0,
closingSessions: 0,
isShuttingDown: isShuttingDown.get(),
uptime: System.currentTimeMillis() - (this.@startTime ?: System.currentTimeMillis()),
sessions: []
]
sessions.values().each { session ->
def sessionStats = session.getSessionStats()
stats.sessions << sessionStats
if (sessionStats.active) {
stats.activeSessions++
}
if (sessionStats.closing) {
stats.closingSessions++
}
}
return stats
}
/**
* Initiate graceful shutdown
*/
void shutdownGracefully() {
if (!isShuttingDown.compareAndSet(false, true)) {
return // Already shutting down
}
logger.info("Initiating graceful MCP session manager shutdown")
// Send shutdown notification to all sessions
def shutdownMessage = new JsonRpcNotification("server_shutdown", [
message: "Server is shutting down gracefully",
timestamp: System.currentTimeMillis()
])
broadcast(shutdownMessage)
// Give sessions time to receive shutdown message
scheduler.schedule({
forceShutdown()
}, 5, TimeUnit.SECONDS)
}
/**
* Force immediate shutdown
*/
void forceShutdown() {
logger.info("Force shutting down MCP session manager")
// Close all sessions
sessions.values().each { session ->
try {
session.close()
} catch (Exception e) {
logger.warn("Error closing session ${session.sessionId}: ${e.message}")
}
}
sessions.clear()
// Shutdown scheduler
scheduler.shutdown()
try {
if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
scheduler.shutdownNow()
}
} catch (InterruptedException e) {
scheduler.shutdownNow()
Thread.currentThread().interrupt()
}
logger.info("MCP session manager shutdown complete")
}
/**
* Clean up inactive sessions
*/
private void cleanupInactiveSessions() {
if (isShuttingDown.get()) {
return
}
def now = System.currentTimeMillis()
def inactiveSessions = []
sessions.values().each { session ->
def sessionStats = session.getSessionStats()
def inactiveTime = now - (sessionStats.lastActivity ?: sessionStats.createdAt.time)
if (!session.isActive() || inactiveTime > sessionTimeoutMs) {
inactiveSessions << session.sessionId
}
}
inactiveSessions.each { sessionId ->
def session = sessions.get(sessionId)
if (session) {
try {
session.closeGracefully()
} catch (Exception e) {
logger.warn("Error during cleanup of session ${sessionId}: ${e.message}")
}
unregisterSession(sessionId)
}
}
if (inactiveSessions.size() > 0) {
logger.info("Cleaned up ${inactiveSessions.size()} inactive MCP sessions")
}
}
/**
* Get active session count
*/
int getActiveSessionCount() {
return (int) sessions.values().count { it.isActive() }
}
/**
* Check if manager is shutting down
*/
boolean isShuttingDown() {
return isShuttingDown.get()
}
}
\ No newline at end of file