Skip to content

Ean Schuessler / mo-mcp

Go to a project
Toggle navigation
Toggle navigation pinning
3a39f97b by Ean Schuessler
Options

Refactor MCP servlet to adapter architecture 

Introduce clean adapter layer between Moqui infrastructure and MCP protocol:

- transport/MoquiMcpTransport: Interface abstracting transport concerns
- transport/SseTransport: SSE implementation with session management
- adapter/McpSessionAdapter: Maps Moqui Visit to MCP sessions
- adapter/McpToolAdapter: Maps MCP tools/methods to Moqui services
- adapter/MoquiNotificationMcpBridge: Bridges Moqui notifications to MCP

Simplify EnhancedMcpServlet to orchestrator role, removing inline session
management, SSE logic, and tool dispatch. Remove redundant session
validation in Initialize service (MoquiAuthFilter handles auth).

Delete obsolete files:
- VisitBasedMcpSession.groovy (replaced by McpSessionAdapter)
- JsonRpcMessage.groovy (using plain Maps)
- MoquiMcpTransport.groovy (replaced by new interface)
1 parent 9aefba5e
Inline Side-by-side
Showing 10 changed files with 1589 additions and 1051 deletions
......@@ -37,46 +37,11 @@
// Permissions are handled by Moqui's artifact authorization system
// Users must be in appropriate groups (McpUser, MCP_BUSINESS) with access to McpServices artifact group
// Disable authz to prevent automatic Visit updates during MCP operations
// Authentication is handled by MoquiAuthFilter - user context is already set
// No need to re-validate session ownership here
ec.artifactExecution.disableAuthz()
// Get Visit (session) created by servlet and validate access
def visit = ec.entity.find("moqui.server.Visit")
.condition("visitId", sessionId)
.one()
if (!visit) {
throw new Exception("Invalid session: ${sessionId}")
}
if (visit.userId != ec.user.userId) {
throw new Exception("Access denied for session: ${sessionId}")
}
// Update Visit with MCP initialization data
UserInfo adminUserInfo = null
try {
adminUserInfo = ec.user.pushUser("ADMIN")
def metadata = [:]
try {
metadata = groovy.json.JsonSlurper().parseText(visit.initialRequest ?: "{}") as Map
} catch (Exception e) {
ec.logger.debug("Failed to parse Visit metadata: ${e.message}")
}
metadata.mcpInitialized = true
metadata.mcpProtocolVersion = protocolVersion
metadata.mcpCapabilities = capabilities
metadata.mcpClientInfo = clientInfo
metadata.mcpInitializedAt = System.currentTimeMillis()
// Session metadata stored in memory only - no Visit updates to prevent lock contention
ec.logger.info("SESSIONID: ${sessionId} - metadata stored in memory")
} finally {
if (adminUserInfo != null) {
ec.user.popUser()
}
}
ec.logger.info("MCP Initialize for session ${sessionId}, user ${ec.user.userId}")
// Validate protocol version - support common MCP versions with version negotiation
def supportedVersions = ["2025-11-25", "2025-06-18", "2024-11-05", "2024-10-07", "2023-06-05"]
......
/*
* This software is in the public domain under CC0 1.0 Universal plus a
* 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/>.
......@@ -14,14 +14,17 @@
package org.moqui.mcp
import groovy.json.JsonSlurper
import org.moqui.impl.context.ExecutionContextFactoryImpl
import groovy.json.JsonBuilder
import groovy.json.JsonOutput
import org.moqui.impl.context.ExecutionContextFactoryImpl
import org.moqui.context.ArtifactAuthorizationException
import org.moqui.context.ArtifactTarpitException
import org.moqui.impl.context.ExecutionContextImpl
import org.moqui.entity.EntityValue
import org.moqui.context.ExecutionContext
import org.moqui.mcp.adapter.McpSessionAdapter
import org.moqui.mcp.adapter.McpSession
import org.moqui.mcp.adapter.McpToolAdapter
import org.moqui.mcp.adapter.MoquiNotificationMcpBridge
import org.moqui.mcp.transport.SseTransport
import org.slf4j.Logger
import org.slf4j.LoggerFactory
......@@ -30,102 +33,93 @@ import jakarta.servlet.ServletException
import jakarta.servlet.http.HttpServlet
import jakarta.servlet.http.HttpServletRequest
import jakarta.servlet.http.HttpServletResponse
import java.sql.Timestamp
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicBoolean
import java.util.UUID
/**
* Enhanced MCP Servlet with proper SSE handling inspired by HttpServletSseServerTransportProvider
* This implementation provides better SSE support and session management.
* Enhanced MCP Servlet with adapter-based architecture.
* Uses adapters for session management, tool dispatch, and notifications.
* This servlet acts as an orchestrator, delegating to specialized adapters.
*/
class EnhancedMcpServlet extends HttpServlet {
protected final static Logger logger = LoggerFactory.getLogger(EnhancedMcpServlet.class)
private JsonSlurper jsonSlurper = new JsonSlurper()
// Session state constants
private static final int STATE_UNINITIALIZED = 0
private static final int STATE_INITIALIZING = 1
private static final int STATE_INITIALIZED = 2
// Simple registry for active connections only (transient HTTP connections)
private final Map<String, PrintWriter> activeConnections = new ConcurrentHashMap<>()
// Session management using Moqui's Visit system directly
// No need for separate session manager - Visit entity handles persistence
private final Map<String, Integer> sessionStates = new ConcurrentHashMap<>()
// Message storage for notifications/subscribe and notifications/unsubscribe
private final Map<String, List<Map>> sessionMessages = new ConcurrentHashMap<>()
// In-memory session tracking to avoid database access for read operations
private final Map<String, String> sessionUsers = new ConcurrentHashMap<>()
// Progress tracking for notifications/progress
private final Map<String, Map> sessionProgress = new ConcurrentHashMap<>()
// Adapter instances
private McpSessionAdapter sessionAdapter
private McpToolAdapter toolAdapter
private SseTransport transport
private MoquiNotificationMcpBridge notificationBridge
// Visit cache to reduce database access and prevent lock contention
private final Map<String, EntityValue> visitCache = new ConcurrentHashMap<>()
// Notification queue for server-initiated notifications (for non-SSE clients)
private static final Map<String, List<Map>> notificationQueues = new ConcurrentHashMap<>()
// Throttled session activity tracking to prevent database lock contention
private final Map<String, Long> lastActivityUpdate = new ConcurrentHashMap<>()
private final Map<String, EntityValue> visitCache = new java.util.concurrent.ConcurrentHashMap<>()
// Throttled session activity tracking
private final Map<String, Long> lastActivityUpdate = new java.util.concurrent.ConcurrentHashMap<>()
private static final long ACTIVITY_UPDATE_INTERVAL_MS = 30000 // 30 seconds
// Session-specific locks to avoid sessionId.intern() deadlocks
private final Map<String, Object> sessionLocks = new ConcurrentHashMap<>()
// 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)
// Initialize adapters
sessionAdapter = new McpSessionAdapter()
toolAdapter = new McpToolAdapter()
transport = new SseTransport(sessionAdapter)
// Initialize notification bridge
notificationBridge = new MoquiNotificationMcpBridge()
// 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") ?:
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}")
// Get ECF and register notification bridge
ExecutionContextFactoryImpl ecfi =
(ExecutionContextFactoryImpl) config.getServletContext().getAttribute("executionContextFactory")
if (ecfi) {
notificationBridge.init(ecfi)
notificationBridge.setTransport(transport)
ecfi.registerNotificationMessageListener(notificationBridge)
logger.info("Registered MoquiNotificationMcpBridge with ECF")
}
logger.info("EnhancedMcpServlet initialized with adapter architecture 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
void service(HttpServletRequest request, HttpServletResponse response)
void service(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
ExecutionContextFactoryImpl ecfi =
ExecutionContextFactoryImpl ecfi =
(ExecutionContextFactoryImpl) getServletContext().getAttribute("executionContextFactory")
String webappName = getInitParameter("moqui-name") ?:
String webappName = getInitParameter("moqui-name") ?:
getServletContext().getInitParameter("moqui-name")
if (ecfi == null || webappName == null) {
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
"System is initializing, try again soon.")
return
}
// Handle CORS
if (handleCors(request, response, webappName, ecfi)) return
if (handleCors(request, response)) return
long startTime = System.currentTimeMillis()
if (logger.traceEnabled) {
......@@ -137,9 +131,9 @@ class EnhancedMcpServlet extends HttpServlet {
logger.warn("No ExecutionContext found from MoquiAuthFilter, creating new one")
ec = ecfi.getEci()
}
try {
// Read request body VERY early before any other processing can consume it
// Read request body early before any other processing can consume it
String requestBody = null
if ("POST".equals(request.getMethod())) {
try {
......@@ -178,7 +172,7 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Get Visit created by web facade
def visit = ec.user.getVisit()
if (!visit) {
......@@ -186,7 +180,7 @@ class EnhancedMcpServlet extends HttpServlet {
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Failed to create Visit")
return
}
// Route based on request method and path
String requestURI = request.getRequestURI()
String method = request.getMethod()
......@@ -197,17 +191,14 @@ class EnhancedMcpServlet extends HttpServlet {
} else if ("POST".equals(method) && requestURI.endsWith("/message")) {
handleMessage(request, response, ec, requestBody)
} else if ("POST".equals(method) && (requestURI.equals("/mcp") || requestURI.endsWith("/mcp"))) {
// Handle POST requests to /mcp for JSON-RPC
logger.debug("About to call handleJsonRpc with visit: ${visit?.visitId}")
handleJsonRpc(request, response, ec, webappName, requestBody, visit)
} else if ("GET".equals(method) && (requestURI.equals("/mcp") || requestURI.endsWith("/mcp"))) {
// Handle GET requests to /mcp - SSE connection for streaming
handleSseConnection(request, response, ec, webappName)
} else {
// Fallback to JSON-RPC handling
handleJsonRpc(request, response, ec, webappName, requestBody, visit)
}
} catch (ArtifactAuthorizationException e) {
logger.warn("Enhanced MCP Access Forbidden (no authz): " + e.message)
response.setStatus(HttpServletResponse.SC_FORBIDDEN)
......@@ -230,207 +221,158 @@ class EnhancedMcpServlet extends HttpServlet {
logger.error("Error in Enhanced MCP request", t)
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR)
response.setContentType("application/json")
// Use simple JSON string to avoid Groovy JSON library issues
def errorMsg = t.message?.toString() ?: "Unknown error"
response.writer.write("{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"Internal error: ${errorMsg.replace("\"", "\\\"")}\"},\"id\":null}")
}
}
private void handleSseConnection(HttpServletRequest request, HttpServletResponse response, ExecutionContextImpl ec, String webappName)
throws IOException {
logger.debug("Handling Enhanced SSE connection from ${request.remoteAddr}")
// Check for existing session ID first
// Check for existing session ID
String sessionId = request.getHeader("Mcp-Session-Id")
def visit = null
// If we have a session ID, validate using in-memory tracking
String userId = ec.user.userId?.toString()
// If we have a session ID, validate it
if (sessionId) {
try {
String sessionUser = sessionUsers.get(sessionId)
if (sessionUser) {
// Verify user has access to this session using in-memory data
if (!ec.user.userId || sessionUser != ec.user.userId.toString()) {
logger.warn("Session userId ${sessionUser} doesn't match current user userId ${ec.user.userId} - access denied")
response.sendError(HttpServletResponse.SC_FORBIDDEN, "Access denied for session: " + sessionId)
return
}
// Get Visit from cache for activity updates (but not for validation)
visit = getCachedVisit(ec, sessionId)
} else {
logger.warn("Session not found in memory: ${sessionId}")
response.sendError(HttpServletResponse.SC_NOT_FOUND, "Session not found: " + sessionId)
def session = sessionAdapter.getSession(sessionId)
if (session) {
// Verify user has access
if (session.userId != userId) {
logger.warn("Session userId ${session.userId} doesn't match current user ${userId} - access denied")
response.sendError(HttpServletResponse.SC_FORBIDDEN, "Access denied for session: " + sessionId)
return
}
} catch (Exception e) {
logger.error("Error validating session: ${e.message}", e)
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Session validation error")
visit = getCachedVisit(ec, sessionId)
} else {
logger.warn("Session not found: ${sessionId}")
response.sendError(HttpServletResponse.SC_NOT_FOUND, "Session not found: " + sessionId)
return
}
}
// Only create new Visit if we didn't find an existing one
// Create new Visit/session if needed
if (!visit) {
// Initialize web facade for Visit creation, but avoid screen resolution
// Modify request path to avoid ScreenResourceNotFoundException
String originalRequestURI = request.getRequestURI()
String originalPathInfo = request.getPathInfo()
request.setAttribute("jakarta.servlet.include.request_uri", "/mcp")
request.setAttribute("jakarta.servlet.include.path_info", "")
try {
ec.initWebFacade(webappName, request, response)
// Web facade should always create a Visit - if it doesn't, that's a system error
visit = ec.user.getVisit()
if (!visit) {
logger.error("Web facade succeeded but no Visit created - this is a system configuration error")
throw new Exception("Web facade succeeded but no Visit created - check Moqui configuration")
try {
ec.initWebFacade(webappName, request, response)
visit = ec.user.getVisit()
if (!visit) {
throw new Exception("Web facade succeeded but no Visit created")
}
// Create session in adapter with authenticated userId
sessionId = visit.visitId?.toString()
sessionAdapter.createSession(sessionId, ec.user.userId?.toString())
logger.info("Created new session ${sessionId} for user ${ec.user.username}")
} catch (Exception e) {
logger.error("Failed to create session: ${e.message}", e)
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Failed to create session")
return
}
logger.debug("Web facade created Visit ${visit.visitId} for user ${ec.user.username}")
// Store user mapping in memory for fast validation
sessionUsers.put(visit.visitId.toString(), ec.user.userId.toString())
logger.info("Created new Visit ${visit.visitId} for user ${ec.user.username}")
} catch (Exception e) {
logger.error("Web facade initialization failed - this is a system configuration error: ${e.message}", e)
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "System configuration error: Web facade failed to initialize. Check Moqui logs for details.")
return
}
}
// Final check that we have a Visit
if (!visit) {
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Failed to create Visit")
return
}
// Enable async support for SSE
if (request.isAsyncSupported()) {
request.startAsync()
}
// Set SSE headers
response.setContentType("text/event-stream")
response.setCharacterEncoding("UTF-8")
response.setHeader("Cache-Control", "no-cache")
response.setHeader("Connection", "keep-alive")
response.setHeader("Access-Control-Allow-Origin", "*")
response.setHeader("X-Accel-Buffering", "no") // Disable nginx buffering
// Register active connection (transient HTTP connection)
activeConnections.put(visit.visitId, response.writer)
// Create Visit-based session transport (for persistence)
VisitBasedMcpSession session = new VisitBasedMcpSession(visit, response.writer, ec)
response.setHeader("X-Accel-Buffering", "no")
response.setHeader("Mcp-Session-Id", sessionId)
// Register SSE writer with transport
transport.registerSseWriter(sessionId, response.writer)
try {
// Check if this is old HTTP+SSE transport (no session ID, no prior initialization)
// Send endpoint event first for backwards compatibility
// Send endpoint event for backwards compatibility
if (!request.getHeader("Mcp-Session-Id")) {
logger.debug("No Mcp-Session-Id header detected, assuming old HTTP+SSE transport")
sendSseEvent(response.writer, "endpoint", "/mcp", 0)
transport.sendSseEventWithId(response.writer, "endpoint", "/mcp", 0)
}
// Send initial connection event for new transport
def connectData = [
version: "2.0.2",
protocolVersion: "2025-06-18",
architecture: "Visit-based sessions with connection registry"
]
// Set MCP session ID header per specification BEFORE sending any data
response.setHeader("Mcp-Session-Id", visit.visitId.toString())
logger.debug("Set Mcp-Session-Id header to ${visit.visitId} for SSE connection")
sendSseEvent(response.writer, "connect", JsonOutput.toJson(connectData), 1)
// Send connect event
def connectData = [
version: "2.0.2",
protocolVersion: "2025-06-18",
architecture: "Adapter-based MCP with session registry"
]
transport.sendSseEventWithId(response.writer, "connect", JsonOutput.toJson(connectData), 1)
// Deliver any queued notifications
transport.deliverQueuedNotifications(sessionId)
// Keep connection alive with periodic pings
int pingCount = 0
while (!response.isCommitted() && pingCount < 60) { // 5 minutes max
Thread.sleep(5000) // Wait 5 seconds
while (!response.isCommitted() && pingCount < 60) {
Thread.sleep(5000)
if (!response.isCommitted()) {
def pingData = [
type: "ping",
timestamp: System.currentTimeMillis(),
sessionId: visit.visitId,
architecture: "Visit-based sessions"
]
sendSseEvent(response.writer, "ping", JsonOutput.toJson(pingData), pingCount + 2)
if (!transport.sendPing(sessionId)) {
logger.debug("Ping failed for session ${sessionId}, ending SSE loop")
break
}
pingCount++
// Update session activity throttled (every 6th ping = every 30 seconds)
// Update session activity throttled
if (pingCount % 6 == 0) {
updateSessionActivityThrottled(visit.visitId.toString())
updateSessionActivityThrottled(sessionId)
}
}
}
} catch (InterruptedException e) {
logger.info("SSE connection interrupted for session ${visit.visitId}")
logger.info("SSE connection interrupted for session ${sessionId}")
Thread.currentThread().interrupt()
} catch (Exception e) {
logger.warn("Enhanced SSE connection error: ${e.message}", e)
} finally {
// Clean up session - Visit persistence handles cleanup automatically
} finally {
// Clean up
transport.unregisterSseWriter(sessionId)
// Complete async context if available
if (request.isAsyncStarted()) {
try {
def closeData = [
type: "disconnected",
sessionId: visit.visitId,
timestamp: System.currentTimeMillis()
]
sendSseEvent(response.writer, "disconnect", JsonOutput.toJson(closeData), -1)
request.getAsyncContext().complete()
} catch (Exception e) {
// Ignore errors during cleanup
}
// Remove from active connections registry
activeConnections.remove(visit.visitId)
// Complete async context if available
if (request.isAsyncStarted()) {
try {
request.getAsyncContext().complete()
} catch (Exception e) {
logger.debug("Error completing async context: ${e.message}")
logger.debug("Error completing async context: ${e.message}")
}
}
}
}
private void handleMessage(HttpServletRequest request, HttpServletResponse response, ExecutionContextImpl ec, String requestBody)
private void handleMessage(HttpServletRequest request, HttpServletResponse response, ExecutionContextImpl ec, String requestBody)
throws IOException {
String sessionId = request.getHeader("Mcp-Session-Id")
def visit = getCachedVisit(ec, sessionId)
if (!visit) {
def session = sessionAdapter.getSession(sessionId)
if (!session) {
response.sendError(HttpServletResponse.SC_NOT_FOUND, "Session not found: " + sessionId)
return
}
// Verify user has access to this Visit - rely on Moqui security
logger.debug("Session validation: visit.userId=${visit.userId}, ec.user.userId=${ec.user.userId}, ec.user.username=${ec.user.username}")
if (visit.userId && ec.user.userId && visit.userId.toString() != ec.user.userId.toString()) {
logger.warn("Visit userId ${visit.userId} doesn't match current user userId ${ec.user.userId} - access denied")
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
// Verify user has access
if (session.userId != ec.user.userId?.toString()) {
response.setStatus(HttpServletResponse.SC_FORBIDDEN)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
error: "Access denied for session: " + sessionId + " (visit.userId=${visit.userId}, ec.user.userId=${ec.user.userId})",
architecture: "Visit-based sessions"
error: "Access denied for session: " + sessionId
]))
return
}
// Create session wrapper for this Visit
VisitBasedMcpSession session = new VisitBasedMcpSession(visit, response.writer, ec)
try {
if (!requestBody || !requestBody.trim()) {
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
jsonrpc: "2.0",
error: [code: -32602, message: "Empty request body"],
......@@ -438,16 +380,14 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Parse JSON-RPC message
def rpcRequest
try {
rpcRequest = jsonSlurper.parseText(requestBody)
} catch (Exception e) {
logger.error("Failed to parse JSON-RPC message: ${e.message}")
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
jsonrpc: "2.0",
error: [code: -32700, message: "Invalid JSON: " + e.message],
......@@ -455,12 +395,11 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Validate JSON-RPC 2.0 structure
if (!rpcRequest?.jsonrpc || rpcRequest.jsonrpc != "2.0" || !rpcRequest?.method) {
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
jsonrpc: "2.0",
error: [code: -32600, message: "Invalid JSON-RPC 2.0 request"],
......@@ -468,31 +407,25 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Process method with session context
def result = processMcpMethod(rpcRequest.method, rpcRequest.params, ec, sessionId)
// Send response via MCP transport to the specific session
def responseMessage = new JsonRpcResponse(result, rpcRequest.id)
session.sendMessage(responseMessage)
def result = processMcpMethod(rpcRequest.method, rpcRequest.params, ec, sessionId, null)
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
response.setStatus(HttpServletResponse.SC_OK)
// Extract actual result from service response (same as regular handler)
def actualResult = result?.result ?: result
response.writer.write(JsonOutput.toJson([
jsonrpc: "2.0",
id: rpcRequest.id,
result: actualResult
]))
} catch (Exception e) {
logger.error("Error processing message for session ${sessionId}: ${e.message}", e)
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
jsonrpc: "2.0",
error: [code: -32603, message: "Internal error: " + e.message],
......@@ -500,46 +433,14 @@ class EnhancedMcpServlet extends HttpServlet {
]))
}
}
private void handleJsonRpc(HttpServletRequest request, HttpServletResponse response, ExecutionContextImpl ec, String webappName, String requestBody, def visit)
private void handleJsonRpc(HttpServletRequest request, HttpServletResponse response, ExecutionContextImpl ec, String webappName, String requestBody, def visit)
throws IOException {
// Initialize web facade for proper session management
try {
// If we have a visit, use it directly (don't create new one)
visit = ec.user.getVisit()
if (visit) {
request.getSession().setAttribute("moqui.visitId", visit.visitId)
logger.debug("JSON-RPC web facade initialized for user: ${ec.user?.username} with visit: ${visit.visitId}")
} else {
// No visit exists, need to create one
logger.info("Creating new Visit record for user: ${ec.user?.username}")
visit = ec.entity.makeValue("moqui.server.Visit")
visit.visitId = ec.userFacade.getVisitId(visit)
visit.userId = ec.user.userId
visit.sessionId = visit.sessionId
visit.userAccountId = ec.user.userAccount?.userAccountId
visit.sessionCreatedDate = ec.user.nowTimestamp
visit.visitStatus = null
visit.lastActiveDate = ec.user.nowTimestamp
visit.visitDeletedDate = null
ec.entity.create(visit)
logger.info("Visit ${visit.visitId} created for user: ${ec.user?.username}")
}
ec.initWebFacade(webappName, request, response)
logger.debug("JSON-RPC web facade initialized for user: ${ec.user?.username} with visit: ${visit.visitId}")
} catch (Exception e) {
logger.warn("Web facade initialization warning: ${e.message}")
// Continue anyway - we may still have basic user context from auth
}
String method = request.getMethod()
String acceptHeader = request.getHeader("Accept")
logger.debug("Enhanced MCP JSON-RPC Request: ${method} ${request.requestURI} - Accept: ${acceptHeader}")
// Validate Accept header per MCP 2025-11-25 spec requirement #2
// Client MUST include Accept header with at least one of: application/json or text/event-stream
// Validate Accept header per MCP spec
if (!acceptHeader || !(acceptHeader.contains("application/json") || acceptHeader.contains("text/event-stream"))) {
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
......@@ -550,7 +451,7 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
if (!"POST".equals(method)) {
response.setStatus(HttpServletResponse.SC_METHOD_NOT_ALLOWED)
response.setContentType("application/json")
......@@ -562,9 +463,6 @@ class EnhancedMcpServlet extends HttpServlet {
return
}
// Use pre-read request body
logger.debug("Using pre-read request body, length: ${requestBody?.length()}")
if (!requestBody) {
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
......@@ -576,16 +474,10 @@ class EnhancedMcpServlet extends HttpServlet {
return
}
// Log request body for debugging (be careful with this in production)
if (requestBody.length() > 0) {
logger.trace("MCP JSON-RPC request body: ${requestBody}")
}
def rpcRequest
try {
rpcRequest = jsonSlurper.parseText(requestBody)
} catch (Exception e) {
logger.error("Failed to parse JSON-RPC request: ${e.message}")
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
......@@ -595,7 +487,7 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Validate JSON-RPC 2.0 structure
if (!rpcRequest?.jsonrpc || rpcRequest.jsonrpc != "2.0" || !rpcRequest?.method) {
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
......@@ -607,10 +499,9 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Validate MCP protocol version per specification
// Validate MCP protocol version
String protocolVersion = request.getHeader("MCP-Protocol-Version")
// Support multiple protocol versions with version negotiation
def supportedVersions = ["2025-06-18", "2025-11-25", "2024-11-05", "2024-10-07", "2023-06-05"]
if (protocolVersion && !supportedVersions.contains(protocolVersion)) {
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
......@@ -623,18 +514,15 @@ class EnhancedMcpServlet extends HttpServlet {
return
}
// Get session ID from Mcp-Session-Id header per MCP specification
// Get session ID from header
String sessionId = request.getHeader("Mcp-Session-Id")
logger.debug("Session ID from header: '${sessionId}', method: '${rpcRequest.method}'")
// For initialize and notifications/initialized methods, use visit ID as session ID if no header
// For initialize, use visit ID as session ID
if (!sessionId && ("initialize".equals(rpcRequest.method) || "notifications/initialized".equals(rpcRequest.method)) && visit) {
sessionId = visit.visitId
logger.debug("${rpcRequest.method} method: using visit ID as session ID: ${sessionId}")
sessionId = visit.visitId?.toString()
}
// Validate session ID for non-initialize requests per MCP spec
// Allow notifications/initialized without session ID as it completes the initialization process
// Validate session ID for non-initialize requests
if (!sessionId && rpcRequest.method != "initialize" && rpcRequest.method != "notifications/initialized") {
response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
response.setContentType("application/json")
......@@ -645,16 +533,13 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// For existing sessions, set visit ID in HTTP session before web facade initialization
// This ensures Moqui picks up the existing Visit when initWebFacade() is called
// For existing sessions, validate ownership
if (sessionId && rpcRequest.method != "initialize") {
try {
ec.artifactExecution.disableAuthz()
def existingVisit = ec.entity.find("moqui.server.Visit")
.condition("visitId", sessionId)
.one()
def session = sessionAdapter.getSession(sessionId)
if (!session) {
// Try loading from database
def existingVisit = getCachedVisit(ec, sessionId)
if (!existingVisit) {
response.setStatus(HttpServletResponse.SC_NOT_FOUND)
response.setContentType("application/json")
......@@ -665,9 +550,9 @@ class EnhancedMcpServlet extends HttpServlet {
]))
return
}
// Rely on Moqui security - only allow access if visit and current user match
if (!existingVisit.userId || !ec.user.userId || existingVisit.userId.toString() != ec.user.userId.toString()) {
// Verify ownership
if (existingVisit.userId?.toString() != ec.user.userId?.toString()) {
response.setStatus(HttpServletResponse.SC_FORBIDDEN)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
......@@ -678,317 +563,217 @@ class EnhancedMcpServlet extends HttpServlet {
return
}
// Set visit ID in HTTP session so Moqui web facade initialization picks it up
request.session.setAttribute("moqui.visitId", sessionId)
logger.debug("Set existing Visit ${sessionId} in HTTP session for user ${ec.user.username}")
} catch (Exception e) {
logger.error("Error finding session ${sessionId}: ${e.message}")
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR)
// Create session in adapter if not exists
if (!sessionAdapter.hasSession(sessionId)) {
sessionAdapter.createSession(sessionId, ec.user.userId?.toString())
}
} else if (session.userId != ec.user.userId?.toString()) {
response.setStatus(HttpServletResponse.SC_FORBIDDEN)
response.setContentType("application/json")
response.writer.write(JsonOutput.toJson([
jsonrpc: "2.0",
error: [code: -32603, message: "Session lookup error: ${e.message}"],
error: [code: -32600, message: "Access denied for session: ${sessionId}"],
id: rpcRequest.id
]))
return
} finally {
ec.artifactExecution.enableAuthz()
}
}
// Check if this is a notification (no id) - notifications get empty response
// Check if this is a notification (no id)
boolean isNotification = !rpcRequest.containsKey('id')
if (isNotification) {
// Special handling for notifications/initialized to transition session state
if ("notifications/initialized".equals(rpcRequest.method)) {
logger.debug("Processing notifications/initialized for sessionId: ${sessionId}")
if (sessionId) {
sessionStates.put(sessionId, STATE_INITIALIZED)
// Store user mapping in memory for fast validation
sessionUsers.put(sessionId, ec.user.userId.toString())
logger.debug("Session ${sessionId} transitioned to INITIALIZED state for user ${ec.user.userId}")
sessionAdapter.setSessionState(sessionId, McpSession.STATE_INITIALIZED)
logger.debug("Session ${sessionId} transitioned to INITIALIZED state")
}
// For notifications/initialized, return 202 Accepted per MCP HTTP Streaming spec
if (sessionId) {
response.setHeader("Mcp-Session-Id", sessionId.toString())
}
response.setContentType("text/event-stream")
response.setStatus(HttpServletResponse.SC_ACCEPTED) // 202 Accepted
logger.debug("Sent 202 Accepted response for notifications/initialized")
response.flushBuffer() // Commit the response immediately
return
response.setHeader("Mcp-Session-Id", sessionId)
}
response.setContentType("text/event-stream")
response.setStatus(HttpServletResponse.SC_ACCEPTED)
response.flushBuffer()
return
}
// For other notifications, set session header if needed but NO response per MCP spec
// Other notifications receive 204 No Content
if (sessionId) {
response.setHeader("Mcp-Session-Id", sessionId.toString())
response.setHeader("Mcp-Session-Id", sessionId)
}
// Other notifications receive NO response per MCP specification
response.setStatus(HttpServletResponse.SC_NO_CONTENT) // 204 No Content
response.flushBuffer() // Commit the response immediately
response.setStatus(HttpServletResponse.SC_NO_CONTENT)
response.flushBuffer()
return
}
// Process MCP method using Moqui services with session ID if available
def result = processMcpMethod(rpcRequest.method, rpcRequest.params, ec, sessionId, visit ?: [:])
// Update session activity throttled for actual user actions (not pings or tools/list)
// tools/list is read-only discovery and shouldn't update session activity to prevent lock contention
// Process MCP method
def result = processMcpMethod(rpcRequest.method, rpcRequest.params, ec, sessionId, visit)
// Update session activity
if (sessionId && !"ping".equals(rpcRequest.method) && !"tools/list".equals(rpcRequest.method)) {
updateSessionActivityThrottled(sessionId)
}
// Set Mcp-Session-Id header BEFORE any response data (per MCP 2025-06-18 spec)
// For initialize method, always use sessionId we have (from visit or header)
// Set session header
String responseSessionId = null
if (rpcRequest.method == "initialize" && sessionId) {
responseSessionId = sessionId.toString()
responseSessionId = sessionId
} else if (result?.sessionId) {
responseSessionId = result.sessionId.toString()
responseSessionId = result.sessionId?.toString()
} else if (sessionId) {
// For other methods, ensure we always return session ID from header
responseSessionId = sessionId.toString()
responseSessionId = sessionId
}
if (responseSessionId) {
response.setHeader("Mcp-Session-Id", responseSessionId)
logger.debug("Set Mcp-Session-Id header to ${responseSessionId} for method ${rpcRequest.method}")
}
// Build JSON-RPC response for regular requests
// Extract the actual result from Moqui service response
// Build response
def actualResult = result?.result ?: result
def rpcResponse = [
jsonrpc: "2.0",
id: rpcRequest.id,
result: actualResult
]
// Standard MCP flow: include notifications in response content array
if (sessionId && notificationQueues.containsKey(sessionId)) {
def pendingNotifications = notificationQueues.get(sessionId)
if (pendingNotifications && !pendingNotifications.isEmpty()) {
logger.debug("Adding ${pendingNotifications.size()} pending notifications to response content for session ${sessionId}")
// Convert notifications to content items and add to result
def notificationContent = []
for (notification in pendingNotifications) {
notificationContent << [
type: "text",
text: "Notification [${notification.method}]: " + JsonOutput.toJson(notification.params ?: notification)
]
}
// Merge notification content with existing result content
def existingContent = actualResult?.content ?: []
actualResult.content = existingContent + notificationContent
// Clear delivered notifications
notificationQueues.put(sessionId, [])
logger.debug("Merged ${pendingNotifications.size()} notifications into response for session ${sessionId}")
}
}
response.setContentType("application/json")
response.setCharacterEncoding("UTF-8")
// Send the main response
response.writer.write(JsonOutput.toJson(rpcResponse))
}
private Map<String, Object> processMcpMethod(String method, Map params, ExecutionContextImpl ec, String sessionId, def visit) {
logger.debug("Enhanced METHOD: ${method} with sessionId: ${sessionId}")
logger.debug("Processing MCP method: ${method} with sessionId: ${sessionId}")
try {
// Ensure params is not null
if (params == null) {
params = [:]
}
// Add session context to parameters for services
params.sessionId = visit?.visitId
if (params == null) params = [:]
params.sessionId = visit?.visitId ?: sessionId
// Check session state for methods that require initialization
// Use the sessionId from header for consistency (this is what the client tracks)
Integer sessionState = sessionId ? sessionStates.get(sessionId) : null
// Methods that don't require initialized session
def session = sessionId ? sessionAdapter.getSession(sessionId) : null
if (!["initialize", "ping"].contains(method)) {
if (sessionState != STATE_INITIALIZED) {
logger.warn("Method ${method} called but session ${sessionId} not initialized (state: ${sessionState})")
if (!session || session.state != McpSession.STATE_INITIALIZED) {
logger.warn("Method ${method} called but session ${sessionId} not initialized")
return [error: "Session not initialized. Call initialize first, then send notifications/initialized."]
}
}
switch (method) {
case "initialize":
// For initialize, use the visitId we just created instead of null sessionId from request
if (visit && visit.visitId) {
params.sessionId = visit.visitId
// Set session to initializing state using actual sessionId as key (for consistency)
sessionStates.put(params.sessionId, STATE_INITIALIZING)
logger.debug("Initialize - using visitId: ${visit.visitId}, set state ${params.sessionId} to INITIALIZING")
} else {
logger.warn("Initialize - no visit available, using null sessionId")
// Create session in adapter with actual authenticated userId
if (!sessionAdapter.hasSession(params.sessionId?.toString())) {
sessionAdapter.createSession(params.sessionId?.toString(), ec.user.userId?.toString())
}
sessionAdapter.setSessionState(params.sessionId?.toString(), McpSession.STATE_INITIALIZING)
}
params.actualUserId = ec.user.userId
logger.debug("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
// Initialize successful - transition session to INITIALIZED state
sessionStates.put(params.sessionId, STATE_INITIALIZED)
logger.debug("Initialize - successful, set state ${params.sessionId} to INITIALIZED")
if (serviceResult && !serviceResult.error) {
serviceResult.sessionId = params.sessionId
sessionAdapter.setSessionState(params.sessionId?.toString(), McpSession.STATE_INITIALIZED)
}
return serviceResult
case "ping":
// Simple ping for testing - bypass service for now
return [pong: System.currentTimeMillis(), sessionId: visit?.visitId, user: ec.user.username]
case "tools/list":
// Ensure sessionId is available to service for notification consistency
if (sessionId) params.sessionId = sessionId
return callMcpService("list#Tools", params, ec)
case "tools/call":
// Ensure sessionId is available to service for notification consistency
if (sessionId) params.sessionId = sessionId
return callMcpService("mcp#ToolsCall", params, ec)
case "resources/list":
return callMcpService("mcp#ResourcesList", params, ec)
case "resources/read":
return callMcpService("mcp#ResourcesRead", params, ec)
case "resources/templates/list":
return callMcpService("mcp#ResourcesTemplatesList", params, ec)
case "resources/subscribe":
return callMcpService("mcp#ResourcesSubscribe", params, ec)
case "resources/unsubscribe":
return callMcpService("mcp#ResourcesUnsubscribe", params, ec)
case "prompts/list":
return callMcpService("mcp#PromptsList", params, ec)
case "prompts/get":
return callMcpService("mcp#PromptsGet", params, ec)
case "roots/list":
return callMcpService("mcp#RootsList", params, ec)
case "sampling/createMessage":
return callMcpService("mcp#SamplingCreateMessage", params, ec)
case "elicitation/create":
return callMcpService("mcp#ElicitationCreate", params, ec)
// NOTE: notifications/initialized is handled as a notification, not a request method
// It will be processed by the notification handling logic above (lines 824-837)
case "notifications/tools/list_changed":
// Handle tools list changed notification
logger.debug("Tools list changed for sessionId: ${sessionId}")
// Could trigger cache invalidation here if needed
return null
case "notifications/resources/list_changed":
// Handle resources list changed notification
logger.debug("Resources list changed for sessionId: ${sessionId}")
// Could trigger cache invalidation here if needed
case "notifications/prompts/list_changed":
case "notifications/roots/list_changed":
case "logging/setLevel":
logger.debug("Notification ${method} for sessionId: ${sessionId}")
return null
case "notifications/send":
// Handle notification sending
def notificationMethod = params?.method
def notificationParams = params?.params
if (!notificationMethod) {
throw new IllegalArgumentException("method is required for sending notification")
}
logger.debug("Sending notification ${notificationMethod} for sessionId: ${sessionId}")
// Queue notification for delivery through SSE or polling
if (sessionId) {
def notification = [
jsonrpc: "2.0",
method: notificationMethod,
params: notificationParams,
timestamp: System.currentTimeMillis()
params: notificationParams
]
// Add to notification queue
def queue = notificationQueues.get(sessionId) ?: []
queue << notification
notificationQueues.put(sessionId, queue)
logger.debug("Notification queued for session ${sessionId}: ${notificationMethod}")
transport.sendNotification(sessionId, notification)
}
return [sent: true, sessionId: sessionId, method: notificationMethod]
case "notifications/subscribe":
// Handle notification subscription
def subscriptionMethod = params?.method
if (!sessionId || !subscriptionMethod) {
throw new IllegalArgumentException("sessionId and method are required for subscription")
}
def subscriptions = sessionSubscriptions.get(sessionId) ?: new HashSet<>()
subscriptions.add(subscriptionMethod)
sessionSubscriptions.put(sessionId, subscriptions)
logger.debug("Session ${sessionId} subscribed to: ${subscriptionMethod}")
session?.subscriptions?.add(subscriptionMethod)
return [subscribed: true, sessionId: sessionId, method: subscriptionMethod]
case "notifications/unsubscribe":
// Handle notification unsubscription
def subscriptionMethod = params?.method
if (!sessionId || !subscriptionMethod) {
throw new IllegalArgumentException("sessionId and method are required for unsubscription")
}
def subscriptions = sessionSubscriptions.get(sessionId)
if (subscriptions) {
subscriptions.remove(subscriptionMethod)
if (subscriptions.isEmpty()) {
sessionSubscriptions.remove(sessionId)
} else {
sessionSubscriptions.put(sessionId, subscriptions)
}
logger.debug("Session ${sessionId} unsubscribed from: ${subscriptionMethod}")
}
session?.subscriptions?.remove(subscriptionMethod)
return [unsubscribed: true, sessionId: sessionId, method: subscriptionMethod]
case "notifications/progress":
// Handle progress notification
def progressToken = params?.progressToken
def progressValue = params?.progress
def total = params?.total
logger.debug("Progress notification for sessionId: ${sessionId}, token: ${progressToken}, progress: ${progressValue}/${total}")
// Store progress for potential polling
if (sessionId && progressToken) {
def progressKey = "${sessionId}_${progressToken}"
sessionProgress.put(progressKey, [progress: progressValue, total: total, timestamp: System.currentTimeMillis()])
}
logger.debug("Progress notification: ${progressToken}, ${progressValue}/${total}")
return null
case "notifications/resources/updated":
// Handle resource updated notification
def uri = params?.uri
logger.debug("Resource updated notification for sessionId: ${sessionId}, uri: ${uri}")
// Could trigger resource cache invalidation here
return null
case "notifications/prompts/list_changed":
// Handle prompts list changed notification
logger.debug("Prompts list changed for sessionId: ${sessionId}")
// Could trigger prompt cache invalidation here
logger.debug("Resource updated: ${params?.uri}")
return null
case "notifications/message":
// Handle general message notification
def level = params?.level ?: "info"
def message = params?.message
def data = params?.data
logger.debug("Message notification for sessionId: ${sessionId}, level: ${level}, message: ${message}")
// Store message for potential retrieval
if (sessionId) {
def messages = sessionMessages.get(sessionId) ?: []
messages << [level: level, message: message, data: data, timestamp: System.currentTimeMillis()]
sessionMessages.put(sessionId, messages)
}
return null
case "notifications/roots/list_changed":
// Handle roots list changed notification
logger.debug("Roots list changed for sessionId: ${sessionId}")
// Could trigger roots cache invalidation here
return null
case "logging/setLevel":
// Handle logging level change notification
logger.debug("Logging level change requested for sessionId: ${sessionId}")
logger.debug("Message notification: level=${level}, message=${message}")
return null
default:
throw new IllegalArgumentException("Method not found: ${method}")
}
......@@ -997,116 +782,41 @@ class EnhancedMcpServlet extends HttpServlet {
throw e
}
}
private Map<String, Object> callMcpService(String serviceName, Map params, ExecutionContextImpl ec) {
logger.debug("Enhanced Calling MCP service: ${serviceName} with params: ${params}")
logger.debug("Calling MCP service: ${serviceName}")
try {
ec.artifactExecution.disableAuthz()
def result = ec.service.sync().name("McpServices.${serviceName}")
.parameters(params ?: [:])
.call()
logger.debug("Enhanced MCP service ${serviceName} result: ${result?.result?.size() ? 'result with ' + (result.result?.tools?.size() ?: 0) + ' tools' : 'empty result'}")
if (result == null) {
logger.error("Enhanced MCP service ${serviceName} returned null result")
return [error: "Service returned null result"]
}
// Service framework returns result in 'result' field when out-parameters are used
// Extract the inner result to avoid double nesting in JSON-RPC response
// The MCP services already set the correct 'result' structure
// Some services return result directly, others nest it in result.result
if (result?.containsKey('result')) {
return result.result
} else {
return result ?: [error: "Service returned null result"]
}
return result
} catch (Exception e) {
logger.error("Error calling Enhanced MCP service ${serviceName}", e)
logger.error("Error calling MCP service ${serviceName}", e)
return [error: e.message]
} finally {
ec.artifactExecution.enableAuthz()
}
}
private void sendSseEvent(PrintWriter writer, String eventType, String data, long eventId = -1) throws IOException {
try {
if (eventId >= 0) {
writer.write("id: " + eventId + "\n")
}
writer.write("event: " + eventType + "\n")
writer.write("data: " + data + "\n\n")
writer.flush()
if (writer.checkError()) {
throw new IOException("Client disconnected")
}
} catch (Exception e) {
throw new IOException("Failed to send SSE event: " + e.message, e)
}
}
// CORS handling based on MoquiServlet pattern
private static boolean handleCors(HttpServletRequest request, HttpServletResponse response, String webappName, ExecutionContextFactoryImpl ecfi) {
String originHeader = request.getHeader("Origin")
if (originHeader) {
response.setHeader("Access-Control-Allow-Origin", originHeader)
response.setHeader("Access-Control-Allow-Credentials", "true")
}
String methodHeader = request.getHeader("Access-Control-Request-Method")
if (methodHeader) {
response.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
response.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization, Mcp-Session-Id, MCP-Protocol-Version, Accept")
response.setHeader("Access-Control-Max-Age", "3600")
return true
}
return false
}
/**
* Queue a server notification for delivery to client
*/
void queueNotification(String sessionId, Map notification) {
if (!sessionId || !notification) return
def queue = notificationQueues.computeIfAbsent(sessionId) { [] }
queue << notification
logger.info("Queued notification for session ${sessionId}: ${notification}")
// Session activity updates handled at JSON-RPC level, not notification level
// This prevents excessive database updates during notification processing
// Also try to send via SSE if active connection exists
def writer = activeConnections.get(sessionId)
if (writer && !writer.checkError()) {
try {
// Send as proper JSON-RPC notification via SSE
def notificationMessage = [
jsonrpc: "2.0",
method: notification.method ?: "notifications/message",
params: notification.params ?: notification
]
sendSseEvent(writer, "message", JsonOutput.toJson(notificationMessage), System.currentTimeMillis())
logger.debug("Sent notification via SSE to session ${sessionId}")
} catch (Exception e) {
logger.warn("Failed to send notification via SSE to session ${sessionId}: ${e.message}")
}
}
}
/**
* Get Visit from cache to reduce database access and prevent lock contention
*/
private EntityValue getCachedVisit(ExecutionContext ec, String sessionId) {
private EntityValue getCachedVisit(ExecutionContextImpl ec, String sessionId) {
if (!sessionId) return null
EntityValue cachedVisit = visitCache.get(sessionId)
if (cachedVisit != null) {
return cachedVisit
}
// Not in cache, load from database with authz disabled
try {
ec.artifactExecution.disableAuthz()
EntityValue visit = ec.entity.find("moqui.server.Visit")
......@@ -1120,165 +830,101 @@ class EnhancedMcpServlet extends HttpServlet {
ec.artifactExecution.enableAuthz()
}
}
/**
* Throttled session activity update to prevent database lock contention
* Uses synchronized per-session to prevent concurrent updates
*/
private void updateSessionActivityThrottled(String sessionId) {
if (!sessionId) return
long now = System.currentTimeMillis()
Long lastUpdate = lastActivityUpdate.get(sessionId)
// Only update if 30 seconds have passed since last update
if (lastUpdate == null || (now - lastUpdate) > ACTIVITY_UPDATE_INTERVAL_MS) {
// Use session-specific lock to avoid sessionId.intern() deadlocks
Object sessionLock = sessionLocks.computeIfAbsent(sessionId, { new Object() })
Object sessionLock = sessionAdapter.getSessionLock(sessionId)
synchronized (sessionLock) {
// Double-check after acquiring lock
lastUpdate = lastActivityUpdate.get(sessionId)
if (lastUpdate == null || (now - lastUpdate) > ACTIVITY_UPDATE_INTERVAL_MS) {
try {
// Look up Visit and update activity
ExecutionContextFactoryImpl ecfi = (ExecutionContextFactoryImpl) getServletContext().getAttribute("executionContextFactory")
if (ecfi) {
def ec = ecfi.getEci()
try {
def visit = getCachedVisit(ec, sessionId)
if (visit) {
visit.thruDate = ec.user.getNowTimestamp()
//visit.update()
// Update cache with new thruDate
visitCache.put(sessionId, visit)
lastActivityUpdate.put(sessionId, now)
logger.debug("Updated activity for session ${sessionId} (throttled, synchronized)")
}
} finally {
ec.destroy()
}
}
} catch (Exception e) {
logger.warn("Failed to update session activity for ${sessionId}: ${e.message}")
}
sessionAdapter.touchSession(sessionId)
lastActivityUpdate.put(sessionId, now)
logger.debug("Updated activity for session ${sessionId}")
}
}
}
}
@Override
void destroy() {
logger.info("Destroying EnhancedMcpServlet")
// Close all active connections
activeConnections.values().each { writer ->
try {
writer.write("event: shutdown\ndata: {\"type\":\"shutdown\",\"timestamp\":\"${System.currentTimeMillis()}\"}\n\n")
writer.flush()
} catch (Exception e) {
logger.debug("Error sending shutdown to connection: ${e.message}")
}
private static boolean handleCors(HttpServletRequest request, HttpServletResponse response) {
String originHeader = request.getHeader("Origin")
if (originHeader) {
response.setHeader("Access-Control-Allow-Origin", originHeader)
response.setHeader("Access-Control-Allow-Credentials", "true")
}
activeConnections.clear()
super.destroy()
String methodHeader = request.getHeader("Access-Control-Request-Method")
if (methodHeader) {
response.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
response.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization, Mcp-Session-Id, MCP-Protocol-Version, Accept")
response.setHeader("Access-Control-Max-Age", "3600")
return true
}
return false
}
/**
* Broadcast message to all active MCP sessions
* Queue a notification for delivery to a session
*/
void broadcastToAllSessions(JsonRpcMessage message) {
try {
ec.artifactExecution.disableAuthz()
// Look up all MCP Visits (persistent)
def mcpVisits = ec.entity.find("moqui.server.Visit")
.condition("initialRequest", "like", "%mcpSession%")
.list()
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, "message", 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)
} finally {
ec.artifactExecution.enableAuthz()
}
void queueNotification(String sessionId, Map notification) {
if (!sessionId || !notification) return
transport.sendNotification(sessionId, notification)
}
/**
* Send SSE event to specific session (helper method)
* Send to a specific session
*/
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)
activeConnections.remove(sessionId)
visitCache.remove(sessionId)
sessionUsers.remove(sessionId)
}
void sendToSession(String sessionId, Map message) {
transport.sendMessage(sessionId, message)
}
/**
* Get session statistics for monitoring
* Get session statistics
*/
Map getSessionStatistics() {
try {
// Look up all MCP Visits (persistent)
def mcpVisits = ec.entity.find("moqui.server.Visit")
.condition("initialRequest", "like", "%mcpSession%")
.disableAuthz()
.list()
return [
totalMcpVisits: mcpVisits.size(),
activeConnections: activeConnections.size(),
maxConnections: maxConnections,
architecture: "Visit-based sessions with connection registry",
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 [
activeConnections: activeConnections.size(),
maxConnections: maxConnections,
error: e.message
]
def stats = transport.getStatistics()
return stats + [
maxConnections: maxConnections,
endpoints: [
sse: sseEndpoint,
message: messageEndpoint
],
keepAliveInterval: keepAliveIntervalSeconds
]
}
/**
* Get the notification bridge for external access
*/
MoquiNotificationMcpBridge getNotificationBridge() {
return notificationBridge
}
/**
* Get the transport for external access
*/
SseTransport getTransport() {
return transport
}
@Override
void destroy() {
logger.info("Destroying EnhancedMcpServlet")
// Close all sessions
for (String sessionId in sessionAdapter.getAllSessionIds()) {
transport.closeSession(sessionId)
}
// Clean up notification bridge
if (notificationBridge) {
notificationBridge.destroy()
}
super.destroy()
}
}
......
/*
* 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
/**
* Simple transport interface for MCP messages
*/
interface MoquiMcpTransport {
void sendMessage(JsonRpcMessage message)
boolean isActive()
String getSessionId()
}
\ 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.moqui.context.ExecutionContext
import org.moqui.impl.context.ExecutionContextImpl
import org.moqui.entity.EntityValue
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong
/**
* MCP Session implementation that uses Moqui's Visit entity directly
* Eliminates custom session management by leveraging Moqui's built-in Visit system
*/
class VisitBasedMcpSession implements MoquiMcpTransport {
protected final static Logger logger = LoggerFactory.getLogger(VisitBasedMcpSession.class)
private final EntityValue visit // The Visit entity record
private final PrintWriter writer
private final ExecutionContextImpl ec
private final AtomicBoolean active = new AtomicBoolean(true)
private final AtomicBoolean closing = new AtomicBoolean(false)
private final AtomicLong messageCount = new AtomicLong(0)
VisitBasedMcpSession(EntityValue visit, PrintWriter writer, ExecutionContextImpl ec) {
this.visit = visit
this.writer = writer
this.ec = ec
// Initialize MCP session in Visit if not already done
initializeMcpSession()
}
private void initializeMcpSession() {
try {
def metadata = getSessionMetadata()
if (!metadata.mcpSession) {
// Mark this Visit as an MCP session
metadata.mcpSession = true
metadata.mcpProtocolVersion = "2025-11-25"
metadata.mcpCreatedAt = System.currentTimeMillis()
metadata.mcpTransportType = "SSE"
metadata.mcpMessageCount = 0
saveSessionMetadata(metadata)
logger.debug("MCP Session initialized for Visit ${visit.visitId}")
}
} catch (Exception e) {
logger.warn("Failed to initialize MCP session for Visit ${visit.visitId}: ${e.message}")
}
}
@Override
void sendMessage(JsonRpcMessage message) {
if (!active.get() || closing.get()) {
logger.warn("Attempted to send message on inactive or closing session ${visit.visitId}")
return
}
try {
String jsonMessage = message.toJson()
sendSseEvent("message", jsonMessage)
messageCount.incrementAndGet()
// Session activity now managed at servlet level to avoid lock contention
} catch (Exception e) {
logger.error("Failed to send message on session ${visit.visitId}: ${e.message}")
if (e.message?.contains("disconnected") || e.message?.contains("Client disconnected")) {
close()
}
}
}
void closeGracefully() {
if (!active.compareAndSet(true, false)) {
return // Already closed
}
closing.set(true)
logger.debug("Gracefully closing MCP session ${visit.visitId}")
try {
// Send graceful shutdown notification
def shutdownMessage = new JsonRpcNotification("shutdown", [
sessionId: visit.visitId,
timestamp: System.currentTimeMillis()
])
sendMessage(shutdownMessage)
// Give some time for message to be sent
Thread.sleep(100)
} catch (Exception e) {
logger.warn("Error during graceful shutdown of session ${visit.visitId}: ${e.message}")
} finally {
close()
}
}
void close() {
if (!active.compareAndSet(true, false)) {
return // Already closed
}
logger.debug("Closing MCP session ${visit.visitId} (messages sent: ${messageCount.get()})")
try {
// Send final close event if writer is still available
if (writer && !writer.checkError()) {
sendSseEvent("close", groovy.json.JsonOutput.toJson([
type: "disconnected",
sessionId: visit.visitId,
messageCount: messageCount.get(),
timestamp: System.currentTimeMillis()
]))
}
} catch (Exception e) {
logger.warn("Error during session close ${visit.visitId}: ${e.message}")
}
}
@Override
boolean isActive() {
return active.get() && !closing.get() && writer && !writer.checkError()
}
@Override
String getSessionId() {
return visit.visitId
}
String getVisitId() {
return visit.visitId
}
EntityValue getVisit() {
return visit
}
/**
* Get session statistics
*/
Map getSessionStats() {
return [
sessionId: visit.visitId,
visitId: visit.visitId,
createdAt: visit.fromDate,
messageCount: messageCount.get(),
active: active.get(),
closing: closing.get(),
duration: System.currentTimeMillis() - visit.fromDate.time
]
}
/**
* Send SSE event with proper formatting
*/
private void sendSseEvent(String eventType, String data) throws IOException {
if (!writer || writer.checkError()) {
throw new IOException("Writer is closed or client disconnected")
}
writer.write("event: " + eventType + "\n")
writer.write("data: " + data + "\n\n")
writer.flush()
if (writer.checkError()) {
throw new IOException("Client disconnected during write")
}
}
// Session activity management moved to servlet level to avoid database lock contention
// This method is no longer called - servlet manages session updates throttled
// Session end management moved to servlet level to avoid database lock contention
// Servlet will handle Visit updates when connections close
/**
* Get session metadata from Visit's initialRequest field
*/
Map getSessionMetadata() {
try {
def metadataJson = visit.initialRequest
if (metadataJson) {
return groovy.json.JsonSlurper().parseText(metadataJson) as Map
}
} catch (Exception e) {
logger.debug("Failed to parse session metadata: ${e.message}")
}
return [:]
}
/**
* Add custom metadata to session
*/
void addSessionMetadata(String key, Object value) {
def metadata = getSessionMetadata()
metadata[key] = value
saveSessionMetadata(metadata)
}
/**
* Save session metadata to Visit's initialRequest field
*/
private void saveSessionMetadata(Map metadata) {
// Session metadata stored in memory only - no Visit updates to prevent lock contention
try {
sessionMetadata.putAll(metadata)
} catch (Exception e) {
logger.debug("Failed to save session metadata: ${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.adapter
import org.moqui.entity.EntityValue
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.concurrent.ConcurrentHashMap
/**
* Adapter that maps Moqui Visit sessions to MCP sessions.
* Provides in-memory session tracking to avoid database lock contention.
*/
class McpSessionAdapter {
protected final static Logger logger = LoggerFactory.getLogger(McpSessionAdapter.class)
// Visit ID → MCP Session state
private final Map<String, McpSession> sessions = new ConcurrentHashMap<>()
// User ID → Set of Visit IDs (for user-targeted notifications)
private final Map<String, Set<String>> userSessions = new ConcurrentHashMap<>()
// Session-specific locks to avoid sessionId.intern() deadlocks
private final Map<String, Object> sessionLocks = new ConcurrentHashMap<>()
/**
* Create a new MCP session from a Moqui Visit
* @param visit The Moqui Visit entity
* @return The created McpSession
*/
McpSession createSession(EntityValue visit) {
String visitId = visit.visitId?.toString()
String userId = visit.userId?.toString()
if (!visitId) {
throw new IllegalArgumentException("Visit must have a visitId")
}
def session = new McpSession(
visitId: visitId,
userId: userId,
state: McpSession.STATE_INITIALIZED
)
sessions.put(visitId, session)
// Track user → sessions mapping
if (userId) {
def userSet = userSessions.computeIfAbsent(userId) { new ConcurrentHashMap<>().newKeySet() }
userSet.add(visitId)
}
logger.debug("Created MCP session ${visitId} for user ${userId}")
return session
}
/**
* Create a new MCP session with explicit parameters
* @param visitId The Visit/session ID
* @param userId The user ID
* @return The created McpSession
*/
McpSession createSession(String visitId, String userId) {
if (!visitId) {
throw new IllegalArgumentException("visitId is required")
}
def session = new McpSession(
visitId: visitId,
userId: userId,
state: McpSession.STATE_INITIALIZED
)
sessions.put(visitId, session)
// Track user → sessions mapping
if (userId) {
def userSet = userSessions.computeIfAbsent(userId) { new ConcurrentHashMap<>().newKeySet() }
userSet.add(visitId)
}
logger.debug("Created MCP session ${visitId} for user ${userId}")
return session
}
/**
* Close and remove a session
* @param visitId The session/visit ID to close
*/
void closeSession(String visitId) {
def session = sessions.remove(visitId)
if (session) {
// Remove from user tracking
if (session.userId) {
def userSet = userSessions.get(session.userId)
if (userSet) {
userSet.remove(visitId)
if (userSet.isEmpty()) {
userSessions.remove(session.userId)
}
}
}
// Clean up session lock
sessionLocks.remove(visitId)
logger.debug("Closed MCP session ${visitId}")
}
}
/**
* Get a session by visit ID
* @param visitId The session/visit ID
* @return The McpSession or null if not found
*/
McpSession getSession(String visitId) {
return sessions.get(visitId)
}
/**
* Check if a session exists and is active
* @param visitId The session/visit ID
* @return true if the session exists
*/
boolean hasSession(String visitId) {
return sessions.containsKey(visitId)
}
/**
* Get all session IDs for a specific user
* @param userId The user ID
* @return Set of session/visit IDs (empty set if none)
*/
Set<String> getSessionsForUser(String userId) {
return userSessions.get(userId) ?: Collections.emptySet()
}
/**
* Get all active session IDs
* @return Set of all session IDs
*/
Set<String> getAllSessionIds() {
return sessions.keySet()
}
/**
* Get the count of active sessions
* @return Number of active sessions
*/
int getSessionCount() {
return sessions.size()
}
/**
* Get a session-specific lock for synchronized operations
* @param visitId The session/visit ID
* @return The lock object
*/
Object getSessionLock(String visitId) {
return sessionLocks.computeIfAbsent(visitId) { new Object() }
}
/**
* Update session state
* @param visitId The session/visit ID
* @param state The new state
*/
void setSessionState(String visitId, int state) {
def session = sessions.get(visitId)
if (session) {
session.state = state
logger.debug("Session ${visitId} state changed to ${state}")
}
}
/**
* Update session activity timestamp
* @param visitId The session/visit ID
*/
void touchSession(String visitId) {
def session = sessions.get(visitId)
if (session) {
session.touch()
}
}
/**
* Get session statistics for monitoring
* @return Map of session statistics
*/
Map getStatistics() {
return [
totalSessions: sessions.size(),
usersWithSessions: userSessions.size(),
sessionsPerUser: userSessions.collectEntries { userId, sessionSet ->
[(userId): sessionSet.size()]
}
]
}
}
/**
* Represents an MCP session state
*/
class McpSession {
static final int STATE_UNINITIALIZED = 0
static final int STATE_INITIALIZING = 1
static final int STATE_INITIALIZED = 2
String visitId
String userId
int state = STATE_UNINITIALIZED
long lastActivity = System.currentTimeMillis()
long createdAt = System.currentTimeMillis()
// SSE writer reference (for active connections)
PrintWriter sseWriter
// Notification queue for this session
List<Map> notificationQueue = Collections.synchronizedList(new ArrayList<>())
// Subscriptions (method names this session is subscribed to)
Set<String> subscriptions = Collections.newSetFromMap(new ConcurrentHashMap<>())
void touch() {
lastActivity = System.currentTimeMillis()
}
boolean isActive() {
return state == STATE_INITIALIZED && sseWriter != null && !sseWriter.checkError()
}
boolean hasActiveWriter() {
return sseWriter != null && !sseWriter.checkError()
}
long getDurationMs() {
return System.currentTimeMillis() - createdAt
}
Map toMap() {
return [
visitId: visitId,
userId: userId,
state: state,
lastActivity: lastActivity,
createdAt: createdAt,
durationMs: getDurationMs(),
active: isActive(),
hasWriter: sseWriter != null,
queuedNotifications: notificationQueue.size(),
subscriptions: subscriptions.toList()
]
}
}
/*
* 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.adapter
import org.moqui.context.ExecutionContext
import org.slf4j.Logger
import org.slf4j.LoggerFactory
/**
* Adapter that maps MCP tool calls to Moqui services.
* Provides a clean translation layer between MCP protocol and Moqui service framework.
*/
class McpToolAdapter {
protected final static Logger logger = LoggerFactory.getLogger(McpToolAdapter.class)
// MCP tool name → Moqui service name mapping
private static final Map<String, String> TOOL_SERVICE_MAP = [
'moqui_browse_screens': 'McpServices.mcp#BrowseScreens',
'moqui_search_screens': 'McpServices.mcp#SearchScreens',
'moqui_get_screen_details': 'McpServices.mcp#GetScreenDetails',
'moqui_get_help': 'McpServices.mcp#GetHelp'
]
// MCP method → Moqui service name mapping for JSON-RPC methods
private static final Map<String, String> METHOD_SERVICE_MAP = [
'initialize': 'McpServices.mcp#Initialize',
'ping': 'McpServices.mcp#Ping',
'tools/list': 'McpServices.list#Tools',
'tools/call': 'McpServices.mcp#ToolsCall',
'resources/list': 'McpServices.mcp#ResourcesList',
'resources/read': 'McpServices.mcp#ResourcesRead',
'resources/templates/list': 'McpServices.mcp#ResourcesTemplatesList',
'resources/subscribe': 'McpServices.mcp#ResourcesSubscribe',
'resources/unsubscribe': 'McpServices.mcp#ResourcesUnsubscribe',
'prompts/list': 'McpServices.mcp#PromptsList',
'prompts/get': 'McpServices.mcp#PromptsGet',
'roots/list': 'McpServices.mcp#RootsList',
'sampling/createMessage': 'McpServices.mcp#SamplingCreateMessage',
'elicitation/create': 'McpServices.mcp#ElicitationCreate'
]
// Tool descriptions for MCP tool definitions
private static final Map<String, String> TOOL_DESCRIPTIONS = [
'moqui_browse_screens': 'Browse Moqui screen hierarchy and render screen content',
'moqui_search_screens': 'Search for screens by name to find their paths',
'moqui_get_screen_details': 'Get screen field details including dropdown options',
'moqui_get_help': 'Fetch extended documentation for a screen or service'
]
/**
* Call an MCP tool, translating to the appropriate Moqui service
* @param ec The execution context
* @param toolName The MCP tool name
* @param arguments The tool arguments
* @return The result map or error map
*/
Map callTool(ExecutionContext ec, String toolName, Map arguments) {
String serviceName = TOOL_SERVICE_MAP.get(toolName)
if (!serviceName) {
logger.warn("Unknown tool: ${toolName}")
return [error: [code: -32601, message: "Unknown tool: ${toolName}"]]
}
logger.debug("Calling tool ${toolName} -> service ${serviceName} with args: ${arguments}")
try {
ec.artifactExecution.disableAuthz()
def result = ec.service.sync()
.name(serviceName)
.parameters(arguments ?: [:])
.call()
logger.debug("Tool ${toolName} completed successfully")
// Extract result from service response if wrapped
if (result?.containsKey('result')) {
return result.result
}
return result ?: [:]
} catch (Exception e) {
logger.error("Error calling tool ${toolName}: ${e.message}", e)
return [error: [code: -32000, message: e.message]]
} finally {
ec.artifactExecution.enableAuthz()
}
}
/**
* Call an MCP method, translating to the appropriate Moqui service
* @param ec The execution context
* @param method The MCP method name
* @param params The method parameters
* @return The result map or error map
*/
Map callMethod(ExecutionContext ec, String method, Map params) {
String serviceName = METHOD_SERVICE_MAP.get(method)
if (!serviceName) {
logger.warn("Unknown method: ${method}")
return [error: [code: -32601, message: "Method not found: ${method}"]]
}
logger.debug("Calling method ${method} -> service ${serviceName}")
try {
ec.artifactExecution.disableAuthz()
def result = ec.service.sync()
.name(serviceName)
.parameters(params ?: [:])
.call()
logger.debug("Method ${method} completed successfully")
// Extract result from service response if wrapped
if (result?.containsKey('result')) {
return result.result
}
return result ?: [:]
} catch (Exception e) {
logger.error("Error calling method ${method}: ${e.message}", e)
return [error: [code: -32603, message: "Internal error: ${e.message}"]]
} finally {
ec.artifactExecution.enableAuthz()
}
}
/**
* Check if a tool name is valid
* @param toolName The tool name to check
* @return true if the tool is known
*/
boolean isValidTool(String toolName) {
return TOOL_SERVICE_MAP.containsKey(toolName)
}
/**
* Check if a method name is valid (has a service mapping)
* @param method The method name to check
* @return true if the method has a service mapping
*/
boolean isValidMethod(String method) {
return METHOD_SERVICE_MAP.containsKey(method)
}
/**
* Get the service name for a given tool
* @param toolName The tool name
* @return The service name or null if not found
*/
String getServiceForTool(String toolName) {
return TOOL_SERVICE_MAP.get(toolName)
}
/**
* Get the service name for a given method
* @param method The method name
* @return The service name or null if not found
*/
String getServiceForMethod(String method) {
return METHOD_SERVICE_MAP.get(method)
}
/**
* Get the list of available tools with their definitions
* @return List of tool definition maps
*/
List<Map> listTools() {
return TOOL_SERVICE_MAP.keySet().collect { toolName ->
[
name: toolName,
description: TOOL_DESCRIPTIONS.get(toolName) ?: "MCP tool: ${toolName}",
serviceName: TOOL_SERVICE_MAP.get(toolName)
]
}
}
/**
* Get tool description
* @param toolName The tool name
* @return The tool description or null if not found
*/
String getToolDescription(String toolName) {
return TOOL_DESCRIPTIONS.get(toolName)
}
/**
* Get all supported tool names
* @return Set of tool names
*/
Set<String> getToolNames() {
return TOOL_SERVICE_MAP.keySet()
}
/**
* Get all supported method names
* @return Set of method names
*/
Set<String> getMethodNames() {
return METHOD_SERVICE_MAP.keySet()
}
}
/*
* 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.adapter
import org.moqui.context.ExecutionContextFactory
import org.moqui.context.NotificationMessage
import org.moqui.context.NotificationMessageListener
import org.moqui.mcp.transport.MoquiMcpTransport
import org.slf4j.Logger
import org.slf4j.LoggerFactory
/**
* Bridge that connects Moqui's NotificationMessage system to MCP notifications.
* Implements NotificationMessageListener to receive all Moqui notifications
* and forwards them to MCP clients via the transport layer.
*/
class MoquiNotificationMcpBridge implements NotificationMessageListener {
protected final static Logger logger = LoggerFactory.getLogger(MoquiNotificationMcpBridge.class)
private ExecutionContextFactory ecf
private MoquiMcpTransport transport
// Topic prefix for MCP-specific notifications (optional filtering)
private static final String MCP_TOPIC_PREFIX = "mcp."
// Whether to forward all notifications or only MCP-prefixed ones
private boolean forwardAllNotifications = true
/**
* Initialize the bridge with the ECF and transport
* Note: This method signature matches what the ECF registration expects
*/
@Override
void init(ExecutionContextFactory ecf) {
this.ecf = ecf
logger.info("MoquiNotificationMcpBridge initialized (transport not yet set)")
}
/**
* Set the transport after initialization
* @param transport The MCP transport to use for sending notifications
*/
void setTransport(MoquiMcpTransport transport) {
this.transport = transport
logger.info("MoquiNotificationMcpBridge transport configured: ${transport?.class?.simpleName}")
}
/**
* Configure whether to forward all notifications or only MCP-prefixed ones
* @param forwardAll If true, forward all notifications; if false, only forward those with topic starting with 'mcp.'
*/
void setForwardAllNotifications(boolean forwardAll) {
this.forwardAllNotifications = forwardAll
logger.info("MoquiNotificationMcpBridge forwardAllNotifications set to: ${forwardAll}")
}
@Override
void onMessage(NotificationMessage nm) {
if (transport == null) {
logger.trace("Transport not configured, skipping notification: ${nm.topic}")
return
}
// Optionally filter by topic prefix
if (!forwardAllNotifications && !nm.topic?.startsWith(MCP_TOPIC_PREFIX)) {
logger.trace("Skipping non-MCP notification: ${nm.topic}")
return
}
try {
// Convert Moqui notification → MCP notification format
Map mcpNotification = convertToMcpNotification(nm)
// Get target users
Set<String> notifyUserIds = nm.getNotifyUserIds()
if (notifyUserIds && !notifyUserIds.isEmpty()) {
// Send to each target user's active MCP sessions
int sentCount = 0
for (String userId in notifyUserIds) {
try {
transport.sendNotificationToUser(userId, mcpNotification)
sentCount++
logger.debug("Sent MCP notification to user ${userId}: ${nm.topic}")
} catch (Exception e) {
logger.warn("Failed to send MCP notification to user ${userId}: ${e.message}")
}
}
logger.info("Forwarded Moqui notification '${nm.topic}' to ${sentCount} users via MCP")
} else {
// No specific users, could broadcast or log
logger.debug("Notification '${nm.topic}' has no target users, skipping MCP forward")
}
} catch (Exception e) {
logger.error("Error converting/sending Moqui notification to MCP: ${e.message}", e)
}
}
/**
* Convert a Moqui NotificationMessage to MCP notification format
* @param nm The Moqui notification
* @return The MCP notification map
*/
private Map convertToMcpNotification(NotificationMessage nm) {
return [
jsonrpc: "2.0",
method: "notifications/message",
params: [
topic: nm.topic,
subTopic: nm.subTopic,
title: nm.title,
type: nm.type,
message: nm.getMessageMap() ?: [:],
link: nm.link,
showAlert: nm.isShowAlert(),
notificationMessageId: nm.notificationMessageId,
timestamp: System.currentTimeMillis()
]
]
}
/**
* Create a custom MCP notification and send to specific users
* @param topic The notification topic
* @param title The notification title
* @param message The message content
* @param userIds The target user IDs
*/
void sendMcpNotification(String topic, String title, Map message, Set<String> userIds) {
if (transport == null) {
logger.warn("Cannot send MCP notification: transport not configured")
return
}
Map mcpNotification = [
jsonrpc: "2.0",
method: "notifications/message",
params: [
topic: topic,
title: title,
message: message,
timestamp: System.currentTimeMillis()
]
]
for (String userId in userIds) {
try {
transport.sendNotificationToUser(userId, mcpNotification)
logger.debug("Sent custom MCP notification to user ${userId}: ${topic}")
} catch (Exception e) {
logger.warn("Failed to send custom MCP notification to user ${userId}: ${e.message}")
}
}
}
/**
* Broadcast an MCP notification to all active sessions
* @param topic The notification topic
* @param title The notification title
* @param message The message content
*/
void broadcastMcpNotification(String topic, String title, Map message) {
if (transport == null) {
logger.warn("Cannot broadcast MCP notification: transport not configured")
return
}
Map mcpNotification = [
jsonrpc: "2.0",
method: "notifications/message",
params: [
topic: topic,
title: title,
message: message,
timestamp: System.currentTimeMillis()
]
]
try {
transport.broadcastNotification(mcpNotification)
logger.info("Broadcast MCP notification: ${topic}")
} catch (Exception e) {
logger.error("Failed to broadcast MCP notification: ${e.message}", e)
}
}
/**
* Send a tools/list_changed notification to inform clients that available tools have changed
*/
void notifyToolsChanged() {
if (transport == null) {
logger.warn("Cannot send tools changed notification: transport not configured")
return
}
Map notification = [
jsonrpc: "2.0",
method: "notifications/tools/list_changed",
params: [:]
]
try {
transport.broadcastNotification(notification)
logger.info("Broadcast tools/list_changed notification")
} catch (Exception e) {
logger.error("Failed to broadcast tools changed notification: ${e.message}", e)
}
}
/**
* Send a resources/list_changed notification
*/
void notifyResourcesChanged() {
if (transport == null) return
Map notification = [
jsonrpc: "2.0",
method: "notifications/resources/list_changed",
params: [:]
]
try {
transport.broadcastNotification(notification)
logger.info("Broadcast resources/list_changed notification")
} catch (Exception e) {
logger.error("Failed to broadcast resources changed notification: ${e.message}", e)
}
}
/**
* Send a progress notification for a long-running operation
* @param sessionId The target session
* @param progressToken The progress token
* @param progress Current progress value
* @param total Total progress value (optional)
*/
void sendProgressNotification(String sessionId, String progressToken, Number progress, Number total = null) {
if (transport == null) return
Map notification = [
jsonrpc: "2.0",
method: "notifications/progress",
params: [
progressToken: progressToken,
progress: progress,
total: total
]
]
try {
transport.sendNotification(sessionId, notification)
logger.debug("Sent progress notification to session ${sessionId}: ${progress}/${total ?: '?'}")
} catch (Exception e) {
logger.warn("Failed to send progress notification: ${e.message}")
}
}
@Override
void destroy() {
logger.info("MoquiNotificationMcpBridge destroyed")
this.ecf = null
this.transport = null
}
}
/*
* 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.transport
/**
* Transport interface for MCP messages.
* Abstracts transport concerns so implementations can be swapped (SSE, WebSocket, etc.)
*/
interface MoquiMcpTransport {
// Session lifecycle
/**
* Open a new MCP session for the given user
* @param sessionId The session ID (typically Visit ID)
* @param userId The user ID associated with this session
*/
void openSession(String sessionId, String userId)
/**
* Close an existing MCP session
* @param sessionId The session ID to close
*/
void closeSession(String sessionId)
/**
* Check if a session is currently active
* @param sessionId The session ID to check
* @return true if the session is active
*/
boolean isSessionActive(String sessionId)
// Message sending
/**
* Send a JSON-RPC message to a specific session
* @param sessionId The target session ID
* @param message The message to send (will be JSON-serialized)
*/
void sendMessage(String sessionId, Map message)
/**
* Send an MCP notification to a specific session
* @param sessionId The target session ID
* @param notification The notification to send
*/
void sendNotification(String sessionId, Map notification)
/**
* Send an MCP notification to all sessions for a specific user
* @param userId The target user ID
* @param notification The notification to send
*/
void sendNotificationToUser(String userId, Map notification)
// Broadcast
/**
* Broadcast a notification to all active sessions
* @param notification The notification to broadcast
*/
void broadcastNotification(Map notification)
/**
* Get the number of active sessions
* @return count of active sessions
*/
int getActiveSessionCount()
/**
* Get session IDs for a specific user
* @param userId The user ID
* @return Set of session IDs for this user
*/
Set<String> getSessionsForUser(String userId)
}
/*
* 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.transport
import groovy.json.JsonOutput
import org.moqui.mcp.adapter.McpSession
import org.moqui.mcp.adapter.McpSessionAdapter
import org.slf4j.Logger
import org.slf4j.LoggerFactory
/**
* SSE (Server-Sent Events) implementation of MoquiMcpTransport.
* Uses McpSessionAdapter for session management and provides SSE-based message delivery.
*/
class SseTransport implements MoquiMcpTransport {
protected final static Logger logger = LoggerFactory.getLogger(SseTransport.class)
private final McpSessionAdapter sessionAdapter
// Event ID counter for SSE
private long eventIdCounter = 0
SseTransport(McpSessionAdapter sessionAdapter) {
this.sessionAdapter = sessionAdapter
}
@Override
void openSession(String sessionId, String userId) {
if (!sessionAdapter.hasSession(sessionId)) {
sessionAdapter.createSession(sessionId, userId)
logger.info("Opened SSE session ${sessionId} for user ${userId}")
} else {
logger.debug("Session ${sessionId} already exists")
}
}
@Override
void closeSession(String sessionId) {
def session = sessionAdapter.getSession(sessionId)
if (session) {
// Try to send close event before removing
if (session.hasActiveWriter()) {
try {
def closeData = [
type: "disconnected",
sessionId: sessionId,
timestamp: System.currentTimeMillis()
]
sendSseEvent(session.sseWriter, "close", JsonOutput.toJson(closeData))
} catch (Exception e) {
logger.debug("Could not send close event to session ${sessionId}: ${e.message}")
}
}
sessionAdapter.closeSession(sessionId)
logger.info("Closed SSE session ${sessionId}")
}
}
@Override
boolean isSessionActive(String sessionId) {
def session = sessionAdapter.getSession(sessionId)
return session?.isActive() ?: false
}
@Override
void sendMessage(String sessionId, Map message) {
def session = sessionAdapter.getSession(sessionId)
if (!session) {
logger.warn("Cannot send message: session ${sessionId} not found")
return
}
if (!session.hasActiveWriter()) {
// Queue message for later delivery
session.notificationQueue.add(message)
logger.debug("Queued message for session ${sessionId} (no active writer)")
return
}
try {
String jsonMessage = JsonOutput.toJson(message)
sendSseEvent(session.sseWriter, "message", jsonMessage)
session.touch()
logger.debug("Sent message to session ${sessionId}")
} catch (Exception e) {
logger.warn("Failed to send message to session ${sessionId}: ${e.message}")
// Queue for later if send fails
session.notificationQueue.add(message)
}
}
@Override
void sendNotification(String sessionId, Map notification) {
def session = sessionAdapter.getSession(sessionId)
if (!session) {
logger.warn("Cannot send notification: session ${sessionId} not found")
return
}
// Ensure notification has proper JSON-RPC format
if (!notification.jsonrpc) {
notification = [
jsonrpc: "2.0",
method: notification.method ?: "notifications/message",
params: notification.params ?: notification
]
}
if (!session.hasActiveWriter()) {
// Queue notification for later delivery
session.notificationQueue.add(notification)
logger.debug("Queued notification for session ${sessionId} (no active writer)")
return
}
try {
String jsonNotification = JsonOutput.toJson(notification)
sendSseEvent(session.sseWriter, "message", jsonNotification)
session.touch()
logger.debug("Sent notification to session ${sessionId}: ${notification.method}")
} catch (Exception e) {
logger.warn("Failed to send notification to session ${sessionId}: ${e.message}")
// Queue for later if send fails
session.notificationQueue.add(notification)
}
}
@Override
void sendNotificationToUser(String userId, Map notification) {
Set<String> sessionIds = sessionAdapter.getSessionsForUser(userId)
if (sessionIds.isEmpty()) {
logger.debug("No active sessions for user ${userId}")
return
}
int sentCount = 0
int queuedCount = 0
for (String sessionId in sessionIds) {
def session = sessionAdapter.getSession(sessionId)
if (session) {
if (session.hasActiveWriter()) {
try {
String jsonNotification = JsonOutput.toJson(notification)
sendSseEvent(session.sseWriter, "message", jsonNotification)
session.touch()
sentCount++
} catch (Exception e) {
logger.warn("Failed to send notification to session ${sessionId}: ${e.message}")
session.notificationQueue.add(notification)
queuedCount++
}
} else {
session.notificationQueue.add(notification)
queuedCount++
}
}
}
logger.debug("Sent notification to user ${userId}: ${sentCount} delivered, ${queuedCount} queued")
}
@Override
void broadcastNotification(Map notification) {
Set<String> allSessionIds = sessionAdapter.getAllSessionIds()
if (allSessionIds.isEmpty()) {
logger.debug("No active sessions for broadcast")
return
}
// Ensure notification has proper JSON-RPC format
if (!notification.jsonrpc) {
notification = [
jsonrpc: "2.0",
method: notification.method ?: "notifications/message",
params: notification.params ?: notification
]
}
int sentCount = 0
int failedCount = 0
for (String sessionId in allSessionIds) {
def session = sessionAdapter.getSession(sessionId)
if (session?.hasActiveWriter()) {
try {
String jsonNotification = JsonOutput.toJson(notification)
sendSseEvent(session.sseWriter, "message", jsonNotification)
session.touch()
sentCount++
} catch (Exception e) {
logger.debug("Failed to broadcast to session ${sessionId}: ${e.message}")
failedCount++
}
} else {
// Queue for sessions without active writers
session?.notificationQueue?.add(notification)
}
}
logger.info("Broadcast notification: ${sentCount} delivered, ${failedCount} failed")
}
@Override
int getActiveSessionCount() {
return sessionAdapter.getSessionCount()
}
@Override
Set<String> getSessionsForUser(String userId) {
return sessionAdapter.getSessionsForUser(userId)
}
/**
* Register an SSE writer for a session
* @param sessionId The session ID
* @param writer The PrintWriter for SSE output
*/
void registerSseWriter(String sessionId, PrintWriter writer) {
def session = sessionAdapter.getSession(sessionId)
if (session) {
session.sseWriter = writer
logger.debug("Registered SSE writer for session ${sessionId}")
// Deliver any queued notifications
deliverQueuedNotifications(sessionId)
} else {
logger.warn("Cannot register SSE writer: session ${sessionId} not found")
}
}
/**
* Unregister the SSE writer for a session (e.g., on disconnect)
* @param sessionId The session ID
*/
void unregisterSseWriter(String sessionId) {
def session = sessionAdapter.getSession(sessionId)
if (session) {
session.sseWriter = null
logger.debug("Unregistered SSE writer for session ${sessionId}")
}
}
/**
* Deliver any queued notifications to a session
* @param sessionId The session ID
*/
void deliverQueuedNotifications(String sessionId) {
def session = sessionAdapter.getSession(sessionId)
if (!session || !session.hasActiveWriter()) {
return
}
List<Map> queue = session.notificationQueue
if (queue.isEmpty()) {
return
}
// Take snapshot and clear queue
List<Map> toDeliver
synchronized (queue) {
toDeliver = new ArrayList<>(queue)
queue.clear()
}
int deliveredCount = 0
for (Map notification in toDeliver) {
try {
String jsonNotification = JsonOutput.toJson(notification)
sendSseEvent(session.sseWriter, "message", jsonNotification)
deliveredCount++
} catch (Exception e) {
logger.warn("Failed to deliver queued notification to ${sessionId}: ${e.message}")
// Re-queue failed notifications
queue.add(notification)
}
}
if (deliveredCount > 0) {
logger.debug("Delivered ${deliveredCount} queued notifications to session ${sessionId}")
}
}
/**
* Send a keep-alive ping to a session
* @param sessionId The session ID
* @return true if ping was sent successfully
*/
boolean sendPing(String sessionId) {
def session = sessionAdapter.getSession(sessionId)
if (!session?.hasActiveWriter()) {
return false
}
try {
def pingData = [
type: "ping",
timestamp: System.currentTimeMillis(),
sessionId: sessionId
]
sendSseEvent(session.sseWriter, "ping", JsonOutput.toJson(pingData))
session.touch()
return true
} catch (Exception e) {
logger.debug("Failed to send ping to session ${sessionId}: ${e.message}")
return false
}
}
/**
* Send an SSE event with proper formatting
* @param writer The output writer
* @param eventType The SSE event type
* @param data The data payload
*/
private void sendSseEvent(PrintWriter writer, String eventType, String data) throws IOException {
if (writer == null || writer.checkError()) {
throw new IOException("Writer is closed or in error state")
}
long eventId = ++eventIdCounter
writer.write("id: ${eventId}\n")
writer.write("event: ${eventType}\n")
writer.write("data: ${data}\n\n")
writer.flush()
if (writer.checkError()) {
throw new IOException("Client disconnected during write")
}
}
/**
* Send an SSE event with a specific event ID
*/
void sendSseEventWithId(PrintWriter writer, String eventType, String data, long eventId) throws IOException {
if (writer == null || writer.checkError()) {
throw new IOException("Writer is closed or in error state")
}
if (eventId >= 0) {
writer.write("id: ${eventId}\n")
}
writer.write("event: ${eventType}\n")
writer.write("data: ${data}\n\n")
writer.flush()
if (writer.checkError()) {
throw new IOException("Client disconnected during write")
}
}
/**
* Get the session adapter (for direct access if needed)
*/
McpSessionAdapter getSessionAdapter() {
return sessionAdapter
}
/**
* Get transport statistics
*/
Map getStatistics() {
def adapterStats = sessionAdapter.getStatistics()
int activeWriters = 0
int totalQueued = 0
for (String sessionId in sessionAdapter.getAllSessionIds()) {
def session = sessionAdapter.getSession(sessionId)
if (session) {
if (session.hasActiveWriter()) activeWriters++
totalQueued += session.notificationQueue.size()
}
}
return adapterStats + [
transportType: "SSE",
activeWriters: activeWriters,
queuedNotifications: totalQueued,
eventIdCounter: eventIdCounter
]
}
}