fdb76042 by Ean Schuessler

Implement MCP notification system for server-to-client communication

- Add notification queue mechanism in EnhancedMcpServlet
- Register servlet instance for service access
- Implement queueNotification() method for async notifications
- Add notification delivery in JSON-RPC responses
- Include notifications in both tools/list and tools/call services
- Add execution timing and success metrics
- Handle notification errors gracefully
- Support both polling and SSE delivery methods

This completes the MCP server notification system allowing real-time
communication about tool execution status and metrics.
1 parent 0f3c1f68
......@@ -180,7 +180,10 @@
}
}
}
*/
*/
// Start timing for execution metrics
def startTime = System.currentTimeMillis()
// Store original user context before switching to ADMIN
def originalUsername = ec.user.username
......@@ -358,13 +361,34 @@
result.nextCursor = String.valueOf(endIndex)
}
ec.logger.info("MCP ToolsList: Returning ${availableTools.size()} tools for user ${originalUsername}")
ec.logger.info("MCP ToolsList: Returning ${availableTools.size()} tools for user ${originalUsername}")
} finally {
// Always restore original user context
if (adminUserInfo != null) {
ec.user.popUser()
}
// Send a simple notification about tool execution
try {
def servlet = ec.web.getServletContext().getAttribute("enhancedMcpServlet")
ec.logger.info("TOOLS CALL: Got servlet reference: ${servlet != null}, sessionId: ${sessionId}")
if (servlet && sessionId) {
def notification = [
method: "notifications/tool_execution",
params: [
toolName: "tools/list",
executionTime: (System.currentTimeMillis() - startTime) / 1000.0,
success: !result?.result?.isError,
timestamp: System.currentTimeMillis()
]
]
servlet.queueNotification(sessionId, notification)
ec.logger.info("Queued tool execution notification for session ${sessionId}")
}
} catch (Exception e) {
ec.logger.warn("Failed to send tool execution notification: ${e.message}")
}
}
]]></script>
</actions>
......@@ -511,6 +535,27 @@
if (adminUserInfo != null) {
ec.user.popUser()
}
// Send a simple notification about tool execution
try {
def servlet = ec.web.getServletContext().getAttribute("enhancedMcpServlet")
ec.logger.info("TOOLS CALL: Got servlet reference: ${servlet != null}, sessionId: ${sessionId}")
if (servlet && sessionId) {
def notification = [
method: "notifications/tool_execution",
params: [
toolName: name,
executionTime: (System.currentTimeMillis() - startTime) / 1000.0,
success: !result?.result?.isError,
timestamp: System.currentTimeMillis()
]
]
servlet.queueNotification(sessionId, notification)
ec.logger.info("Queued tool execution notification for session ${sessionId}")
}
} catch (Exception e) {
ec.logger.warn("Failed to send tool execution notification: ${e.message}")
}
}
]]></script>
</actions>
......
......@@ -56,6 +56,9 @@ class EnhancedMcpServlet extends HttpServlet {
// No need for separate session manager - Visit entity handles persistence
private final Map<String, Integer> sessionStates = new ConcurrentHashMap<>()
// Notification queue for server-initiated notifications (for non-SSE clients)
private final Map<String, List<Map>> notificationQueues = new ConcurrentHashMap<>()
// Configuration parameters
private String sseEndpoint = "/sse"
private String messageEndpoint = "/message"
......@@ -75,9 +78,13 @@ class EnhancedMcpServlet extends HttpServlet {
String webappName = config.getInitParameter("moqui-name") ?:
config.getServletContext().getInitParameter("moqui-name")
// Register servlet instance in context for service access
config.getServletContext().setAttribute("enhancedMcpServlet", this)
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}")
logger.info("Servlet instance registered in context as 'enhancedMcpServlet'")
}
@Override
......@@ -828,6 +835,17 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
result: actualResult
]
// Check for pending server notifications and include them in response
if (sessionId && notificationQueues.containsKey(sessionId)) {
def pendingNotifications = notificationQueues.get(sessionId)
if (pendingNotifications && !pendingNotifications.isEmpty()) {
rpcResponse.notifications = pendingNotifications
// Clear delivered notifications
notificationQueues.put(sessionId, [])
logger.info("Delivered ${pendingNotifications.size()} pending notifications to session ${sessionId}")
}
}
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
......@@ -872,6 +890,10 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
params.actualUserId = ec.user.userId
logger.info("Initialize - actualUserId: ${params.actualUserId}, sessionId: ${params.sessionId}")
def serviceResult = callMcpService("mcp#Initialize", params, ec)
// Add sessionId to the response for mcp.sh compatibility
if (serviceResult && serviceResult.result) {
serviceResult.result.sessionId = params.sessionId
}
return serviceResult
case "ping":
// Simple ping for testing - bypass service for now
......@@ -983,6 +1005,28 @@ logger.info("Handling Enhanced SSE connection from ${request.remoteAddr}")
return false
}
/**
* Queue a server notification for delivery to client
*/
void queueNotification(String sessionId, Map notification) {
if (!sessionId) return
def queue = notificationQueues.computeIfAbsent(sessionId) { [] }
queue << notification
logger.info("Queued notification for session ${sessionId}: ${notification}")
// Also try to send via SSE if active connection exists
def writer = activeConnections.get(sessionId)
if (writer && !writer.checkError()) {
try {
sendSseEvent(writer, "notification", JsonOutput.toJson(notification), System.currentTimeMillis())
logger.info("Sent notification via SSE to session ${sessionId}")
} catch (Exception e) {
logger.warn("Failed to send notification via SSE to session ${sessionId}: ${e.message}")
}
}
}
@Override
void destroy() {
logger.info("Destroying EnhancedMcpServlet")
......