diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index 52780bef2d..9d874478da 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -82,6 +82,7 @@ + diff --git a/src/ServiceControl.AcceptanceTests.RavenDB/ApprovalFiles/When_mcp_server_is_enabled.Should_list_primary_instance_tools.approved.txt b/src/ServiceControl.AcceptanceTests.RavenDB/ApprovalFiles/When_mcp_server_is_enabled.Should_list_primary_instance_tools.approved.txt new file mode 100644 index 0000000000..d3f82a86b8 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests.RavenDB/ApprovalFiles/When_mcp_server_is_enabled.Should_list_primary_instance_tools.approved.txt @@ -0,0 +1,427 @@ +[ + { + "name": "archive_failed_message", + "description": "Use this tool to dismiss a single failed message that does not need to be retried. Good for questions like: \u0027archive this message\u0027, \u0027dismiss this failure\u0027, or \u0027I do not need to retry this one\u0027. Archiving moves the message out of the unresolved list so it no longer shows up as an active problem. This is an asynchronous operation \u2014 the message will be archived shortly after the request is accepted. If you need to archive many messages with the same root cause, use ArchiveFailureGroup instead.", + "inputSchema": { + "type": "object", + "properties": { + "failedMessageId": { + "description": "The unique message ID from a previous query result", + "type": "string" + } + }, + "required": [ + "failedMessageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "archive_failed_messages", + "description": "Use this tool to dismiss multiple failed messages at once that do not need to be retried. Good for questions like: \u0027archive these messages\u0027, \u0027dismiss these failures\u0027, or \u0027archive messages msg-1, msg-2, msg-3\u0027. Prefer ArchiveFailureGroup when all messages share the same failure cause \u2014 use this tool when you have a specific set of message IDs to archive.", + "inputSchema": { + "type": "object", + "properties": { + "messageIds": { + "description": "The unique message IDs from a previous query result", + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": [ + "messageIds" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "archive_failure_group", + "description": "Use this tool to dismiss an entire failure group \u2014 all messages that failed with the same exception type and stack trace. Good for questions like: \u0027archive this failure group\u0027, \u0027dismiss all NullReferenceException failures\u0027, or \u0027archive the whole group\u0027. This is the most efficient way to archive many related failures at once. You need a group ID, which you can get from GetFailureGroups. Returns InProgress if an archive operation is already running for this group.", + "inputSchema": { + "type": "object", + "properties": { + "groupId": { + "description": "The failure group ID from get_failure_groups results", + "type": "string" + } + }, + "required": [ + "groupId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_errors_summary", + "description": "Use this tool as a quick health check to see how many messages are in each failure state. Good for questions like: \u0027how many errors are there?\u0027, \u0027what is the error situation?\u0027, or \u0027are there unresolved failures?\u0027. Returns counts for unresolved, archived, resolved, and retryissued statuses. This is a good first tool to call when asked about the overall error situation before drilling into specific messages.", + "inputSchema": { + "type": "object", + "properties": {} + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_failed_message_by_id", + "description": "Use this tool to get the full details of a specific failed message, including all processing attempts and exception information. Good for questions like: \u0027show me details for this failed message\u0027, \u0027what exception caused this failure?\u0027, or \u0027how many times has this message failed?\u0027. You need the message\u0027s unique ID, which you can get from GetFailedMessages or GetFailureGroups results. If you only need the most recent failure attempt, use GetFailedMessageLastAttempt instead \u2014 it returns less data.", + "inputSchema": { + "type": "object", + "properties": { + "failedMessageId": { + "description": "The unique message ID from a previous query result", + "type": "string" + } + }, + "required": [ + "failedMessageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_failed_message_last_attempt", + "description": "Use this tool to see how a specific message failed most recently. Good for questions like: \u0027what was the last error for this message?\u0027, \u0027show me the latest exception\u0027, or \u0027what happened on the last attempt?\u0027. Returns the latest processing attempt with its exception, stack trace, and headers. Lighter than GetFailedMessageById when you only care about the most recent failure rather than the full history.", + "inputSchema": { + "type": "object", + "properties": { + "failedMessageId": { + "description": "The unique message ID from a previous query result", + "type": "string" + } + }, + "required": [ + "failedMessageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_failed_messages", + "description": "Use this tool to browse failed messages when the user wants to see what is failing. Good for questions like: \u0027what messages are currently failing?\u0027, \u0027are there failures in a specific queue?\u0027, or \u0027what failed recently?\u0027. Returns a paged list of failed messages with their status, exception details, and queue information. For broad requests, call with no parameters to get the most recent failures \u2014 only add filters when you need to narrow down results. Prefer GetFailedMessagesByEndpoint when the user mentions a specific endpoint.", + "inputSchema": { + "type": "object", + "properties": { + "status": { + "description": "Narrow results to a specific status: unresolved (still failing), resolved (succeeded on retry), archived (dismissed), or retryissued (retry in progress). Omit to include all statuses.", + "type": [ + "string", + "null" + ], + "default": null + }, + "modified": { + "description": "Only return messages modified after this date (ISO 8601). Useful for checking recent failures.", + "type": [ + "string", + "null" + ], + "default": null + }, + "queueAddress": { + "description": "Only return messages from this queue address, e.g. \u0027Sales@machine\u0027. Use when investigating a specific queue.", + "type": [ + "string", + "null" + ], + "default": null + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, message_type, or time_of_failure", + "type": "string", + "default": "time_of_failure" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + } + } + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_failed_messages_by_endpoint", + "description": "Use this tool to see failed messages for a specific NServiceBus endpoint. Good for questions like: \u0027what is failing in the Sales endpoint?\u0027, \u0027show errors for Shipping\u0027, or \u0027are there failures in this endpoint?\u0027. Returns the same paged failure data as GetFailedMessages but scoped to one endpoint. Prefer this tool over GetFailedMessages when the user mentions a specific endpoint name.", + "inputSchema": { + "type": "object", + "properties": { + "endpointName": { + "description": "The NServiceBus endpoint name, e.g. \u0027Sales\u0027 or \u0027Shipping.MessageHandler\u0027", + "type": "string" + }, + "status": { + "description": "Narrow results to a specific status: unresolved, resolved, archived, or retryissued. Omit to include all.", + "type": [ + "string", + "null" + ], + "default": null + }, + "modified": { + "description": "Only return messages modified after this date (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, message_type, or time_of_failure", + "type": "string", + "default": "time_of_failure" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + } + }, + "required": [ + "endpointName" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_failure_groups", + "description": "Use this tool to understand why messages are failing by seeing failures grouped by root cause. Good for questions like: \u0027why are messages failing?\u0027, \u0027what errors are happening?\u0027, \u0027group failures by exception\u0027, or \u0027what are the top failure causes?\u0027. Each group represents a distinct exception type and stack trace, showing how many messages are affected and when failures started and last occurred. This is usually the best starting point for diagnosing production issues \u2014 call it before drilling into individual messages. Call with no parameters to use the default grouping by exception type and stack trace.", + "inputSchema": { + "type": "object", + "properties": { + "classifier": { + "description": "How to group failures. The default \u0027Exception Type and Stack Trace\u0027 is almost always what you want. Use \u0027Message Type\u0027 to group by the NServiceBus message type instead.", + "type": "string", + "default": "Exception Type and Stack Trace" + }, + "classifierFilter": { + "description": "Only include groups matching this filter text", + "type": [ + "string", + "null" + ], + "default": null + } + } + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_retry_history", + "description": "Use this tool to check the history of retry operations. Good for questions like: \u0027has someone already retried these?\u0027, \u0027what happened the last time we retried this group?\u0027, \u0027show retry history\u0027, or \u0027were any retries attempted today?\u0027. Returns which groups were retried, when, and whether the retries succeeded or failed. Use this before retrying a group to avoid duplicate retry attempts.", + "inputSchema": { + "type": "object", + "properties": {} + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "retry_all_failed_messages", + "description": "Use this tool to retry every unresolved failed message across all queues and endpoints. Good for questions like: \u0027retry everything\u0027, \u0027reprocess all failures\u0027, or \u0027retry all failed messages\u0027. This is a broad operation \u2014 prefer RetryFailedMessagesByQueue, RetryAllFailedMessagesByEndpoint, or RetryFailureGroup when you can scope the retry more narrowly.", + "inputSchema": { + "type": "object", + "properties": {} + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "retry_all_failed_messages_by_endpoint", + "description": "Use this tool to retry all failed messages for a specific NServiceBus endpoint. Good for questions like: \u0027retry all failures in the Sales endpoint\u0027, \u0027the bug in Shipping is fixed, retry its failures\u0027, or \u0027reprocess all errors for this endpoint\u0027. Useful when a bug in one endpoint has been fixed and all its failures should be reprocessed.", + "inputSchema": { + "type": "object", + "properties": { + "endpointName": { + "description": "The NServiceBus endpoint name, e.g. \u0027Sales\u0027 or \u0027Shipping.MessageHandler\u0027", + "type": "string" + } + }, + "required": [ + "endpointName" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "retry_failed_message", + "description": "Use this tool to reprocess a single failed message by sending it back to its original queue. Good for questions like: \u0027retry this message\u0027, \u0027reprocess this failure\u0027, or \u0027send this message back for processing\u0027. The message will go through normal processing again. Only use after the underlying issue (bug fix, infrastructure problem) has been resolved. If you need to retry many messages with the same root cause, use RetryFailureGroup instead.", + "inputSchema": { + "type": "object", + "properties": { + "failedMessageId": { + "description": "The unique message ID from a previous query result", + "type": "string" + } + }, + "required": [ + "failedMessageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "retry_failed_messages", + "description": "Use this tool to reprocess multiple specific failed messages at once. Good for questions like: \u0027retry these messages\u0027, \u0027reprocess messages msg-1, msg-2, msg-3\u0027, or \u0027retry this batch\u0027. Prefer RetryFailureGroup when all messages share the same failure cause \u2014 use this tool when you have a specific set of message IDs to retry.", + "inputSchema": { + "type": "object", + "properties": { + "messageIds": { + "description": "The unique message IDs from a previous query result", + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": [ + "messageIds" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "retry_failed_messages_by_queue", + "description": "Use this tool to retry all unresolved failed messages from a specific queue. Good for questions like: \u0027retry all failures in the Sales queue\u0027, \u0027reprocess everything from this queue\u0027, or \u0027the queue consumer is back, retry its failures\u0027. Useful when a queue\u0027s consumer was down or misconfigured and is now fixed. Only retries messages with unresolved status.", + "inputSchema": { + "type": "object", + "properties": { + "queueAddress": { + "description": "The full queue address including machine name, e.g. \u0027Sales@machine\u0027", + "type": "string" + } + }, + "required": [ + "queueAddress" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "retry_failure_group", + "description": "Use this tool to retry all failed messages that share the same exception type and stack trace. Good for questions like: \u0027retry this failure group\u0027, \u0027the bug causing these NullReferenceExceptions is fixed, retry them\u0027, or \u0027retry all messages in this group\u0027. This is the most targeted way to retry related failures after fixing a specific bug. You need a group ID, which you can get from GetFailureGroups. Returns InProgress if a retry is already running for this group.", + "inputSchema": { + "type": "object", + "properties": { + "groupId": { + "description": "The failure group ID from get_failure_groups results", + "type": "string" + } + }, + "required": [ + "groupId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "unarchive_failed_message", + "description": "Use this tool to restore a previously archived failed message back to the unresolved list so it can be retried. Good for questions like: \u0027unarchive this message\u0027, \u0027restore this failure\u0027, or \u0027I need to retry this archived message\u0027. Use when a message was archived by mistake or when the underlying issue has been fixed and the message should be reprocessed. If you need to restore many messages from the same failure group, use UnarchiveFailureGroup instead.", + "inputSchema": { + "type": "object", + "properties": { + "failedMessageId": { + "description": "The unique message ID to restore", + "type": "string" + } + }, + "required": [ + "failedMessageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "unarchive_failed_messages", + "description": "Use this tool to restore multiple previously archived failed messages back to the unresolved list. Good for questions like: \u0027unarchive these messages\u0027, \u0027restore these failures\u0027, or \u0027unarchive messages msg-1, msg-2, msg-3\u0027. Prefer UnarchiveFailureGroup when restoring an entire group \u2014 use this tool when you have a specific set of message IDs.", + "inputSchema": { + "type": "object", + "properties": { + "messageIds": { + "description": "The unique message IDs to restore", + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": [ + "messageIds" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "unarchive_failure_group", + "description": "Use this tool to restore an entire archived failure group back to the unresolved list. Good for questions like: \u0027unarchive this failure group\u0027, \u0027restore all archived NullReferenceException failures\u0027, or \u0027unarchive the whole group\u0027. All messages that were archived together under this group will become available for retry again. You need a group ID, which you can get from GetFailureGroups. Returns InProgress if an unarchive operation is already running for this group.", + "inputSchema": { + "type": "object", + "properties": { + "groupId": { + "description": "The failure group ID from get_failure_groups results", + "type": "string" + } + }, + "required": [ + "groupId" + ] + }, + "execution": { + "taskSupport": "optional" + } + } +] \ No newline at end of file diff --git a/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj b/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj index f8033e858f..1923947295 100644 --- a/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj +++ b/src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj @@ -22,6 +22,7 @@ + diff --git a/src/ServiceControl.AcceptanceTests/Mcp/When_mcp_server_is_enabled.cs b/src/ServiceControl.AcceptanceTests/Mcp/When_mcp_server_is_enabled.cs new file mode 100644 index 0000000000..c3928caa07 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Mcp/When_mcp_server_is_enabled.cs @@ -0,0 +1,151 @@ +namespace ServiceControl.AcceptanceTests.Mcp; + +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Net.Http.Json; +using System.Text.Json; +using System.Threading.Tasks; +using AcceptanceTesting; +using NServiceBus.AcceptanceTesting; +using NUnit.Framework; +using Particular.Approvals; + +[TestFixture] +class When_mcp_server_is_enabled : AcceptanceTest +{ + [SetUp] + public void EnableMcp() => SetSettings = s => s.EnableMcpServer = true; + + [Test] + public async Task Should_expose_mcp_endpoint() + { + await Define() + .Done(async _ => + { + var response = await InitializeMcpSession(); + return response.StatusCode == HttpStatusCode.OK; + }) + .Run(); + } + + [Test] + public async Task Should_list_primary_instance_tools() + { + string toolsJson = null; + + await Define() + .Done(async _ => + { + var sessionId = await InitializeAndGetSessionId(); + if (sessionId == null) + { + return false; + } + + var response = await SendMcpRequest(sessionId, "tools/list", new { }); + if (response == null) + { + return false; + } + + toolsJson = await ReadMcpResponseJson(response); + return response.StatusCode == HttpStatusCode.OK; + }) + .Run(); + + Assert.That(toolsJson, Is.Not.Null); + var mcpResponse = JsonSerializer.Deserialize(toolsJson, JsonOptions)!; + var sortedTools = mcpResponse.Result.Tools.Cast().OrderBy(t => t.GetProperty("name").GetString()).ToList(); + var formattedTools = JsonSerializer.Serialize(sortedTools, new JsonSerializerOptions { WriteIndented = true, PropertyNamingPolicy = JsonNamingPolicy.CamelCase }); + Approver.Verify(formattedTools); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpListToolsResponse + { + public McpListToolsResult Result { get; set; } + } + + class McpListToolsResult + { + public List Tools { get; set; } = []; + } + + async Task InitializeMcpSession() + { + var request = new HttpRequestMessage(HttpMethod.Post, "/mcp") + { + Content = JsonContent.Create(new + { + jsonrpc = "2.0", + id = 1, + method = "initialize", + @params = new + { + protocolVersion = "2025-03-26", + capabilities = new { }, + clientInfo = new { name = "test-client", version = "1.0" } + } + }) + }; + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")); + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("text/event-stream")); + return await HttpClient.SendAsync(request); + } + + async Task InitializeAndGetSessionId() + { + var response = await InitializeMcpSession(); + if (response.StatusCode != HttpStatusCode.OK) + { + return null; + } + + if (response.Headers.TryGetValues("mcp-session-id", out var values)) + { + return values.FirstOrDefault(); + } + + return null; + } + + static async Task ReadMcpResponseJson(HttpResponseMessage response) + { + var body = await response.Content.ReadAsStringAsync(); + var contentType = response.Content.Headers.ContentType?.MediaType; + + if (contentType == "text/event-stream") + { + foreach (var line in body.Split('\n')) + { + if (line.StartsWith("data: ")) + { + return line.Substring("data: ".Length); + } + } + } + + return body; + } + + async Task SendMcpRequest(string sessionId, string method, object @params) + { + var request = new HttpRequestMessage(HttpMethod.Post, "/mcp") + { + Content = JsonContent.Create(new + { + jsonrpc = "2.0", + id = 2, + method, + @params + }) + }; + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")); + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("text/event-stream")); + request.Headers.Add("mcp-session-id", sessionId); + return await HttpClient.SendAsync(request); + } +} diff --git a/src/ServiceControl.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs b/src/ServiceControl.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs index 657a84244d..b6b3b8048a 100644 --- a/src/ServiceControl.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs +++ b/src/ServiceControl.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs @@ -125,7 +125,7 @@ async Task InitializeServiceControl(ScenarioContext context) hostBuilder.AddServiceControlAuthentication(settings.OpenIdConnectSettings); hostBuilder.AddServiceControl(settings, configuration); hostBuilder.AddServiceControlHttps(settings.HttpsSettings); - hostBuilder.AddServiceControlApi(settings.CorsSettings); + hostBuilder.AddServiceControlApi(settings); hostBuilder.AddServiceControlTesting(settings); @@ -135,7 +135,7 @@ async Task InitializeServiceControl(ScenarioContext context) host.UseTestRemoteIp(); host.UseServiceControlAuthentication(settings.OpenIdConnectSettings.Enabled); - host.UseServiceControl(settings.ForwardedHeadersSettings, settings.HttpsSettings); + host.UseServiceControl(settings.ForwardedHeadersSettings, settings.HttpsSettings, settings.EnableMcpServer); await host.StartAsync(); DomainEvents = host.Services.GetRequiredService(); // Bring this back and look into the base address of the client diff --git a/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ApprovalFiles/When_mcp_server_is_enabled.Should_list_audit_message_tools.approved.txt b/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ApprovalFiles/When_mcp_server_is_enabled.Should_list_audit_message_tools.approved.txt new file mode 100644 index 0000000000..017bd25122 --- /dev/null +++ b/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ApprovalFiles/When_mcp_server_is_enabled.Should_list_audit_message_tools.approved.txt @@ -0,0 +1,266 @@ +[ + { + "name": "get_audit_message_body", + "description": "Use this tool to inspect the actual payload of a processed message. Good for questions like: \u0027show me the message body\u0027, \u0027what data was in this message?\u0027, or \u0027let me see the content of message X\u0027. Returns the serialized message body content, typically JSON. You need a message ID, which you can get from any audit message query result. Use this when the user wants to see what data was actually sent, not just message metadata.", + "inputSchema": { + "type": "object", + "properties": { + "messageId": { + "description": "The message ID from a previous audit message query result", + "type": "string" + } + }, + "required": [ + "messageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_audit_messages", + "description": "Use this tool to browse successfully processed audit messages when the user wants an overview rather than a text search. Good for questions like: \u0027show recent audit messages\u0027, \u0027what messages were processed today?\u0027, \u0027list messages from endpoint X\u0027, or \u0027show slow messages\u0027. Returns message metadata such as message type, endpoints, sent time, processed time, and timing metrics. For broad requests, use the default paging and sorting. Prefer this tool over SearchAuditMessages when the user does not provide a specific keyword or phrase. If the user is looking for a specific term, id, or text fragment, use SearchAuditMessages instead.", + "inputSchema": { + "type": "object", + "properties": { + "includeSystemMessages": { + "description": "Set to true to include NServiceBus infrastructure messages. Usually leave as false to see only business messages.", + "type": "boolean", + "default": false + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + }, + "timeSentFrom": { + "description": "Only return messages sent after this time (ISO 8601). Use with timeSentTo to query a specific time window.", + "type": [ + "string", + "null" + ], + "default": null + }, + "timeSentTo": { + "description": "Only return messages sent before this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + } + } + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_audit_messages_by_conversation", + "description": "Use this tool to trace the full chain of messages triggered by an initial message. Good for questions like: \u0027what happened after this message was sent?\u0027, \u0027show me the full message flow\u0027, or \u0027trace this conversation\u0027. A conversation groups all related messages together \u2014 the original command and every event, reply, or saga message it caused. You need a conversation ID, which you can get from any audit message query result. Essential for understanding message flow and debugging cascading issues.", + "inputSchema": { + "type": "object", + "properties": { + "conversationId": { + "description": "The conversation ID from a previous audit message query result", + "type": "string" + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + } + }, + "required": [ + "conversationId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_audit_messages_by_endpoint", + "description": "Use this tool to see what messages a specific NServiceBus endpoint has processed. Good for questions like: \u0027what messages did Sales process?\u0027, \u0027show messages handled by Shipping\u0027, or \u0027find OrderPlaced messages in the Billing endpoint\u0027. Returns the same metadata as GetAuditMessages but scoped to one endpoint. Prefer this tool over GetAuditMessages when the user mentions a specific endpoint name. Optionally pass a keyword to search within that endpoint\u0027s messages.", + "inputSchema": { + "type": "object", + "properties": { + "endpointName": { + "description": "The NServiceBus endpoint name, e.g. \u0027Sales\u0027 or \u0027Shipping.MessageHandler\u0027", + "type": "string" + }, + "keyword": { + "description": "Optional keyword to search within this endpoint\u0027s messages", + "type": [ + "string", + "null" + ], + "default": null + }, + "includeSystemMessages": { + "description": "Set to true to include NServiceBus infrastructure messages", + "type": "boolean", + "default": false + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + }, + "timeSentFrom": { + "description": "Only return messages sent after this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + }, + "timeSentTo": { + "description": "Only return messages sent before this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + } + }, + "required": [ + "endpointName" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_endpoint_audit_counts", + "description": "Use this tool to see daily message volume trends for a specific endpoint. Good for questions like: \u0027how much traffic does Sales handle?\u0027, \u0027has throughput changed recently?\u0027, or \u0027show me message counts for this endpoint\u0027. Returns message counts per day, which helps identify throughput changes, traffic spikes, or drops in activity that might indicate problems. You need an endpoint name \u2014 use GetKnownEndpoints first if you do not have one.", + "inputSchema": { + "type": "object", + "properties": { + "endpointName": { + "description": "The NServiceBus endpoint name, e.g. \u0027Sales\u0027 or \u0027Shipping.MessageHandler\u0027", + "type": "string" + } + }, + "required": [ + "endpointName" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_known_endpoints", + "description": "Use this tool to discover what NServiceBus endpoints exist in the system. Good for questions like: \u0027what endpoints do we have?\u0027, \u0027what services are running?\u0027, or \u0027list all endpoints\u0027. Returns all endpoints that have processed audit messages, including their name and host information. This is a good starting point when you need an endpoint name for other tools like GetAuditMessagesByEndpoint or GetEndpointAuditCounts.", + "inputSchema": { + "type": "object", + "properties": {} + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "search_audit_messages", + "description": "Use this tool to find audit messages by a keyword or phrase. Good for questions like: \u0027find messages containing order 12345\u0027, \u0027search for CustomerCreated messages\u0027, or \u0027look for messages mentioning this ID\u0027. Searches across message body content, headers, and metadata using full-text search. Prefer this tool over GetAuditMessages when the user provides a specific term, identifier, or phrase to search for. If the user just wants to browse recent messages without a search term, use GetAuditMessages instead.", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "description": "Free-text search query \u2014 matches against message body, headers, and metadata", + "type": "string" + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + }, + "timeSentFrom": { + "description": "Only return messages sent after this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + }, + "timeSentTo": { + "description": "Only return messages sent before this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + } + }, + "required": [ + "query" + ] + }, + "execution": { + "taskSupport": "optional" + } + } +] \ No newline at end of file diff --git a/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ServiceControl.Audit.AcceptanceTests.RavenDB.csproj b/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ServiceControl.Audit.AcceptanceTests.RavenDB.csproj index 223e117c20..b47764e74c 100644 --- a/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ServiceControl.Audit.AcceptanceTests.RavenDB.csproj +++ b/src/ServiceControl.Audit.AcceptanceTests.RavenDB/ServiceControl.Audit.AcceptanceTests.RavenDB.csproj @@ -20,6 +20,7 @@ + diff --git a/src/ServiceControl.Audit.AcceptanceTests/ApprovalFiles/When_mcp_server_is_enabled.Should_list_audit_message_tools.approved.txt b/src/ServiceControl.Audit.AcceptanceTests/ApprovalFiles/When_mcp_server_is_enabled.Should_list_audit_message_tools.approved.txt new file mode 100644 index 0000000000..017bd25122 --- /dev/null +++ b/src/ServiceControl.Audit.AcceptanceTests/ApprovalFiles/When_mcp_server_is_enabled.Should_list_audit_message_tools.approved.txt @@ -0,0 +1,266 @@ +[ + { + "name": "get_audit_message_body", + "description": "Use this tool to inspect the actual payload of a processed message. Good for questions like: \u0027show me the message body\u0027, \u0027what data was in this message?\u0027, or \u0027let me see the content of message X\u0027. Returns the serialized message body content, typically JSON. You need a message ID, which you can get from any audit message query result. Use this when the user wants to see what data was actually sent, not just message metadata.", + "inputSchema": { + "type": "object", + "properties": { + "messageId": { + "description": "The message ID from a previous audit message query result", + "type": "string" + } + }, + "required": [ + "messageId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_audit_messages", + "description": "Use this tool to browse successfully processed audit messages when the user wants an overview rather than a text search. Good for questions like: \u0027show recent audit messages\u0027, \u0027what messages were processed today?\u0027, \u0027list messages from endpoint X\u0027, or \u0027show slow messages\u0027. Returns message metadata such as message type, endpoints, sent time, processed time, and timing metrics. For broad requests, use the default paging and sorting. Prefer this tool over SearchAuditMessages when the user does not provide a specific keyword or phrase. If the user is looking for a specific term, id, or text fragment, use SearchAuditMessages instead.", + "inputSchema": { + "type": "object", + "properties": { + "includeSystemMessages": { + "description": "Set to true to include NServiceBus infrastructure messages. Usually leave as false to see only business messages.", + "type": "boolean", + "default": false + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + }, + "timeSentFrom": { + "description": "Only return messages sent after this time (ISO 8601). Use with timeSentTo to query a specific time window.", + "type": [ + "string", + "null" + ], + "default": null + }, + "timeSentTo": { + "description": "Only return messages sent before this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + } + } + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_audit_messages_by_conversation", + "description": "Use this tool to trace the full chain of messages triggered by an initial message. Good for questions like: \u0027what happened after this message was sent?\u0027, \u0027show me the full message flow\u0027, or \u0027trace this conversation\u0027. A conversation groups all related messages together \u2014 the original command and every event, reply, or saga message it caused. You need a conversation ID, which you can get from any audit message query result. Essential for understanding message flow and debugging cascading issues.", + "inputSchema": { + "type": "object", + "properties": { + "conversationId": { + "description": "The conversation ID from a previous audit message query result", + "type": "string" + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + } + }, + "required": [ + "conversationId" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_audit_messages_by_endpoint", + "description": "Use this tool to see what messages a specific NServiceBus endpoint has processed. Good for questions like: \u0027what messages did Sales process?\u0027, \u0027show messages handled by Shipping\u0027, or \u0027find OrderPlaced messages in the Billing endpoint\u0027. Returns the same metadata as GetAuditMessages but scoped to one endpoint. Prefer this tool over GetAuditMessages when the user mentions a specific endpoint name. Optionally pass a keyword to search within that endpoint\u0027s messages.", + "inputSchema": { + "type": "object", + "properties": { + "endpointName": { + "description": "The NServiceBus endpoint name, e.g. \u0027Sales\u0027 or \u0027Shipping.MessageHandler\u0027", + "type": "string" + }, + "keyword": { + "description": "Optional keyword to search within this endpoint\u0027s messages", + "type": [ + "string", + "null" + ], + "default": null + }, + "includeSystemMessages": { + "description": "Set to true to include NServiceBus infrastructure messages", + "type": "boolean", + "default": false + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + }, + "timeSentFrom": { + "description": "Only return messages sent after this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + }, + "timeSentTo": { + "description": "Only return messages sent before this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + } + }, + "required": [ + "endpointName" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_endpoint_audit_counts", + "description": "Use this tool to see daily message volume trends for a specific endpoint. Good for questions like: \u0027how much traffic does Sales handle?\u0027, \u0027has throughput changed recently?\u0027, or \u0027show me message counts for this endpoint\u0027. Returns message counts per day, which helps identify throughput changes, traffic spikes, or drops in activity that might indicate problems. You need an endpoint name \u2014 use GetKnownEndpoints first if you do not have one.", + "inputSchema": { + "type": "object", + "properties": { + "endpointName": { + "description": "The NServiceBus endpoint name, e.g. \u0027Sales\u0027 or \u0027Shipping.MessageHandler\u0027", + "type": "string" + } + }, + "required": [ + "endpointName" + ] + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "get_known_endpoints", + "description": "Use this tool to discover what NServiceBus endpoints exist in the system. Good for questions like: \u0027what endpoints do we have?\u0027, \u0027what services are running?\u0027, or \u0027list all endpoints\u0027. Returns all endpoints that have processed audit messages, including their name and host information. This is a good starting point when you need an endpoint name for other tools like GetAuditMessagesByEndpoint or GetEndpointAuditCounts.", + "inputSchema": { + "type": "object", + "properties": {} + }, + "execution": { + "taskSupport": "optional" + } + }, + { + "name": "search_audit_messages", + "description": "Use this tool to find audit messages by a keyword or phrase. Good for questions like: \u0027find messages containing order 12345\u0027, \u0027search for CustomerCreated messages\u0027, or \u0027look for messages mentioning this ID\u0027. Searches across message body content, headers, and metadata using full-text search. Prefer this tool over GetAuditMessages when the user provides a specific term, identifier, or phrase to search for. If the user just wants to browse recent messages without a search term, use GetAuditMessages instead.", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "description": "Free-text search query \u2014 matches against message body, headers, and metadata", + "type": "string" + }, + "page": { + "description": "Page number, 1-based", + "type": "integer", + "default": 1 + }, + "perPage": { + "description": "Results per page", + "type": "integer", + "default": 50 + }, + "sort": { + "description": "Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time", + "type": "string", + "default": "time_sent" + }, + "direction": { + "description": "Sort direction: asc or desc", + "type": "string", + "default": "desc" + }, + "timeSentFrom": { + "description": "Only return messages sent after this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + }, + "timeSentTo": { + "description": "Only return messages sent before this time (ISO 8601)", + "type": [ + "string", + "null" + ], + "default": null + } + }, + "required": [ + "query" + ] + }, + "execution": { + "taskSupport": "optional" + } + } +] \ No newline at end of file diff --git a/src/ServiceControl.Audit.AcceptanceTests/Mcp/When_mcp_server_is_enabled.cs b/src/ServiceControl.Audit.AcceptanceTests/Mcp/When_mcp_server_is_enabled.cs new file mode 100644 index 0000000000..2c90c8b89b --- /dev/null +++ b/src/ServiceControl.Audit.AcceptanceTests/Mcp/When_mcp_server_is_enabled.cs @@ -0,0 +1,258 @@ +namespace ServiceControl.Audit.AcceptanceTests.Mcp; + +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Net.Http.Json; +using System.Text.Json; +using System.Threading.Tasks; +using AcceptanceTesting; +using AcceptanceTesting.EndpointTemplates; +using Audit.Auditing.MessagesView; +using NServiceBus; +using NServiceBus.AcceptanceTesting; +using NServiceBus.AcceptanceTesting.Customization; +using NServiceBus.Settings; +using NUnit.Framework; +using Particular.Approvals; + +class When_mcp_server_is_enabled : AcceptanceTest +{ + [SetUp] + public void EnableMcp() => SetSettings = s => s.EnableMcpServer = true; + + [Test] + public async Task Should_expose_mcp_endpoint() + { + await Define() + .Done(async _ => + { + var response = await InitializeMcpSession(); + return response.StatusCode == HttpStatusCode.OK; + }) + .Run(); + } + + [Test] + public async Task Should_list_audit_message_tools() + { + string toolsJson = null; + + await Define() + .Done(async _ => + { + var sessionId = await InitializeAndGetSessionId(); + if (sessionId == null) + { + return false; + } + + var response = await SendMcpRequest(sessionId, "tools/list", new { }); + if (response == null) + { + return false; + } + + toolsJson = await ReadMcpResponseJson(response); + return response.StatusCode == HttpStatusCode.OK; + }) + .Run(); + + Assert.That(toolsJson, Is.Not.Null); + var mcpResponse = JsonSerializer.Deserialize(toolsJson, JsonOptions)!; + var sortedTools = mcpResponse.Result.Tools.Cast().OrderBy(t => t.GetProperty("name").GetString()).ToList(); + var formattedTools = JsonSerializer.Serialize(sortedTools, new JsonSerializerOptions { WriteIndented = true, PropertyNamingPolicy = JsonNamingPolicy.CamelCase }); + Approver.Verify(formattedTools); + } + + [Test] + public async Task Should_call_get_audit_messages_tool() + { + string toolResult = null; + + var context = await Define() + .WithEndpoint(b => b.When((bus, c) => bus.Send(new MyMessage()))) + .WithEndpoint() + .Done(async c => + { + if (c.MessageId == null) + { + return false; + } + + // Wait for the message to be ingested + if (!await this.TryGetMany("/api/messages?include_system_messages=false&sort=id", m => m.MessageId == c.MessageId)) + { + return false; + } + + var sessionId = await InitializeAndGetSessionId(); + if (sessionId == null) + { + return false; + } + + var response = await SendMcpRequest(sessionId, "tools/call", new + { + name = "get_audit_messages", + arguments = new { includeSystemMessages = false, page = 1, perPage = 50 } + }); + + if (response == null || response.StatusCode != HttpStatusCode.OK) + { + return false; + } + + toolResult = await ReadMcpResponseJson(response); + return true; + }) + .Run(); + + Assert.That(toolResult, Is.Not.Null); + var mcpResponse = JsonSerializer.Deserialize(toolResult, JsonOptions)!; + var textContent = mcpResponse.Result.Content[0].Text; + var messagesResult = JsonSerializer.Deserialize(textContent, JsonOptions)!; + Assert.That(messagesResult.TotalCount, Is.GreaterThanOrEqualTo(1)); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpListToolsResponse + { + public McpListToolsResult Result { get; set; } + } + + class McpListToolsResult + { + public List Tools { get; set; } = []; + } + + class McpCallToolResponse + { + public McpCallToolResult Result { get; set; } + } + + class McpCallToolResult + { + public List Content { get; set; } = []; + } + + class McpContent + { + public string Text { get; set; } + } + + class McpToolResult + { + public int TotalCount { get; set; } + } + + async Task InitializeMcpSession() + { + var request = new HttpRequestMessage(HttpMethod.Post, "/mcp") + { + Content = JsonContent.Create(new + { + jsonrpc = "2.0", + id = 1, + method = "initialize", + @params = new + { + protocolVersion = "2025-03-26", + capabilities = new { }, + clientInfo = new { name = "test-client", version = "1.0" } + } + }) + }; + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")); + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("text/event-stream")); + return await HttpClient.SendAsync(request); + } + + async Task InitializeAndGetSessionId() + { + var response = await InitializeMcpSession(); + if (response.StatusCode != HttpStatusCode.OK) + { + return null; + } + + if (response.Headers.TryGetValues("mcp-session-id", out var values)) + { + return values.FirstOrDefault(); + } + + return null; + } + + static async Task ReadMcpResponseJson(HttpResponseMessage response) + { + var body = await response.Content.ReadAsStringAsync(); + var contentType = response.Content.Headers.ContentType?.MediaType; + + if (contentType == "text/event-stream") + { + foreach (var line in body.Split('\n')) + { + if (line.StartsWith("data: ")) + { + return line.Substring("data: ".Length); + } + } + } + + return body; + } + + async Task SendMcpRequest(string sessionId, string method, object @params) + { + var request = new HttpRequestMessage(HttpMethod.Post, "/mcp") + { + Content = JsonContent.Create(new + { + jsonrpc = "2.0", + id = 2, + method, + @params + }) + }; + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")); + request.Headers.Accept.Add(new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("text/event-stream")); + request.Headers.Add("mcp-session-id", sessionId); + return await HttpClient.SendAsync(request); + } + + public class Sender : EndpointConfigurationBuilder + { + public Sender() => + EndpointSetup(c => + { + var routing = c.ConfigureRouting(); + routing.RouteToEndpoint(typeof(MyMessage), typeof(Receiver)); + }); + } + + public class Receiver : EndpointConfigurationBuilder + { + public Receiver() => EndpointSetup(); + + public class MyMessageHandler(MyContext testContext, IReadOnlySettings settings) : IHandleMessages + { + public Task Handle(MyMessage message, IMessageHandlerContext context) + { + testContext.EndpointNameOfReceivingEndpoint = settings.EndpointName(); + testContext.MessageId = context.MessageId; + return Task.CompletedTask; + } + } + } + + public class MyMessage : ICommand; + + public class MyContext : ScenarioContext + { + public string MessageId { get; set; } + public string EndpointNameOfReceivingEndpoint { get; set; } + } +} diff --git a/src/ServiceControl.Audit.AcceptanceTests/ServiceControl.Audit.AcceptanceTests.csproj b/src/ServiceControl.Audit.AcceptanceTests/ServiceControl.Audit.AcceptanceTests.csproj index 2bbf59bc45..96b63d8264 100644 --- a/src/ServiceControl.Audit.AcceptanceTests/ServiceControl.Audit.AcceptanceTests.csproj +++ b/src/ServiceControl.Audit.AcceptanceTests/ServiceControl.Audit.AcceptanceTests.csproj @@ -20,6 +20,7 @@ + \ No newline at end of file diff --git a/src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs b/src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs index efcd99c0f6..c197a2e54a 100644 --- a/src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs +++ b/src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs @@ -133,7 +133,7 @@ async Task InitializeServiceControl(ScenarioContext context) return criticalErrorContext.Stop(cancellationToken); }, settings, configuration); - hostBuilder.AddServiceControlAuditApi(settings.CorsSettings); + hostBuilder.AddServiceControlAuditApi(settings); hostBuilder.AddServiceControlHttps(settings.HttpsSettings); hostBuilder.AddServiceControlAuditTesting(settings); @@ -144,7 +144,7 @@ async Task InitializeServiceControl(ScenarioContext context) host.UseTestRemoteIp(); host.UseServiceControlAuthentication(settings.OpenIdConnectSettings.Enabled); - host.UseServiceControlAudit(settings.ForwardedHeadersSettings, settings.HttpsSettings); + host.UseServiceControlAudit(settings.ForwardedHeadersSettings, settings.HttpsSettings, settings.EnableMcpServer); await host.StartAsync(); ServiceProvider = host.Services; InstanceTestServer = host.GetTestServer(); diff --git a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index 83897faeba..fa983cf7d6 100644 --- a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -58,5 +58,6 @@ "ServiceControlQueueAddress": "Particular.ServiceControl", "TimeToRestartAuditIngestionAfterFailure": "00:01:00", "EnableFullTextSearchOnBodies": true, + "EnableMcpServer": false, "ShutdownTimeout": "00:00:05" } \ No newline at end of file diff --git a/src/ServiceControl.Audit.UnitTests/Mcp/AuditMessageMcpToolsTests.cs b/src/ServiceControl.Audit.UnitTests/Mcp/AuditMessageMcpToolsTests.cs new file mode 100644 index 0000000000..6405534165 --- /dev/null +++ b/src/ServiceControl.Audit.UnitTests/Mcp/AuditMessageMcpToolsTests.cs @@ -0,0 +1,209 @@ +#nullable enable + +namespace ServiceControl.Audit.UnitTests.Mcp; + +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Audit.Auditing; +using Audit.Auditing.MessagesView; +using Audit.Infrastructure; +using Audit.Mcp; +using Audit.Monitoring; +using Audit.Persistence; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceControl.SagaAudit; + +[TestFixture] +class AuditMessageMcpToolsTests +{ + StubAuditDataStore store = null!; + AuditMessageTools tools = null!; + + [SetUp] + public void SetUp() + { + store = new StubAuditDataStore(); + tools = new AuditMessageTools(store, NullLogger.Instance); + } + + [Test] + public async Task GetAuditMessages_returns_messages() + { + store.MessagesResult = new QueryResult>( + [new() { MessageId = "msg-1", MessageType = "MyNamespace.MyMessage" }], + new QueryStatsInfo("etag", 1)); + + var result = await tools.GetAuditMessages(); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response.TotalCount, Is.EqualTo(1)); + Assert.That(response.Results, Has.Count.EqualTo(1)); + } + + [Test] + public async Task GetAuditMessages_passes_paging_and_sort_parameters() + { + await tools.GetAuditMessages(page: 2, perPage: 25, sort: "processed_at", direction: "asc"); + + Assert.That(store.LastGetMessagesArgs, Is.Not.Null); + Assert.That(store.LastGetMessagesArgs!.Value.PagingInfo.Page, Is.EqualTo(2)); + Assert.That(store.LastGetMessagesArgs!.Value.PagingInfo.PageSize, Is.EqualTo(25)); + Assert.That(store.LastGetMessagesArgs!.Value.SortInfo.Sort, Is.EqualTo("processed_at")); + Assert.That(store.LastGetMessagesArgs!.Value.SortInfo.Direction, Is.EqualTo("asc")); + } + + [Test] + public async Task SearchAuditMessages_passes_query() + { + await tools.SearchAuditMessages("OrderPlaced"); + + Assert.That(store.LastQueryMessagesSearchParam, Is.EqualTo("OrderPlaced")); + } + + [Test] + public async Task GetAuditMessagesByEndpoint_queries_by_endpoint() + { + await tools.GetAuditMessagesByEndpoint("Sales"); + + Assert.That(store.LastQueryByEndpointName, Is.EqualTo("Sales")); + Assert.That(store.LastQueryByEndpointKeyword, Is.Null); + } + + [Test] + public async Task GetAuditMessagesByEndpoint_with_keyword_uses_keyword_query() + { + await tools.GetAuditMessagesByEndpoint("Sales", keyword: "OrderPlaced"); + + Assert.That(store.LastQueryByEndpointAndKeywordEndpoint, Is.EqualTo("Sales")); + Assert.That(store.LastQueryByEndpointAndKeywordKeyword, Is.EqualTo("OrderPlaced")); + } + + [Test] + public async Task GetAuditMessagesByConversation_queries_by_conversation_id() + { + await tools.GetAuditMessagesByConversation("conv-123"); + + Assert.That(store.LastConversationId, Is.EqualTo("conv-123")); + } + + [Test] + public async Task GetAuditMessageBody_returns_body_content() + { + store.MessageBodyResult = MessageBodyView.FromString("{\"orderId\": 123}", "application/json", 16, "etag"); + + var result = await tools.GetAuditMessageBody("msg-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.ContentType, Is.EqualTo("application/json")); + Assert.That(response.Body, Is.EqualTo("{\"orderId\": 123}")); + } + + [Test] + public async Task GetAuditMessageBody_returns_error_when_not_found() + { + store.MessageBodyResult = MessageBodyView.NotFound(); + + var result = await tools.GetAuditMessageBody("msg-missing"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("not found")); + } + + [Test] + public async Task GetAuditMessageBody_returns_error_when_no_content() + { + store.MessageBodyResult = MessageBodyView.NoContent(); + + var result = await tools.GetAuditMessageBody("msg-empty"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("no body content")); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpToolResponse + { + public int TotalCount { get; set; } + public List Results { get; set; } = []; + } + + class McpMessageBodyResponse + { + public string? ContentType { get; set; } + public int ContentLength { get; set; } + public string? Body { get; set; } + } + + class McpErrorResponse + { + public string? Error { get; set; } + } + + class StubAuditDataStore : IAuditDataStore + { + static readonly QueryResult> EmptyMessagesResult = new([], QueryStatsInfo.Zero); + static readonly QueryResult> EmptyEndpointsResult = new([], QueryStatsInfo.Zero); + static readonly QueryResult> EmptyAuditCountsResult = new([], QueryStatsInfo.Zero); + + public QueryResult>? MessagesResult { get; set; } + public MessageBodyView MessageBodyResult { get; set; } = MessageBodyView.NotFound(); + + // Captured arguments + public (bool IncludeSystemMessages, PagingInfo PagingInfo, SortInfo SortInfo, DateTimeRange? TimeSentRange)? LastGetMessagesArgs { get; private set; } + public string? LastQueryMessagesSearchParam { get; private set; } + public string? LastQueryByEndpointName { get; private set; } + public string? LastQueryByEndpointKeyword { get; private set; } + public string? LastQueryByEndpointAndKeywordEndpoint { get; private set; } + public string? LastQueryByEndpointAndKeywordKeyword { get; private set; } + public string? LastConversationId { get; private set; } + + public Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + { + LastGetMessagesArgs = (includeSystemMessages, pagingInfo, sortInfo, timeSentRange); + return Task.FromResult(MessagesResult ?? EmptyMessagesResult); + } + + public Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + { + LastQueryMessagesSearchParam = searchParam; + return Task.FromResult(MessagesResult ?? EmptyMessagesResult); + } + + public Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + { + LastQueryByEndpointName = endpointName; + LastQueryByEndpointKeyword = null; + return Task.FromResult(MessagesResult ?? EmptyMessagesResult); + } + + public Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + { + LastQueryByEndpointAndKeywordEndpoint = endpoint; + LastQueryByEndpointAndKeywordKeyword = keyword; + return Task.FromResult(MessagesResult ?? EmptyMessagesResult); + } + + public Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken) + { + LastConversationId = conversationId; + return Task.FromResult(MessagesResult ?? EmptyMessagesResult); + } + + public Task GetMessageBody(string messageId, CancellationToken cancellationToken) + => Task.FromResult(MessageBodyResult); + + public Task>> QueryKnownEndpoints(CancellationToken cancellationToken) + => Task.FromResult(EmptyEndpointsResult); + + public Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken) + => Task.FromResult(QueryResult.Empty()); + + public Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken) + => Task.FromResult(EmptyAuditCountsResult); + } +} diff --git a/src/ServiceControl.Audit.UnitTests/Mcp/EndpointMcpToolsTests.cs b/src/ServiceControl.Audit.UnitTests/Mcp/EndpointMcpToolsTests.cs new file mode 100644 index 0000000000..e1c1b0eaf0 --- /dev/null +++ b/src/ServiceControl.Audit.UnitTests/Mcp/EndpointMcpToolsTests.cs @@ -0,0 +1,106 @@ +#nullable enable + +namespace ServiceControl.Audit.UnitTests.Mcp; + +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Audit.Auditing; +using Audit.Auditing.MessagesView; +using Audit.Infrastructure; +using Audit.Mcp; +using Audit.Monitoring; +using Audit.Persistence; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceControl.SagaAudit; + + +[TestFixture] +class EndpointMcpToolsTests +{ + StubAuditDataStore store = null!; + EndpointTools tools = null!; + + [SetUp] + public void SetUp() + { + store = new StubAuditDataStore(); + tools = new EndpointTools(store, NullLogger.Instance); + } + + [Test] + public async Task GetKnownEndpoints_returns_endpoints() + { + store.KnownEndpointsResult = new QueryResult>( + [new() { EndpointDetails = new EndpointDetails { Name = "Sales", Host = "server1" } }], + new QueryStatsInfo("etag", 1)); + + var result = await tools.GetKnownEndpoints(); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response.TotalCount, Is.EqualTo(1)); + Assert.That(response.Results, Has.Count.EqualTo(1)); + } + + [Test] + public async Task GetEndpointAuditCounts_returns_counts() + { + store.AuditCountsResult = new QueryResult>( + [new() { UtcDate = DateTime.UtcNow.Date, Count = 42 }], + new QueryStatsInfo("etag", 1)); + + var result = await tools.GetEndpointAuditCounts("Sales"); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response.TotalCount, Is.EqualTo(1)); + Assert.That(store.LastAuditCountsEndpointName, Is.EqualTo("Sales")); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpToolResponse + { + public int TotalCount { get; set; } + public List Results { get; set; } = []; + } + + class StubAuditDataStore : IAuditDataStore + { + public QueryResult>? KnownEndpointsResult { get; set; } + public QueryResult>? AuditCountsResult { get; set; } + public string? LastAuditCountsEndpointName { get; private set; } + + public Task>> QueryKnownEndpoints(CancellationToken cancellationToken) + => Task.FromResult(KnownEndpointsResult ?? new QueryResult>([], QueryStatsInfo.Zero)); + + public Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken) + { + LastAuditCountsEndpointName = endpointName; + return Task.FromResult(AuditCountsResult ?? new QueryResult>([], QueryStatsInfo.Zero)); + } + + public Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + => Task.FromResult(new QueryResult>([], QueryStatsInfo.Zero)); + + public Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + => Task.FromResult(new QueryResult>([], QueryStatsInfo.Zero)); + + public Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + => Task.FromResult(new QueryResult>([], QueryStatsInfo.Zero)); + + public Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) + => Task.FromResult(new QueryResult>([], QueryStatsInfo.Zero)); + + public Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken) + => Task.FromResult(new QueryResult>([], QueryStatsInfo.Zero)); + + public Task GetMessageBody(string messageId, CancellationToken cancellationToken) + => Task.FromResult(MessageBodyView.NotFound()); + + public Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken) + => Task.FromResult(QueryResult.Empty()); + } +} diff --git a/src/ServiceControl.Audit/App.config b/src/ServiceControl.Audit/App.config index a3f5781c51..8adfc4075d 100644 --- a/src/ServiceControl.Audit/App.config +++ b/src/ServiceControl.Audit/App.config @@ -8,6 +8,8 @@ These settings are only here so that we can debug ServiceControl while developin + + diff --git a/src/ServiceControl.Audit/Infrastructure/EventSourceCreator.cs b/src/ServiceControl.Audit/Infrastructure/EventSourceCreator.cs index 004fd541e4..ea5d898744 100644 --- a/src/ServiceControl.Audit/Infrastructure/EventSourceCreator.cs +++ b/src/ServiceControl.Audit/Infrastructure/EventSourceCreator.cs @@ -8,10 +8,10 @@ static class EventSourceCreator [SupportedOSPlatform("windows")] public static void Create() { - if (!EventLog.SourceExists(SourceName)) - { - EventLog.CreateEventSource(SourceName, null); - } + //if (!EventLog.SourceExists(SourceName)) + //{ + // EventLog.CreateEventSource(SourceName, null); + //} } public const string SourceName = "ServiceControl.Audit"; diff --git a/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/RunCommand.cs b/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/RunCommand.cs index 22e2fff776..7ddfcf46cd 100644 --- a/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/RunCommand.cs +++ b/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/RunCommand.cs @@ -25,10 +25,10 @@ public override async Task Execute(HostArguments args, Settings settings) //Do nothing. The transports in NSB 8 are designed to handle broker outages. Audit ingestion will be paused when broker is unavailable. return Task.CompletedTask; }, settings, endpointConfiguration); - hostBuilder.AddServiceControlAuditApi(settings.CorsSettings); + hostBuilder.AddServiceControlAuditApi(settings); var app = hostBuilder.Build(); - app.UseServiceControlAudit(settings.ForwardedHeadersSettings, settings.HttpsSettings); + app.UseServiceControlAudit(settings.ForwardedHeadersSettings, settings.HttpsSettings, settings.EnableMcpServer); app.UseServiceControlAuthentication(settings.OpenIdConnectSettings.Enabled); await app.RunAsync(settings.RootUrl); diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs index 3203bd349e..22ec971d12 100644 --- a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs @@ -54,6 +54,7 @@ public Settings(string transportType = null, string persisterType = null, Loggin ServiceControlQueueAddress = SettingsReader.Read(SettingsRootNamespace, "ServiceControlQueueAddress"); TimeToRestartAuditIngestionAfterFailure = GetTimeToRestartAuditIngestionAfterFailure(); EnableFullTextSearchOnBodies = SettingsReader.Read(SettingsRootNamespace, "EnableFullTextSearchOnBodies", true); + EnableMcpServer = SettingsReader.Read(SettingsRootNamespace, "EnableMcpServer", false); ShutdownTimeout = SettingsReader.Read(SettingsRootNamespace, "ShutdownTimeout", ShutdownTimeout); AssemblyLoadContextResolver = static assemblyPath => new PluginAssemblyLoadContext(assemblyPath); @@ -187,6 +188,8 @@ public int MaxBodySizeToStore public bool EnableFullTextSearchOnBodies { get; set; } + public bool EnableMcpServer { get; set; } + // The default value is set to the maximum allowed time by the most // restrictive hosting platform, which is Linux containers. Linux // containers allow for a maximum of 10 seconds. We set it to 5 to diff --git a/src/ServiceControl.Audit/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs b/src/ServiceControl.Audit/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs index 638041d4b1..f650640314 100644 --- a/src/ServiceControl.Audit/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs +++ b/src/ServiceControl.Audit/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs @@ -4,13 +4,22 @@ using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; + using ModelContextProtocol.AspNetCore; using ServiceControl.Infrastructure; static class HostApplicationBuilderExtensions { - public static void AddServiceControlAuditApi(this IHostApplicationBuilder builder, CorsSettings corsSettings) + public static void AddServiceControlAuditApi(this IHostApplicationBuilder builder, Settings.Settings settings) { - builder.Services.AddCors(options => options.AddDefaultPolicy(Cors.GetDefaultPolicy(corsSettings))); + if (settings.EnableMcpServer) + { + builder.Services + .AddMcpServer() + .WithHttpTransport() + .WithToolsFromAssembly(); + } + + builder.Services.AddCors(options => options.AddDefaultPolicy(Cors.GetDefaultPolicy(settings.CorsSettings))); // We're not explicitly adding Gzip here because it's already in the default list of supported compressors builder.Services.AddResponseCompression(); diff --git a/src/ServiceControl.Audit/Mcp/AuditMessageTools.cs b/src/ServiceControl.Audit/Mcp/AuditMessageTools.cs new file mode 100644 index 0000000000..f6caa32422 --- /dev/null +++ b/src/ServiceControl.Audit/Mcp/AuditMessageTools.cs @@ -0,0 +1,208 @@ +#nullable enable + +namespace ServiceControl.Audit.Mcp; + +using System.ComponentModel; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Infrastructure; +using Microsoft.Extensions.Logging; +using ModelContextProtocol.Server; +using Persistence; + +[McpServerToolType, Description( + "Tools for exploring audit messages.\n\n" + + "Agent guidance:\n" + + "1. For broad requests like 'show recent messages', start with GetAuditMessages using defaults.\n" + + "2. For requests containing a concrete text term, identifier, or phrase, use SearchAuditMessages.\n" + + "3. Keep page=1 unless the user asks for more results.\n" + + "4. Keep perPage modest, such as 20 to 50, unless the user asks for a larger batch.\n" + + "5. Use time filters when the user mentions a date or time window like 'today' or 'last hour'.\n" + + "6. Only change sorting when the user explicitly asks for it." +)] +public class AuditMessageTools(IAuditDataStore store, ILogger logger) +{ + [McpServerTool, Description( + "Use this tool to browse successfully processed audit messages when the user wants an overview rather than a text search. " + + "Good for questions like: 'show recent audit messages', 'what messages were processed today?', 'list messages from endpoint X', or 'show slow messages'. " + + "Returns message metadata such as message type, endpoints, sent time, processed time, and timing metrics. " + + "For broad requests, use the default paging and sorting. " + + "Prefer this tool over SearchAuditMessages when the user does not provide a specific keyword or phrase. " + + "If the user is looking for a specific term, id, or text fragment, use SearchAuditMessages instead." + )] + public async Task GetAuditMessages( + [Description("Set to true to include NServiceBus infrastructure messages. Usually leave as false to see only business messages.")] bool includeSystemMessages = false, + [Description("Page number, 1-based")] int page = 1, + [Description("Results per page")] int perPage = 50, + [Description("Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time")] string sort = "time_sent", + [Description("Sort direction: asc or desc")] string direction = "desc", + [Description("Only return messages sent after this time (ISO 8601). Use with timeSentTo to query a specific time window.")] string? timeSentFrom = null, + [Description("Only return messages sent before this time (ISO 8601)")] string? timeSentTo = null, + CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP GetAuditMessages invoked (page={Page}, includeSystemMessages={IncludeSystem})", page, includeSystemMessages); + + var pagingInfo = new PagingInfo(page, perPage); + var sortInfo = new SortInfo(sort, direction); + var timeSentRange = new DateTimeRange(timeSentFrom, timeSentTo); + + var results = await store.GetMessages(includeSystemMessages, pagingInfo, sortInfo, timeSentRange, cancellationToken); + + logger.LogInformation("MCP GetAuditMessages returned {Count} results", results.QueryStats.TotalCount); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to find audit messages by a keyword or phrase. " + + "Good for questions like: 'find messages containing order 12345', 'search for CustomerCreated messages', or 'look for messages mentioning this ID'. " + + "Searches across message body content, headers, and metadata using full-text search. " + + "Prefer this tool over GetAuditMessages when the user provides a specific term, identifier, or phrase to search for. " + + "If the user just wants to browse recent messages without a search term, use GetAuditMessages instead." + )] + public async Task SearchAuditMessages( + [Description("Free-text search query — matches against message body, headers, and metadata")] string query, + [Description("Page number, 1-based")] int page = 1, + [Description("Results per page")] int perPage = 50, + [Description("Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time")] string sort = "time_sent", + [Description("Sort direction: asc or desc")] string direction = "desc", + [Description("Only return messages sent after this time (ISO 8601)")] string? timeSentFrom = null, + [Description("Only return messages sent before this time (ISO 8601)")] string? timeSentTo = null, + CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP SearchAuditMessages invoked (query={Query}, page={Page})", query, page); + + var pagingInfo = new PagingInfo(page, perPage); + var sortInfo = new SortInfo(sort, direction); + var timeSentRange = new DateTimeRange(timeSentFrom, timeSentTo); + + var results = await store.QueryMessages(query, pagingInfo, sortInfo, timeSentRange, cancellationToken); + + logger.LogInformation("MCP SearchAuditMessages returned {Count} results", results.QueryStats.TotalCount); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to see what messages a specific NServiceBus endpoint has processed. " + + "Good for questions like: 'what messages did Sales process?', 'show messages handled by Shipping', or 'find OrderPlaced messages in the Billing endpoint'. " + + "Returns the same metadata as GetAuditMessages but scoped to one endpoint. " + + "Prefer this tool over GetAuditMessages when the user mentions a specific endpoint name. " + + "Optionally pass a keyword to search within that endpoint's messages." + )] + public async Task GetAuditMessagesByEndpoint( + [Description("The NServiceBus endpoint name, e.g. 'Sales' or 'Shipping.MessageHandler'")] string endpointName, + [Description("Optional keyword to search within this endpoint's messages")] string? keyword = null, + [Description("Set to true to include NServiceBus infrastructure messages")] bool includeSystemMessages = false, + [Description("Page number, 1-based")] int page = 1, + [Description("Results per page")] int perPage = 50, + [Description("Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time")] string sort = "time_sent", + [Description("Sort direction: asc or desc")] string direction = "desc", + [Description("Only return messages sent after this time (ISO 8601)")] string? timeSentFrom = null, + [Description("Only return messages sent before this time (ISO 8601)")] string? timeSentTo = null, + CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP GetAuditMessagesByEndpoint invoked (endpoint={EndpointName}, keyword={Keyword}, page={Page})", endpointName, keyword, page); + + var pagingInfo = new PagingInfo(page, perPage); + var sortInfo = new SortInfo(sort, direction); + var timeSentRange = new DateTimeRange(timeSentFrom, timeSentTo); + + var results = keyword != null + ? await store.QueryMessagesByReceivingEndpointAndKeyword(endpointName, keyword, pagingInfo, sortInfo, timeSentRange, cancellationToken) + : await store.QueryMessagesByReceivingEndpoint(includeSystemMessages, endpointName, pagingInfo, sortInfo, timeSentRange, cancellationToken); + + logger.LogInformation("MCP GetAuditMessagesByEndpoint returned {Count} results for endpoint '{EndpointName}'", results.QueryStats.TotalCount, endpointName); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to trace the full chain of messages triggered by an initial message. " + + "Good for questions like: 'what happened after this message was sent?', 'show me the full message flow', or 'trace this conversation'. " + + "A conversation groups all related messages together — the original command and every event, reply, or saga message it caused. " + + "You need a conversation ID, which you can get from any audit message query result. " + + "Essential for understanding message flow and debugging cascading issues." + )] + public async Task GetAuditMessagesByConversation( + [Description("The conversation ID from a previous audit message query result")] string conversationId, + [Description("Page number, 1-based")] int page = 1, + [Description("Results per page")] int perPage = 50, + [Description("Sort by: time_sent, processed_at, message_type, critical_time, delivery_time, or processing_time")] string sort = "time_sent", + [Description("Sort direction: asc or desc")] string direction = "desc", + CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP GetAuditMessagesByConversation invoked (conversationId={ConversationId}, page={Page})", conversationId, page); + + var pagingInfo = new PagingInfo(page, perPage); + var sortInfo = new SortInfo(sort, direction); + + var results = await store.QueryMessagesByConversationId(conversationId, pagingInfo, sortInfo, cancellationToken); + + logger.LogInformation("MCP GetAuditMessagesByConversation returned {Count} results", results.QueryStats.TotalCount); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to inspect the actual payload of a processed message. " + + "Good for questions like: 'show me the message body', 'what data was in this message?', or 'let me see the content of message X'. " + + "Returns the serialized message body content, typically JSON. " + + "You need a message ID, which you can get from any audit message query result. " + + "Use this when the user wants to see what data was actually sent, not just message metadata." + )] + public async Task GetAuditMessageBody( + [Description("The message ID from a previous audit message query result")] string messageId, + CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP GetAuditMessageBody invoked (messageId={MessageId})", messageId); + + var result = await store.GetMessageBody(messageId, cancellationToken); + + if (!result.Found) + { + logger.LogWarning("MCP GetAuditMessageBody: message '{MessageId}' not found", messageId); + return JsonSerializer.Serialize(new { Error = $"Message '{messageId}' not found." }, McpJsonOptions.Default); + } + + if (!result.HasContent) + { + logger.LogWarning("MCP GetAuditMessageBody: message '{MessageId}' has no body content", messageId); + return JsonSerializer.Serialize(new { Error = $"Message '{messageId}' has no body content." }, McpJsonOptions.Default); + } + + if (result.StringContent != null) + { + return JsonSerializer.Serialize(new + { + result.ContentType, + result.ContentLength, + Body = result.StringContent + }, McpJsonOptions.Default); + } + + return JsonSerializer.Serialize(new + { + result.ContentType, + result.ContentLength, + Body = "(stream content - not available as text)" + }, McpJsonOptions.Default); + } +} diff --git a/src/ServiceControl.Audit/Mcp/EndpointTools.cs b/src/ServiceControl.Audit/Mcp/EndpointTools.cs new file mode 100644 index 0000000000..cc15e03b43 --- /dev/null +++ b/src/ServiceControl.Audit/Mcp/EndpointTools.cs @@ -0,0 +1,64 @@ +#nullable enable + +namespace ServiceControl.Audit.Mcp; + +using System.ComponentModel; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using ModelContextProtocol.Server; +using Persistence; + +[McpServerToolType, Description( + "Tools for discovering and inspecting NServiceBus endpoints.\n\n" + + "Agent guidance:\n" + + "1. Use GetKnownEndpoints to discover endpoint names before calling endpoint-specific tools.\n" + + "2. Use GetEndpointAuditCounts to spot throughput trends, traffic spikes, or drops in activity." +)] +public class EndpointTools(IAuditDataStore store, ILogger logger) +{ + [McpServerTool, Description( + "Use this tool to discover what NServiceBus endpoints exist in the system. " + + "Good for questions like: 'what endpoints do we have?', 'what services are running?', or 'list all endpoints'. " + + "Returns all endpoints that have processed audit messages, including their name and host information. " + + "This is a good starting point when you need an endpoint name for other tools like GetAuditMessagesByEndpoint or GetEndpointAuditCounts." + )] + public async Task GetKnownEndpoints(CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP GetKnownEndpoints invoked"); + + var results = await store.QueryKnownEndpoints(cancellationToken); + + logger.LogInformation("MCP GetKnownEndpoints returned {Count} endpoints", results.QueryStats.TotalCount); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to see daily message volume trends for a specific endpoint. " + + "Good for questions like: 'how much traffic does Sales handle?', 'has throughput changed recently?', or 'show me message counts for this endpoint'. " + + "Returns message counts per day, which helps identify throughput changes, traffic spikes, or drops in activity that might indicate problems. " + + "You need an endpoint name — use GetKnownEndpoints first if you do not have one." + )] + public async Task GetEndpointAuditCounts( + [Description("The NServiceBus endpoint name, e.g. 'Sales' or 'Shipping.MessageHandler'")] string endpointName, + CancellationToken cancellationToken = default) + { + logger.LogInformation("MCP GetEndpointAuditCounts invoked (endpoint={EndpointName})", endpointName); + + var results = await store.QueryAuditCounts(endpointName, cancellationToken); + + logger.LogInformation("MCP GetEndpointAuditCounts returned {Count} entries for endpoint '{EndpointName}'", results.QueryStats.TotalCount, endpointName); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } +} diff --git a/src/ServiceControl.Audit/Mcp/McpJsonOptions.cs b/src/ServiceControl.Audit/Mcp/McpJsonOptions.cs new file mode 100644 index 0000000000..ff03d91eae --- /dev/null +++ b/src/ServiceControl.Audit/Mcp/McpJsonOptions.cs @@ -0,0 +1,16 @@ +#nullable enable + +namespace ServiceControl.Audit.Mcp; + +using System.Text.Json; +using System.Text.Json.Serialization; + +static class McpJsonOptions +{ + public static JsonSerializerOptions Default { get; } = new() + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + WriteIndented = false + }; +} diff --git a/src/ServiceControl.Audit/ServiceControl.Audit.csproj b/src/ServiceControl.Audit/ServiceControl.Audit.csproj index 1752bf81bd..b7394443c1 100644 --- a/src/ServiceControl.Audit/ServiceControl.Audit.csproj +++ b/src/ServiceControl.Audit/ServiceControl.Audit.csproj @@ -26,6 +26,7 @@ + diff --git a/src/ServiceControl.Audit/WebApplicationExtensions.cs b/src/ServiceControl.Audit/WebApplicationExtensions.cs index 76785dd77d..e8edece77f 100644 --- a/src/ServiceControl.Audit/WebApplicationExtensions.cs +++ b/src/ServiceControl.Audit/WebApplicationExtensions.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Audit; public static class WebApplicationExtensions { - public static void UseServiceControlAudit(this WebApplication app, ForwardedHeadersSettings forwardedHeadersSettings, HttpsSettings httpsSettings) + public static void UseServiceControlAudit(this WebApplication app, ForwardedHeadersSettings forwardedHeadersSettings, HttpsSettings httpsSettings, bool enableMcpServer) { app.UseServiceControlForwardedHeaders(forwardedHeadersSettings); app.UseServiceControlHttps(httpsSettings); @@ -17,5 +17,10 @@ public static void UseServiceControlAudit(this WebApplication app, ForwardedHead app.UseHttpLogging(); app.UseCors(); app.MapControllers(); + + if (enableMcpServer) + { + app.MapMcp("/mcp"); + } } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index 6873e229b3..5de2540e03 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -37,6 +37,7 @@ }, "NotificationsFilter": null, "AllowMessageEditing": false, + "EnableMcpServer": false, "EnableIntegratedServicePulse": false, "ServicePulseSettings": null, "MessageFilter": null, diff --git a/src/ServiceControl.UnitTests/Mcp/ArchiveMcpToolsTests.cs b/src/ServiceControl.UnitTests/Mcp/ArchiveMcpToolsTests.cs new file mode 100644 index 0000000000..84d98811d2 --- /dev/null +++ b/src/ServiceControl.UnitTests/Mcp/ArchiveMcpToolsTests.cs @@ -0,0 +1,154 @@ +#nullable enable + +namespace ServiceControl.UnitTests.Mcp; + +using System.Collections.Generic; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using NServiceBus.Testing; +using NUnit.Framework; +using ServiceControl.Mcp; +using ServiceControl.Persistence; +using ServiceControl.Persistence.Recoverability; +using ServiceControl.Recoverability; + +[TestFixture] +class ArchiveMcpToolsTests +{ + TestableMessageSession messageSession = null!; + StubArchiveMessages archiver = null!; + ArchiveTools tools = null!; + + [SetUp] + public void SetUp() + { + messageSession = new TestableMessageSession(); + archiver = new StubArchiveMessages(); + tools = new ArchiveTools(messageSession, archiver, NullLogger.Instance); + } + + [Test] + public async Task ArchiveFailedMessage_returns_accepted() + { + var result = await tools.ArchiveFailedMessage("msg-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(1)); + } + + [Test] + public async Task ArchiveFailedMessages_returns_accepted() + { + var result = await tools.ArchiveFailedMessages(["msg-1", "msg-2"]); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(2)); + } + + [Test] + public async Task ArchiveFailedMessages_rejects_empty_ids() + { + var result = await tools.ArchiveFailedMessages(["msg-1", ""]); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("non-empty")); + } + + [Test] + public async Task ArchiveFailureGroup_returns_accepted() + { + var result = await tools.ArchiveFailureGroup("group-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + } + + [Test] + public async Task ArchiveFailureGroup_returns_in_progress_when_already_running() + { + archiver.OperationInProgress = true; + + var result = await tools.ArchiveFailureGroup("group-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("InProgress")); + } + + [Test] + public async Task UnarchiveFailedMessage_returns_accepted() + { + var result = await tools.UnarchiveFailedMessage("msg-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(1)); + } + + [Test] + public async Task UnarchiveFailedMessages_returns_accepted() + { + var result = await tools.UnarchiveFailedMessages(["msg-1", "msg-2"]); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + } + + [Test] + public async Task UnarchiveFailedMessages_rejects_empty_ids() + { + var result = await tools.UnarchiveFailedMessages(["msg-1", ""]); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("non-empty")); + } + + [Test] + public async Task UnarchiveFailureGroup_returns_accepted() + { + var result = await tools.UnarchiveFailureGroup("group-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + } + + [Test] + public async Task UnarchiveFailureGroup_returns_in_progress_when_already_running() + { + archiver.OperationInProgress = true; + + var result = await tools.UnarchiveFailureGroup("group-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("InProgress")); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpStatusResponse + { + public string? Status { get; set; } + public string? Message { get; set; } + } + + class McpErrorResponse + { + public string? Error { get; set; } + } + + class StubArchiveMessages : IArchiveMessages + { + public bool OperationInProgress { get; set; } + + public bool IsOperationInProgressFor(string groupId, ArchiveType archiveType) => OperationInProgress; + public bool IsArchiveInProgressFor(string groupId) => OperationInProgress; + public Task StartArchiving(string groupId, ArchiveType archiveType) => Task.CompletedTask; + public Task StartUnarchiving(string groupId, ArchiveType archiveType) => Task.CompletedTask; + public Task ArchiveAllInGroup(string groupId) => Task.CompletedTask; + public Task UnarchiveAllInGroup(string groupId) => Task.CompletedTask; + public void DismissArchiveOperation(string groupId, ArchiveType archiveType) { } + public IEnumerable GetArchivalOperations() => []; + } +} diff --git a/src/ServiceControl.UnitTests/Mcp/FailedMessageMcpToolsTests.cs b/src/ServiceControl.UnitTests/Mcp/FailedMessageMcpToolsTests.cs new file mode 100644 index 0000000000..2a9c0b8c94 --- /dev/null +++ b/src/ServiceControl.UnitTests/Mcp/FailedMessageMcpToolsTests.cs @@ -0,0 +1,230 @@ +#nullable enable + +namespace ServiceControl.UnitTests.Mcp; + +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceControl.CompositeViews.Messages; +using ServiceControl.EventLog; +using ServiceControl.Infrastructure; +using ServiceControl.MessageFailures; +using ServiceControl.MessageFailures.Api; +using ServiceControl.Mcp; +using ServiceControl.Operations; +using ServiceControl.Persistence; +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.Recoverability; + +[TestFixture] +class FailedMessageMcpToolsTests +{ + StubErrorMessageDataStore store = null!; + FailedMessageTools tools = null!; + + [SetUp] + public void SetUp() + { + store = new StubErrorMessageDataStore(); + tools = new FailedMessageTools(store, NullLogger.Instance); + } + + [Test] + public async Task GetFailedMessages_returns_messages() + { + store.ErrorGetResult = new QueryResult>( + [new() { Id = "msg-1", MessageType = "MyNamespace.MyMessage", Status = FailedMessageStatus.Unresolved }], + new QueryStatsInfo("etag", 1, false)); + + var result = await tools.GetFailedMessages(); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response.TotalCount, Is.EqualTo(1)); + Assert.That(response.Results, Has.Count.EqualTo(1)); + } + + [Test] + public async Task GetFailedMessages_passes_paging_and_sort_parameters() + { + await tools.GetFailedMessages(page: 3, perPage: 10, sort: "time_sent", direction: "asc"); + + Assert.That(store.LastErrorGetArgs, Is.Not.Null); + Assert.That(store.LastErrorGetArgs!.Value.PagingInfo.Page, Is.EqualTo(3)); + Assert.That(store.LastErrorGetArgs!.Value.PagingInfo.PageSize, Is.EqualTo(10)); + Assert.That(store.LastErrorGetArgs!.Value.SortInfo.Sort, Is.EqualTo("time_sent")); + Assert.That(store.LastErrorGetArgs!.Value.SortInfo.Direction, Is.EqualTo("asc")); + } + + [Test] + public async Task GetFailedMessages_passes_filter_parameters() + { + await tools.GetFailedMessages(status: "unresolved", modified: "2026-01-01", queueAddress: "Sales"); + + Assert.That(store.LastErrorGetArgs!.Value.Status, Is.EqualTo("unresolved")); + Assert.That(store.LastErrorGetArgs!.Value.Modified, Is.EqualTo("2026-01-01")); + Assert.That(store.LastErrorGetArgs!.Value.QueueAddress, Is.EqualTo("Sales")); + } + + [Test] + public async Task GetFailedMessageById_returns_message() + { + store.ErrorByResult = new FailedMessage + { + Id = "msg-1", + UniqueMessageId = "unique-1", + Status = FailedMessageStatus.Unresolved + }; + + var result = await tools.GetFailedMessageById("msg-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.UniqueMessageId, Is.EqualTo("unique-1")); + } + + [Test] + public async Task GetFailedMessageById_returns_error_when_not_found() + { + store.ErrorByResult = null; + + var result = await tools.GetFailedMessageById("msg-missing"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("not found")); + } + + [Test] + public async Task GetFailedMessageLastAttempt_returns_view() + { + store.ErrorLastByResult = new FailedMessageView + { + Id = "msg-1", + MessageType = "MyMessage", + Status = FailedMessageStatus.Unresolved + }; + + var result = await tools.GetFailedMessageLastAttempt("msg-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.MessageType, Is.EqualTo("MyMessage")); + } + + [Test] + public async Task GetFailedMessageLastAttempt_returns_error_when_not_found() + { + store.ErrorLastByResult = null; + + var result = await tools.GetFailedMessageLastAttempt("msg-missing"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("not found")); + } + + [Test] + public async Task GetErrorsSummary_returns_summary() + { + store.ErrorsSummaryResult = new Dictionary + { + { "unresolved", 5 }, + { "archived", 3 } + }; + + var result = await tools.GetErrorsSummary(); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response, Contains.Key("unresolved")); + Assert.That(response, Contains.Key("archived")); + } + + [Test] + public async Task GetFailedMessagesByEndpoint_returns_messages() + { + store.ErrorsByEndpointResult = new QueryResult>( + [new() { Id = "msg-1", MessageType = "MyMessage" }], + new QueryStatsInfo("etag", 1, false)); + + var result = await tools.GetFailedMessagesByEndpoint("Sales"); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response.TotalCount, Is.EqualTo(1)); + Assert.That(store.LastErrorsByEndpointName, Is.EqualTo("Sales")); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpToolResponse + { + public int TotalCount { get; set; } + public List Results { get; set; } = []; + } + + class McpErrorResponse + { + public string? Error { get; set; } + } + + class StubErrorMessageDataStore : IErrorMessageDataStore + { + static readonly QueryResult> EmptyResult = new([], QueryStatsInfo.Zero); + + public QueryResult>? ErrorGetResult { get; set; } + public QueryResult>? ErrorsByEndpointResult { get; set; } + public FailedMessage? ErrorByResult { get; set; } + public FailedMessageView? ErrorLastByResult { get; set; } + public IDictionary? ErrorsSummaryResult { get; set; } + + public (string? Status, string? Modified, string? QueueAddress, PagingInfo PagingInfo, SortInfo SortInfo)? LastErrorGetArgs { get; private set; } + public string? LastErrorsByEndpointName { get; private set; } + + public Task>> ErrorGet(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) + { + LastErrorGetArgs = (status, modified, queueAddress, pagingInfo, sortInfo); + return Task.FromResult(ErrorGetResult ?? EmptyResult); + } + + public Task ErrorBy(string failedMessageId) => Task.FromResult(ErrorByResult)!; + + public Task ErrorLastBy(string failedMessageId) => Task.FromResult(ErrorLastByResult)!; + + public Task> ErrorsSummary() => Task.FromResult(ErrorsSummaryResult ?? new Dictionary()); + + public Task>> ErrorsByEndpointName(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) + { + LastErrorsByEndpointName = endpointName; + return Task.FromResult(ErrorsByEndpointResult ?? EmptyResult); + } + + // Unused interface members + public Task ErrorsHead(string status, string modified, string queueAddress) => throw new NotImplementedException(); + public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); + public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); + public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => throw new NotImplementedException(); + public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); + public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); + public Task FailedMessageMarkAsArchived(string failedMessageId) => throw new NotImplementedException(); + public Task FailedMessagesFetch(Guid[] ids) => throw new NotImplementedException(); + public Task StoreFailedErrorImport(FailedErrorImport failure) => throw new NotImplementedException(); + public Task CreateEditFailedMessageManager() => throw new NotImplementedException(); + public Task> GetFailureGroupView(string groupId, string status, string modified) => throw new NotImplementedException(); + public Task> GetFailureGroupsByClassifier(string classifier) => throw new NotImplementedException(); + public Task EditComment(string groupId, string comment) => throw new NotImplementedException(); + public Task DeleteComment(string groupId) => throw new NotImplementedException(); + public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => throw new NotImplementedException(); + public Task GetGroupErrorsCount(string groupId, string status, string modified) => throw new NotImplementedException(); + public Task>> GetGroup(string groupId, string status, string modified) => throw new NotImplementedException(); + public Task MarkMessageAsResolved(string failedMessageId) => throw new NotImplementedException(); + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); + public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => throw new NotImplementedException(); + public Task UnArchiveMessages(IEnumerable failedMessageIds) => throw new NotImplementedException(); + public Task RevertRetry(string messageUniqueId) => throw new NotImplementedException(); + public Task RemoveFailedMessageRetryDocument(string uniqueMessageId) => throw new NotImplementedException(); + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => throw new NotImplementedException(); + public Task FetchFromFailedMessage(string uniqueMessageId) => throw new NotImplementedException(); + public Task StoreEventLogItem(EventLogItem logItem) => throw new NotImplementedException(); + public Task StoreFailedMessagesForTestsOnly(params FailedMessage[] failedMessages) => throw new NotImplementedException(); + public Task CreateNotificationsManager() => throw new NotImplementedException(); + } +} diff --git a/src/ServiceControl.UnitTests/Mcp/FailureGroupMcpToolsTests.cs b/src/ServiceControl.UnitTests/Mcp/FailureGroupMcpToolsTests.cs new file mode 100644 index 0000000000..5745ee095a --- /dev/null +++ b/src/ServiceControl.UnitTests/Mcp/FailureGroupMcpToolsTests.cs @@ -0,0 +1,108 @@ +#nullable enable + +namespace ServiceControl.UnitTests.Mcp; + +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceControl.Mcp; +using ServiceControl.Persistence; +using ServiceControl.Persistence.Recoverability; +using ServiceControl.Recoverability; +using ServiceControl.UnitTests.Operations; + +[TestFixture] +class FailureGroupMcpToolsTests +{ + StubGroupsDataStore groupsStore = null!; + StubRetryHistoryDataStore retryStore = null!; + FailureGroupTools tools = null!; + + [SetUp] + public void SetUp() + { + groupsStore = new StubGroupsDataStore(); + retryStore = new StubRetryHistoryDataStore(); + var domainEvents = new FakeDomainEvents(); + var retryingManager = new RetryingManager(domainEvents, NullLogger.Instance); + var archiver = new StubArchiveMessages(); + var fetcher = new GroupFetcher(groupsStore, retryStore, retryingManager, archiver); + tools = new FailureGroupTools(fetcher, retryStore, NullLogger.Instance); + } + + [Test] + public async Task GetFailureGroups_returns_groups() + { + groupsStore.FailureGroups = + [ + new FailureGroupView { Id = "group-1", Title = "NullReferenceException", Type = "Exception Type and Stack Trace", Count = 5, First = DateTime.UtcNow.AddHours(-1), Last = DateTime.UtcNow } + ]; + + var result = await tools.GetFailureGroups(); + var response = JsonSerializer.Deserialize>(result, JsonOptions)!; + + Assert.That(response, Has.Count.EqualTo(1)); + Assert.That(response[0].Id, Is.EqualTo("group-1")); + Assert.That(response[0].Count, Is.EqualTo(5)); + } + + [Test] + public async Task GetFailureGroups_passes_classifier() + { + await tools.GetFailureGroups(classifier: "Message Type"); + + Assert.That(groupsStore.LastClassifier, Is.EqualTo("Message Type")); + } + + [Test] + public async Task GetRetryHistory_returns_history() + { + retryStore.RetryHistoryResult = RetryHistory.CreateNew(); + + var result = await tools.GetRetryHistory(); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.HistoricOperations, Is.Empty); + Assert.That(response.UnacknowledgedOperations, Is.Empty); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class StubGroupsDataStore : IGroupsDataStore + { + public IList FailureGroups { get; set; } = []; + public string? LastClassifier { get; private set; } + + public Task> GetFailureGroupsByClassifier(string classifier, string classifierFilter) + { + LastClassifier = classifier; + return Task.FromResult(FailureGroups); + } + + public Task GetCurrentForwardingBatch() => Task.FromResult(null!); + } + + class StubRetryHistoryDataStore : IRetryHistoryDataStore + { + public RetryHistory? RetryHistoryResult { get; set; } + + public Task GetRetryHistory() => Task.FromResult(RetryHistoryResult ?? RetryHistory.CreateNew()); + public Task AcknowledgeRetryGroup(string groupId) => Task.FromResult(true); + public Task RecordRetryOperationCompleted(string requestId, RetryType retryType, DateTime startTime, DateTime completionTime, string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth) => Task.CompletedTask; + } + + class StubArchiveMessages : IArchiveMessages + { + public bool IsOperationInProgressFor(string groupId, ArchiveType archiveType) => false; + public bool IsArchiveInProgressFor(string groupId) => false; + public Task StartArchiving(string groupId, ArchiveType archiveType) => Task.CompletedTask; + public Task StartUnarchiving(string groupId, ArchiveType archiveType) => Task.CompletedTask; + public Task ArchiveAllInGroup(string groupId) => Task.CompletedTask; + public Task UnarchiveAllInGroup(string groupId) => Task.CompletedTask; + public void DismissArchiveOperation(string groupId, ArchiveType archiveType) { } + public IEnumerable GetArchivalOperations() => []; + } +} diff --git a/src/ServiceControl.UnitTests/Mcp/RetryMcpToolsTests.cs b/src/ServiceControl.UnitTests/Mcp/RetryMcpToolsTests.cs new file mode 100644 index 0000000000..62ed5eaa7d --- /dev/null +++ b/src/ServiceControl.UnitTests/Mcp/RetryMcpToolsTests.cs @@ -0,0 +1,121 @@ +#nullable enable + +namespace ServiceControl.UnitTests.Mcp; + +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using NServiceBus.Testing; +using NUnit.Framework; +using ServiceControl.Mcp; +using ServiceControl.Persistence; +using ServiceControl.Recoverability; +using ServiceControl.UnitTests.Operations; + +[TestFixture] +class RetryMcpToolsTests +{ + TestableMessageSession messageSession = null!; + RetryingManager retryingManager = null!; + RetryTools tools = null!; + + [SetUp] + public void SetUp() + { + messageSession = new TestableMessageSession(); + retryingManager = new RetryingManager(new FakeDomainEvents(), NullLogger.Instance); + tools = new RetryTools(messageSession, retryingManager, NullLogger.Instance); + } + + [Test] + public async Task RetryFailedMessage_returns_accepted() + { + var result = await tools.RetryFailedMessage("msg-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(1)); + } + + [Test] + public async Task RetryFailedMessages_returns_accepted() + { + var result = await tools.RetryFailedMessages(["msg-1", "msg-2"]); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(1)); + } + + [Test] + public async Task RetryFailedMessages_rejects_empty_ids() + { + var result = await tools.RetryFailedMessages(["msg-1", ""]); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Error, Does.Contain("non-empty")); + } + + [Test] + public async Task RetryFailedMessagesByQueue_returns_accepted() + { + var result = await tools.RetryFailedMessagesByQueue("Sales@machine"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(1)); + } + + [Test] + public async Task RetryAllFailedMessages_returns_accepted() + { + var result = await tools.RetryAllFailedMessages(); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + Assert.That(messageSession.SentMessages, Has.Length.EqualTo(1)); + } + + [Test] + public async Task RetryAllFailedMessagesByEndpoint_returns_accepted() + { + var result = await tools.RetryAllFailedMessagesByEndpoint("Sales"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + } + + [Test] + public async Task RetryFailureGroup_returns_accepted() + { + var result = await tools.RetryFailureGroup("group-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("Accepted")); + } + + [Test] + public async Task RetryFailureGroup_returns_in_progress_when_already_running() + { + await retryingManager.Wait("group-1", RetryType.FailureGroup, System.DateTime.UtcNow); + await retryingManager.Preparing("group-1", RetryType.FailureGroup, 1); + + var result = await tools.RetryFailureGroup("group-1"); + var response = JsonSerializer.Deserialize(result, JsonOptions)!; + + Assert.That(response.Status, Is.EqualTo("InProgress")); + } + + static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + + class McpStatusResponse + { + public string? Status { get; set; } + public string? Message { get; set; } + } + + class McpErrorResponse + { + public string? Error { get; set; } + } +} diff --git a/src/ServiceControl/App.config b/src/ServiceControl/App.config index d6271805e5..698755c9a7 100644 --- a/src/ServiceControl/App.config +++ b/src/ServiceControl/App.config @@ -5,6 +5,8 @@ These settings are only here so that we can debug ServiceControl while developin --> + + diff --git a/src/ServiceControl/Hosting/Commands/ImportFailedErrorsCommand.cs b/src/ServiceControl/Hosting/Commands/ImportFailedErrorsCommand.cs index 105f756daf..932e301047 100644 --- a/src/ServiceControl/Hosting/Commands/ImportFailedErrorsCommand.cs +++ b/src/ServiceControl/Hosting/Commands/ImportFailedErrorsCommand.cs @@ -26,7 +26,7 @@ public override async Task Execute(HostArguments args, Settings settings) var hostBuilder = Host.CreateApplicationBuilder(); hostBuilder.AddServiceControl(settings, endpointConfiguration); - hostBuilder.AddServiceControlApi(settings.CorsSettings); + hostBuilder.AddServiceControlApi(settings); using var app = hostBuilder.Build(); await app.StartAsync(); diff --git a/src/ServiceControl/Hosting/Commands/RunCommand.cs b/src/ServiceControl/Hosting/Commands/RunCommand.cs index ebc08958cf..9778db2cc0 100644 --- a/src/ServiceControl/Hosting/Commands/RunCommand.cs +++ b/src/ServiceControl/Hosting/Commands/RunCommand.cs @@ -27,10 +27,10 @@ public override async Task Execute(HostArguments args, Settings settings) hostBuilder.AddServiceControlAuthentication(settings.OpenIdConnectSettings); hostBuilder.AddServiceControlHttps(settings.HttpsSettings); hostBuilder.AddServiceControl(settings, endpointConfiguration); - hostBuilder.AddServiceControlApi(settings.CorsSettings); + hostBuilder.AddServiceControlApi(settings); var app = hostBuilder.Build(); - app.UseServiceControl(settings.ForwardedHeadersSettings, settings.HttpsSettings); + app.UseServiceControl(settings.ForwardedHeadersSettings, settings.HttpsSettings, settings.EnableMcpServer); if (settings.EnableIntegratedServicePulse) { app.UseServicePulse(settings.ServicePulseSettings); diff --git a/src/ServiceControl/Infrastructure/Settings/Settings.cs b/src/ServiceControl/Infrastructure/Settings/Settings.cs index d71b9dca66..24e7082863 100644 --- a/src/ServiceControl/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl/Infrastructure/Settings/Settings.cs @@ -81,6 +81,7 @@ public Settings( DisableExternalIntegrationsPublishing = SettingsReader.Read(SettingsRootNamespace, "DisableExternalIntegrationsPublishing", false); TrackInstancesInitialValue = SettingsReader.Read(SettingsRootNamespace, "TrackInstancesInitialValue", true); ShutdownTimeout = SettingsReader.Read(SettingsRootNamespace, "ShutdownTimeout", ShutdownTimeout); + EnableMcpServer = SettingsReader.Read(SettingsRootNamespace, "EnableMcpServer", false); AssemblyLoadContextResolver = static assemblyPath => new PluginAssemblyLoadContext(assemblyPath); } @@ -113,6 +114,8 @@ public Settings( public bool AllowMessageEditing { get; set; } + public bool EnableMcpServer { get; set; } + public bool EnableIntegratedServicePulse { get; set; } public ServicePulseSettings ServicePulseSettings { get; set; } diff --git a/src/ServiceControl/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs b/src/ServiceControl/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs index 298885ae0f..17dc44d5d3 100644 --- a/src/ServiceControl/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs +++ b/src/ServiceControl/Infrastructure/WebApi/HostApplicationBuilderExtensions.cs @@ -9,10 +9,11 @@ using Microsoft.Extensions.Hosting; using Particular.LicensingComponent.WebApi; using Particular.ServiceControl; + using ServiceBus.Management.Infrastructure.Settings; static class HostApplicationBuilderExtensions { - public static void AddServiceControlApi(this IHostApplicationBuilder builder, CorsSettings corsSettings) + public static void AddServiceControlApi(this IHostApplicationBuilder builder, Settings settings) { // This registers concrete classes that implement IApi. Currently it is hard to find out to what // component those APIs should belong to so we leave it here for now. @@ -20,7 +21,15 @@ public static void AddServiceControlApi(this IHostApplicationBuilder builder, Co builder.AddServiceControlApis(); - builder.Services.AddCors(options => options.AddDefaultPolicy(Cors.GetDefaultPolicy(corsSettings))); + if (settings.EnableMcpServer) + { + builder.Services + .AddMcpServer() + .WithHttpTransport() + .WithToolsFromAssembly(); + } + + builder.Services.AddCors(options => options.AddDefaultPolicy(Cors.GetDefaultPolicy(settings.CorsSettings))); // We're not explicitly adding Gzip here because it's already in the default list of supported compressors builder.Services.AddResponseCompression(); diff --git a/src/ServiceControl/Mcp/ArchiveTools.cs b/src/ServiceControl/Mcp/ArchiveTools.cs new file mode 100644 index 0000000000..2312145b8f --- /dev/null +++ b/src/ServiceControl/Mcp/ArchiveTools.cs @@ -0,0 +1,146 @@ +namespace ServiceControl.Mcp; + +using System.ComponentModel; +using System.Linq; +using System.Text.Json; +using System.Threading.Tasks; +using MessageFailures.InternalMessages; +using Microsoft.Extensions.Logging; +using ModelContextProtocol.Server; +using NServiceBus; +using Persistence.Recoverability; +using ServiceControl.Recoverability; + +[McpServerToolType, Description( + "Tools for archiving and unarchiving failed messages.\n\n" + + "Agent guidance:\n" + + "1. Archiving dismisses a failed message — it moves out of the unresolved list and no longer counts as an active problem.\n" + + "2. Unarchiving restores a previously archived message back to the unresolved list so it can be retried.\n" + + "3. Prefer ArchiveFailureGroup or UnarchiveFailureGroup when acting on an entire failure group — it is more efficient than archiving messages individually.\n" + + "4. Use ArchiveFailedMessages or UnarchiveFailedMessages when you have a specific set of message IDs.\n" + + "5. All operations are asynchronous — they return Accepted immediately and complete in the background." +)] +public class ArchiveTools(IMessageSession messageSession, IArchiveMessages archiver, ILogger logger) +{ + [McpServerTool, Description( + "Use this tool to dismiss a single failed message that does not need to be retried. " + + "Good for questions like: 'archive this message', 'dismiss this failure', or 'I do not need to retry this one'. " + + "Archiving moves the message out of the unresolved list so it no longer shows up as an active problem. " + + "This is an asynchronous operation — the message will be archived shortly after the request is accepted. " + + "If you need to archive many messages with the same root cause, use ArchiveFailureGroup instead." + )] + public async Task ArchiveFailedMessage( + [Description("The unique message ID from a previous query result")] string failedMessageId) + { + logger.LogInformation("MCP ArchiveFailedMessage invoked (failedMessageId={FailedMessageId})", failedMessageId); + + await messageSession.SendLocal(m => m.FailedMessageId = failedMessageId); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Archive requested for message '{failedMessageId}'." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to dismiss multiple failed messages at once that do not need to be retried. " + + "Good for questions like: 'archive these messages', 'dismiss these failures', or 'archive messages msg-1, msg-2, msg-3'. " + + "Prefer ArchiveFailureGroup when all messages share the same failure cause — use this tool when you have a specific set of message IDs to archive." + )] + public async Task ArchiveFailedMessages( + [Description("The unique message IDs from a previous query result")] string[] messageIds) + { + logger.LogInformation("MCP ArchiveFailedMessages invoked (count={Count})", messageIds.Length); + + if (messageIds.Any(string.IsNullOrEmpty)) + { + logger.LogWarning("MCP ArchiveFailedMessages: rejected due to empty message IDs"); + return JsonSerializer.Serialize(new { Error = "All message IDs must be non-empty strings." }, McpJsonOptions.Default); + } + + foreach (var id in messageIds) + { + await messageSession.SendLocal(m => m.FailedMessageId = id); + } + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Archive requested for {messageIds.Length} messages." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to dismiss an entire failure group — all messages that failed with the same exception type and stack trace. " + + "Good for questions like: 'archive this failure group', 'dismiss all NullReferenceException failures', or 'archive the whole group'. " + + "This is the most efficient way to archive many related failures at once. " + + "You need a group ID, which you can get from GetFailureGroups. " + + "Returns InProgress if an archive operation is already running for this group." + )] + public async Task ArchiveFailureGroup( + [Description("The failure group ID from get_failure_groups results")] string groupId) + { + logger.LogInformation("MCP ArchiveFailureGroup invoked (groupId={GroupId})", groupId); + + if (archiver.IsOperationInProgressFor(groupId, ArchiveType.FailureGroup)) + { + logger.LogInformation("MCP ArchiveFailureGroup: operation already in progress for group '{GroupId}'", groupId); + return JsonSerializer.Serialize(new { Status = "InProgress", Message = $"An archive operation is already in progress for group '{groupId}'." }, McpJsonOptions.Default); + } + + await archiver.StartArchiving(groupId, ArchiveType.FailureGroup); + await messageSession.SendLocal(m => m.GroupId = groupId); + + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Archive requested for all messages in failure group '{groupId}'." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to restore a previously archived failed message back to the unresolved list so it can be retried. " + + "Good for questions like: 'unarchive this message', 'restore this failure', or 'I need to retry this archived message'. " + + "Use when a message was archived by mistake or when the underlying issue has been fixed and the message should be reprocessed. " + + "If you need to restore many messages from the same failure group, use UnarchiveFailureGroup instead." + )] + public async Task UnarchiveFailedMessage( + [Description("The unique message ID to restore")] string failedMessageId) + { + logger.LogInformation("MCP UnarchiveFailedMessage invoked (failedMessageId={FailedMessageId})", failedMessageId); + + await messageSession.SendLocal(m => m.FailedMessageIds = [failedMessageId]); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Unarchive requested for message '{failedMessageId}'." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to restore multiple previously archived failed messages back to the unresolved list. " + + "Good for questions like: 'unarchive these messages', 'restore these failures', or 'unarchive messages msg-1, msg-2, msg-3'. " + + "Prefer UnarchiveFailureGroup when restoring an entire group — use this tool when you have a specific set of message IDs." + )] + public async Task UnarchiveFailedMessages( + [Description("The unique message IDs to restore")] string[] messageIds) + { + logger.LogInformation("MCP UnarchiveFailedMessages invoked (count={Count})", messageIds.Length); + + if (messageIds.Any(string.IsNullOrEmpty)) + { + logger.LogWarning("MCP UnarchiveFailedMessages: rejected due to empty message IDs"); + return JsonSerializer.Serialize(new { Error = "All message IDs must be non-empty strings." }, McpJsonOptions.Default); + } + + await messageSession.SendLocal(m => m.FailedMessageIds = messageIds); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Unarchive requested for {messageIds.Length} messages." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to restore an entire archived failure group back to the unresolved list. " + + "Good for questions like: 'unarchive this failure group', 'restore all archived NullReferenceException failures', or 'unarchive the whole group'. " + + "All messages that were archived together under this group will become available for retry again. " + + "You need a group ID, which you can get from GetFailureGroups. " + + "Returns InProgress if an unarchive operation is already running for this group." + )] + public async Task UnarchiveFailureGroup( + [Description("The failure group ID from get_failure_groups results")] string groupId) + { + logger.LogInformation("MCP UnarchiveFailureGroup invoked (groupId={GroupId})", groupId); + + if (archiver.IsOperationInProgressFor(groupId, ArchiveType.FailureGroup)) + { + logger.LogInformation("MCP UnarchiveFailureGroup: operation already in progress for group '{GroupId}'", groupId); + return JsonSerializer.Serialize(new { Status = "InProgress", Message = $"An archive operation is already in progress for group '{groupId}'." }, McpJsonOptions.Default); + } + + await archiver.StartUnarchiving(groupId, ArchiveType.FailureGroup); + await messageSession.SendLocal(m => m.GroupId = groupId); + + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Unarchive requested for all messages in failure group '{groupId}'." }, McpJsonOptions.Default); + } +} diff --git a/src/ServiceControl/Mcp/FailedMessageTools.cs b/src/ServiceControl/Mcp/FailedMessageTools.cs new file mode 100644 index 0000000000..57c30fe11d --- /dev/null +++ b/src/ServiceControl/Mcp/FailedMessageTools.cs @@ -0,0 +1,146 @@ +#nullable enable + +namespace ServiceControl.Mcp; + +using System.ComponentModel; +using System.Text.Json; +using System.Threading.Tasks; +using MessageFailures.Api; +using Microsoft.Extensions.Logging; +using ModelContextProtocol.Server; +using Persistence; +using Persistence.Infrastructure; + +[McpServerToolType, Description( + "Tools for investigating failed messages.\n\n" + + "Agent guidance:\n" + + "1. Start with GetErrorsSummary to get a quick health check of failure counts by status.\n" + + "2. Use GetFailureGroups (from FailureGroupTools) to see failures grouped by root cause before drilling into individual messages.\n" + + "3. Use GetFailedMessages for broad listing, or GetFailedMessagesByEndpoint when you already know the endpoint.\n" + + "4. Use GetFailedMessageById for full details including all processing attempts, or GetFailedMessageLastAttempt for just the most recent failure.\n" + + "5. Keep page=1 unless the user asks for more results.\n" + + "6. Only change sorting when the user explicitly asks for it." +)] +public class FailedMessageTools(IErrorMessageDataStore store, ILogger logger) +{ + [McpServerTool, Description( + "Use this tool to browse failed messages when the user wants to see what is failing. " + + "Good for questions like: 'what messages are currently failing?', 'are there failures in a specific queue?', or 'what failed recently?'. " + + "Returns a paged list of failed messages with their status, exception details, and queue information. " + + "For broad requests, call with no parameters to get the most recent failures — only add filters when you need to narrow down results. " + + "Prefer GetFailedMessagesByEndpoint when the user mentions a specific endpoint." + )] + public async Task GetFailedMessages( + [Description("Narrow results to a specific status: unresolved (still failing), resolved (succeeded on retry), archived (dismissed), or retryissued (retry in progress). Omit to include all statuses.")] string? status = null, + [Description("Only return messages modified after this date (ISO 8601). Useful for checking recent failures.")] string? modified = null, + [Description("Only return messages from this queue address, e.g. 'Sales@machine'. Use when investigating a specific queue.")] string? queueAddress = null, + [Description("Page number, 1-based")] int page = 1, + [Description("Results per page")] int perPage = 50, + [Description("Sort by: time_sent, message_type, or time_of_failure")] string sort = "time_of_failure", + [Description("Sort direction: asc or desc")] string direction = "desc") + { + logger.LogInformation("MCP GetFailedMessages invoked (status={Status}, queueAddress={QueueAddress}, page={Page})", status, queueAddress, page); + + var pagingInfo = new PagingInfo(page, perPage); + var sortInfo = new SortInfo(sort, direction); + + var results = await store.ErrorGet(status, modified, queueAddress, pagingInfo, sortInfo); + + logger.LogInformation("MCP GetFailedMessages returned {Count} results", results.QueryStats.TotalCount); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to get the full details of a specific failed message, including all processing attempts and exception information. " + + "Good for questions like: 'show me details for this failed message', 'what exception caused this failure?', or 'how many times has this message failed?'. " + + "You need the message's unique ID, which you can get from GetFailedMessages or GetFailureGroups results. " + + "If you only need the most recent failure attempt, use GetFailedMessageLastAttempt instead — it returns less data." + )] + public async Task GetFailedMessageById( + [Description("The unique message ID from a previous query result")] string failedMessageId) + { + logger.LogInformation("MCP GetFailedMessageById invoked (failedMessageId={FailedMessageId})", failedMessageId); + + var result = await store.ErrorBy(failedMessageId); + + if (result == null) + { + logger.LogWarning("MCP GetFailedMessageById: message '{FailedMessageId}' not found", failedMessageId); + return JsonSerializer.Serialize(new { Error = $"Failed message '{failedMessageId}' not found." }, McpJsonOptions.Default); + } + + return JsonSerializer.Serialize(result, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to see how a specific message failed most recently. " + + "Good for questions like: 'what was the last error for this message?', 'show me the latest exception', or 'what happened on the last attempt?'. " + + "Returns the latest processing attempt with its exception, stack trace, and headers. " + + "Lighter than GetFailedMessageById when you only care about the most recent failure rather than the full history." + )] + public async Task GetFailedMessageLastAttempt( + [Description("The unique message ID from a previous query result")] string failedMessageId) + { + logger.LogInformation("MCP GetFailedMessageLastAttempt invoked (failedMessageId={FailedMessageId})", failedMessageId); + + var result = await store.ErrorLastBy(failedMessageId); + + if (result == null) + { + logger.LogWarning("MCP GetFailedMessageLastAttempt: message '{FailedMessageId}' not found", failedMessageId); + return JsonSerializer.Serialize(new { Error = $"Failed message '{failedMessageId}' not found." }, McpJsonOptions.Default); + } + + return JsonSerializer.Serialize(result, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool as a quick health check to see how many messages are in each failure state. " + + "Good for questions like: 'how many errors are there?', 'what is the error situation?', or 'are there unresolved failures?'. " + + "Returns counts for unresolved, archived, resolved, and retryissued statuses. " + + "This is a good first tool to call when asked about the overall error situation before drilling into specific messages." + )] + public async Task GetErrorsSummary() + { + logger.LogInformation("MCP GetErrorsSummary invoked"); + + var result = await store.ErrorsSummary(); + return JsonSerializer.Serialize(result, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to see failed messages for a specific NServiceBus endpoint. " + + "Good for questions like: 'what is failing in the Sales endpoint?', 'show errors for Shipping', or 'are there failures in this endpoint?'. " + + "Returns the same paged failure data as GetFailedMessages but scoped to one endpoint. " + + "Prefer this tool over GetFailedMessages when the user mentions a specific endpoint name." + )] + public async Task GetFailedMessagesByEndpoint( + [Description("The NServiceBus endpoint name, e.g. 'Sales' or 'Shipping.MessageHandler'")] string endpointName, + [Description("Narrow results to a specific status: unresolved, resolved, archived, or retryissued. Omit to include all.")] string? status = null, + [Description("Only return messages modified after this date (ISO 8601)")] string? modified = null, + [Description("Page number, 1-based")] int page = 1, + [Description("Results per page")] int perPage = 50, + [Description("Sort by: time_sent, message_type, or time_of_failure")] string sort = "time_of_failure", + [Description("Sort direction: asc or desc")] string direction = "desc") + { + logger.LogInformation("MCP GetFailedMessagesByEndpoint invoked (endpoint={EndpointName}, status={Status}, page={Page})", endpointName, status, page); + + var pagingInfo = new PagingInfo(page, perPage); + var sortInfo = new SortInfo(sort, direction); + + var results = await store.ErrorsByEndpointName(status, endpointName, modified, pagingInfo, sortInfo); + + logger.LogInformation("MCP GetFailedMessagesByEndpoint returned {Count} results for endpoint '{EndpointName}'", results.QueryStats.TotalCount, endpointName); + + return JsonSerializer.Serialize(new + { + results.QueryStats.TotalCount, + results.Results + }, McpJsonOptions.Default); + } +} diff --git a/src/ServiceControl/Mcp/FailureGroupTools.cs b/src/ServiceControl/Mcp/FailureGroupTools.cs new file mode 100644 index 0000000000..4fce32514f --- /dev/null +++ b/src/ServiceControl/Mcp/FailureGroupTools.cs @@ -0,0 +1,55 @@ +#nullable enable + +namespace ServiceControl.Mcp; + +using System.ComponentModel; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using ModelContextProtocol.Server; +using Persistence; +using Recoverability; + +[McpServerToolType, Description( + "Tools for inspecting failure groups and retry history.\n\n" + + "Agent guidance:\n" + + "1. GetFailureGroups is usually the best starting point for diagnosing production issues — call it before drilling into individual messages.\n" + + "2. Call GetFailureGroups with no parameters to use the default grouping by exception type and stack trace.\n" + + "3. Use GetRetryHistory to check whether someone has already retried a group before retrying it again." +)] +public class FailureGroupTools(GroupFetcher fetcher, IRetryHistoryDataStore retryStore, ILogger logger) +{ + [McpServerTool, Description( + "Use this tool to understand why messages are failing by seeing failures grouped by root cause. " + + "Good for questions like: 'why are messages failing?', 'what errors are happening?', 'group failures by exception', or 'what are the top failure causes?'. " + + "Each group represents a distinct exception type and stack trace, showing how many messages are affected and when failures started and last occurred. " + + "This is usually the best starting point for diagnosing production issues — call it before drilling into individual messages. " + + "Call with no parameters to use the default grouping by exception type and stack trace." + )] + public async Task GetFailureGroups( + [Description("How to group failures. The default 'Exception Type and Stack Trace' is almost always what you want. Use 'Message Type' to group by the NServiceBus message type instead.")] string classifier = "Exception Type and Stack Trace", + [Description("Only include groups matching this filter text")] string? classifierFilter = null) + { + logger.LogInformation("MCP GetFailureGroups invoked (classifier={Classifier})", classifier); + + var results = await fetcher.GetGroups(classifier, classifierFilter); + + logger.LogInformation("MCP GetFailureGroups returned {Count} groups", results.Length); + + return JsonSerializer.Serialize(results, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to check the history of retry operations. " + + "Good for questions like: 'has someone already retried these?', 'what happened the last time we retried this group?', 'show retry history', or 'were any retries attempted today?'. " + + "Returns which groups were retried, when, and whether the retries succeeded or failed. " + + "Use this before retrying a group to avoid duplicate retry attempts." + )] + public async Task GetRetryHistory() + { + logger.LogInformation("MCP GetRetryHistory invoked"); + + var retryHistory = await retryStore.GetRetryHistory(); + return JsonSerializer.Serialize(retryHistory, McpJsonOptions.Default); + } +} diff --git a/src/ServiceControl/Mcp/McpJsonOptions.cs b/src/ServiceControl/Mcp/McpJsonOptions.cs new file mode 100644 index 0000000000..1e37e52d37 --- /dev/null +++ b/src/ServiceControl/Mcp/McpJsonOptions.cs @@ -0,0 +1,14 @@ +namespace ServiceControl.Mcp; + +using System.Text.Json; +using System.Text.Json.Serialization; + +static class McpJsonOptions +{ + public static JsonSerializerOptions Default { get; } = new() + { + PropertyNamingPolicy = JsonNamingPolicy.CamelCase, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + WriteIndented = false + }; +} diff --git a/src/ServiceControl/Mcp/RetryTools.cs b/src/ServiceControl/Mcp/RetryTools.cs new file mode 100644 index 0000000000..6edd34a9d4 --- /dev/null +++ b/src/ServiceControl/Mcp/RetryTools.cs @@ -0,0 +1,135 @@ +namespace ServiceControl.Mcp; + +using System.ComponentModel; +using System.Linq; +using System.Text.Json; +using System.Threading.Tasks; +using MessageFailures; +using MessageFailures.InternalMessages; +using Microsoft.Extensions.Logging; +using ModelContextProtocol.Server; +using NServiceBus; +using Recoverability; +using Persistence; + +[McpServerToolType, Description( + "Tools for retrying failed messages.\n\n" + + "Agent guidance:\n" + + "1. Retrying sends a failed message back to its original queue for reprocessing. Only retry after the underlying issue has been resolved.\n" + + "2. Prefer RetryFailureGroup when all messages share the same root cause — it is the most targeted approach.\n" + + "3. Use RetryAllFailedMessagesByEndpoint when a bug in one endpoint has been fixed.\n" + + "4. Use RetryFailedMessagesByQueue when a queue's consumer was down and is now back.\n" + + "5. Use RetryAllFailedMessages only as a last resort — it retries everything.\n" + + "6. All operations are asynchronous — they return Accepted immediately and complete in the background." +)] +public class RetryTools(IMessageSession messageSession, RetryingManager retryingManager, ILogger logger) +{ + [McpServerTool, Description( + "Use this tool to reprocess a single failed message by sending it back to its original queue. " + + "Good for questions like: 'retry this message', 'reprocess this failure', or 'send this message back for processing'. " + + "The message will go through normal processing again. Only use after the underlying issue (bug fix, infrastructure problem) has been resolved. " + + "If you need to retry many messages with the same root cause, use RetryFailureGroup instead." + )] + public async Task RetryFailedMessage( + [Description("The unique message ID from a previous query result")] string failedMessageId) + { + logger.LogInformation("MCP RetryFailedMessage invoked (failedMessageId={FailedMessageId})", failedMessageId); + + await messageSession.SendLocal(m => m.FailedMessageId = failedMessageId); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Retry requested for message '{failedMessageId}'." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to reprocess multiple specific failed messages at once. " + + "Good for questions like: 'retry these messages', 'reprocess messages msg-1, msg-2, msg-3', or 'retry this batch'. " + + "Prefer RetryFailureGroup when all messages share the same failure cause — use this tool when you have a specific set of message IDs to retry." + )] + public async Task RetryFailedMessages( + [Description("The unique message IDs from a previous query result")] string[] messageIds) + { + logger.LogInformation("MCP RetryFailedMessages invoked (count={Count})", messageIds.Length); + + if (messageIds.Any(string.IsNullOrEmpty)) + { + logger.LogWarning("MCP RetryFailedMessages: rejected due to empty message IDs"); + return JsonSerializer.Serialize(new { Error = "All message IDs must be non-empty strings." }, McpJsonOptions.Default); + } + + await messageSession.SendLocal(m => m.MessageUniqueIds = messageIds); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Retry requested for {messageIds.Length} messages." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to retry all unresolved failed messages from a specific queue. " + + "Good for questions like: 'retry all failures in the Sales queue', 'reprocess everything from this queue', or 'the queue consumer is back, retry its failures'. " + + "Useful when a queue's consumer was down or misconfigured and is now fixed. Only retries messages with unresolved status." + )] + public async Task RetryFailedMessagesByQueue( + [Description("The full queue address including machine name, e.g. 'Sales@machine'")] string queueAddress) + { + logger.LogInformation("MCP RetryFailedMessagesByQueue invoked (queueAddress={QueueAddress})", queueAddress); + + await messageSession.SendLocal(m => + { + m.QueueAddress = queueAddress; + m.Status = FailedMessageStatus.Unresolved; + }); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Retry requested for all failed messages in queue '{queueAddress}'." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to retry every unresolved failed message across all queues and endpoints. " + + "Good for questions like: 'retry everything', 'reprocess all failures', or 'retry all failed messages'. " + + "This is a broad operation — prefer RetryFailedMessagesByQueue, RetryAllFailedMessagesByEndpoint, or RetryFailureGroup when you can scope the retry more narrowly." + )] + public async Task RetryAllFailedMessages() + { + logger.LogInformation("MCP RetryAllFailedMessages invoked"); + + await messageSession.SendLocal(new RequestRetryAll()); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = "Retry requested for all failed messages." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to retry all failed messages for a specific NServiceBus endpoint. " + + "Good for questions like: 'retry all failures in the Sales endpoint', 'the bug in Shipping is fixed, retry its failures', or 'reprocess all errors for this endpoint'. " + + "Useful when a bug in one endpoint has been fixed and all its failures should be reprocessed." + )] + public async Task RetryAllFailedMessagesByEndpoint( + [Description("The NServiceBus endpoint name, e.g. 'Sales' or 'Shipping.MessageHandler'")] string endpointName) + { + logger.LogInformation("MCP RetryAllFailedMessagesByEndpoint invoked (endpoint={EndpointName})", endpointName); + + await messageSession.SendLocal(new RequestRetryAll { Endpoint = endpointName }); + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Retry requested for all failed messages in endpoint '{endpointName}'." }, McpJsonOptions.Default); + } + + [McpServerTool, Description( + "Use this tool to retry all failed messages that share the same exception type and stack trace. " + + "Good for questions like: 'retry this failure group', 'the bug causing these NullReferenceExceptions is fixed, retry them', or 'retry all messages in this group'. " + + "This is the most targeted way to retry related failures after fixing a specific bug. " + + "You need a group ID, which you can get from GetFailureGroups. " + + "Returns InProgress if a retry is already running for this group." + )] + public async Task RetryFailureGroup( + [Description("The failure group ID from get_failure_groups results")] string groupId) + { + logger.LogInformation("MCP RetryFailureGroup invoked (groupId={GroupId})", groupId); + + if (retryingManager.IsOperationInProgressFor(groupId, RetryType.FailureGroup)) + { + logger.LogInformation("MCP RetryFailureGroup: operation already in progress for group '{GroupId}'", groupId); + return JsonSerializer.Serialize(new { Status = "InProgress", Message = $"A retry operation is already in progress for group '{groupId}'." }, McpJsonOptions.Default); + } + + var started = System.DateTime.UtcNow; + await retryingManager.Wait(groupId, RetryType.FailureGroup, started); + await messageSession.SendLocal(new RetryAllInGroup + { + GroupId = groupId, + Started = started + }); + + return JsonSerializer.Serialize(new { Status = "Accepted", Message = $"Retry requested for all messages in failure group '{groupId}'." }, McpJsonOptions.Default); + } +} diff --git a/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs b/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs index 2e317cba54..ae852de26e 100644 --- a/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs @@ -21,7 +21,7 @@ public async Task Handle(ArchiveMessage message, IMessageHandlerContext context) var failedMessage = await dataStore.ErrorBy(failedMessageId); - if (failedMessage.Status != FailedMessageStatus.Archived) + if (failedMessage is not null && failedMessage.Status != FailedMessageStatus.Archived) { await domainEvents.Raise(new FailedMessageArchived { diff --git a/src/ServiceControl/ServiceControl.csproj b/src/ServiceControl/ServiceControl.csproj index d931751d34..2475998650 100644 --- a/src/ServiceControl/ServiceControl.csproj +++ b/src/ServiceControl/ServiceControl.csproj @@ -33,6 +33,7 @@ + diff --git a/src/ServiceControl/WebApplicationExtensions.cs b/src/ServiceControl/WebApplicationExtensions.cs index 685bc7dc16..4d3be18f2c 100644 --- a/src/ServiceControl/WebApplicationExtensions.cs +++ b/src/ServiceControl/WebApplicationExtensions.cs @@ -3,13 +3,15 @@ namespace ServiceControl; using Infrastructure.SignalR; using Infrastructure.WebApi; using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.HttpOverrides; +using ModelContextProtocol.AspNetCore; using ServiceControl.Hosting.ForwardedHeaders; using ServiceControl.Hosting.Https; using ServiceControl.Infrastructure; public static class WebApplicationExtensions { - public static void UseServiceControl(this WebApplication app, ForwardedHeadersSettings forwardedHeadersSettings, HttpsSettings httpsSettings) + public static void UseServiceControl(this WebApplication app, ForwardedHeadersSettings forwardedHeadersSettings, HttpsSettings httpsSettings, bool enableMcpServer) { app.UseServiceControlForwardedHeaders(forwardedHeadersSettings); app.UseServiceControlHttps(httpsSettings); @@ -19,5 +21,10 @@ public static void UseServiceControl(this WebApplication app, ForwardedHeadersSe app.MapHub("/api/messagestream"); app.UseCors(); app.MapControllers(); + + if (enableMcpServer) + { + app.MapMcp("/mcp"); + } } } \ No newline at end of file