Changed the publish event implementation

Author: anannya03

docs/azmcp-commands.md

@@ -576,7 +576,7 @@ azmcp eventgrid events publish --subscription "my-subscription" \
 azmcp eventgrid events publish --subscription "my-subscription" \
                                --resource-group "my-rg" \
                                --topic "my-topic" \
-                               --event-data '{"specversion": "1.0", "type": "com.example.order.created", "source": "/orders", "id": "123", "time": "2023-01-01T12:00:00Z", "data": {"orderId": "123"}}' \
+                               --event-data '{"specversion": "1.0", "type": "com.example.order.created", "source": "/orders", "id": "123", "time": "2023-01-01T12:00:00Z", "datacontenttype": "application/json", "data": {"orderId": "123"}}' \
                                --event-schema "CloudEvents"
 
 # Publish with custom schema

tools/Azure.Mcp.Tools.EventGrid/src/Commands/EventGridJsonContext.cs

@@ -5,15 +5,18 @@
 using Azure.Mcp.Tools.EventGrid.Commands.Events;
 using Azure.Mcp.Tools.EventGrid.Commands.Subscription;
 using Azure.Mcp.Tools.EventGrid.Commands.Topic;
+using Azure.Mcp.Tools.EventGrid.Models;
 
 namespace Azure.Mcp.Tools.EventGrid.Commands;
 
+// JsonSerializable attributes for all types used in EventGrid command responses and event serialization
 [JsonSerializable(typeof(EventsPublishCommand.EventsPublishCommandResult))]
 [JsonSerializable(typeof(SubscriptionListCommand.SubscriptionListCommandResult))]
 [JsonSerializable(typeof(TopicListCommand.TopicListCommandResult))]
 [JsonSerializable(typeof(EventGridSubscriptionInfo))]
 [JsonSerializable(typeof(EventGridTopicInfo))]
 [JsonSerializable(typeof(EventPublishResult))]
+[JsonSerializable(typeof(EventGridEventSchema))] // For individual event serialization to EventGrid
 [JsonSourceGenerationOptions(PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase, WriteIndented = true, DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull)]
 internal sealed partial class EventGridJsonContext : JsonSerializerContext
 {

tools/Azure.Mcp.Tools.EventGrid/src/Commands/Events/EventsPublishCommand.cs

@@ -99,7 +99,7 @@ public override async Task<CommandResponse> ExecuteAsync(CommandContext context,
         Azure.RequestFailedException reqEx when reqEx.Status == 400 =>
             "Invalid event data or schema format. Please verify the event data is valid JSON and matches the expected schema.",
         ArgumentException argEx when argEx.Message.Contains("schema") =>
-            "Invalid event schema specified. Supported schemas are: CloudEvents, EventGridEvent, or Custom.",
+            "Invalid event schema specified. Supported schemas are: CloudEvents, EventGrid, or Custom.",
         System.Text.Json.JsonException jsonEx =>
             $"Invalid JSON format in event data: {jsonEx.Message}",
         _ => base.GetErrorMessage(ex)

tools/Azure.Mcp.Tools.EventGrid/src/Models/EventGridEventSchema.cs

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System.Text.Json.Nodes;
+using System.Text.Json.Serialization;
+
+namespace Azure.Mcp.Tools.EventGrid.Models;
+
+/// <summary>
+/// Represents an event conforming to the EventGrid event schema for HTTP API publishing.
+/// </summary>
+public sealed class EventGridEventSchema
+{
+    [JsonPropertyName("id")]
+    public required string Id { get; set; }
+
+    [JsonPropertyName("subject")]
+    public required string Subject { get; set; }
+
+    [JsonPropertyName("eventType")]
+    public required string EventType { get; set; }
+
+    [JsonPropertyName("dataVersion")]
+    public required string DataVersion { get; set; }
+
+    [JsonPropertyName("data")]
+    public JsonNode? Data { get; set; }
+
+    [JsonPropertyName("eventTime")]
+    public DateTimeOffset EventTime { get; set; }
+}

tools/Azure.Mcp.Tools.EventGrid/src/Options/EventGridOptionDefinitions.cs

@@ -35,6 +35,6 @@ public static class EventGridOptionDefinitions
         $"--{EventSchemaParam}"
     )
     {
-        Description = "The event schema type (CloudEvents, EventGridEvent, or Custom). Defaults to EventGridEvent."
+        Description = "The event schema type (CloudEvents, EventGrid, or Custom). Defaults to EventGrid."
     };
 }

tools/Azure.Mcp.Tools.EventGrid/src/Services/EventGridService.cs

@@ -1,11 +1,12 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-using System.Net.Http;
-using System.Text;
 using System.Text.Json;
+using System.Text.Json.Nodes;
 using Azure.Core;
 using Azure.Mcp.Core.Options;
+using Azure.Mcp.Tools.EventGrid.Commands;
+using Azure.Mcp.Tools.EventGrid.Models;
 using Azure.Messaging.EventGrid;
 using Azure.ResourceManager.EventGrid;
 using Azure.ResourceManager.EventGrid.Models;
@@ -131,62 +132,28 @@ public async Task<EventPublishResult> PublishEventsAsync(
             // Get credential using standardized method from base class for Azure AD authentication
             var credential = await GetCredential(tenant);
 
-            // Parse and validate event data
-            var events = ParseAndValidateEventData(eventData, eventSchema ?? "EventGridEvent");
+            // Parse and validate event data directly to EventGridEventSchema
+            var eventGridEventSchemas = ParseAndValidateEventData(eventData, eventSchema ?? "EventGridEvent");
 
-            // Use raw HTTP approach to completely avoid AOT serialization issues
-            using var httpClient = new HttpClient();
+            // Use EventGridPublisherClient with BinaryData for AOT-compatible publishing
+            var publisherClient = new EventGridPublisherClient(topic.Data.Endpoint, credential);
 
-            // Create events as raw JSON strings using lazy evaluation
-            var eventJsonStrings = events.Select(evt =>
+            // Serialize each event individually to JSON using source-generated context and convert to BinaryData
+            var eventsBinaryData = eventGridEventSchemas.Select(eventSchema =>
             {
-                // Use JsonEncodedText.Encode for AOT-compatible JSON escaping
-                var escapedSubject = JsonEncodedText.Encode(evt.Subject);
-                var escapedEventType = JsonEncodedText.Encode(evt.EventType);
-                var escapedId = JsonEncodedText.Encode(evt.Id);
-                var escapedDataVersion = JsonEncodedText.Encode(evt.DataVersion);
-                var formattedEventTime = evt.EventTime.ToString("yyyy-MM-ddTHH:mm:ss.fffZ");
-
-                return $$"""
-                {
-                    "id": "{{escapedId}}",
-                    "subject": "{{escapedSubject}}",
-                    "eventType": "{{escapedEventType}}",
-                    "dataVersion": "{{escapedDataVersion}}",
-                    "eventTime": "{{formattedEventTime}}",
-                    "data": {{evt.Data ?? "{}"}}
-                }
-                """;
-            });
-
-            var jsonPayload = $"[{string.Join(",", eventJsonStrings)}]";
+                var jsonString = JsonSerializer.Serialize(eventSchema, EventGridJsonContext.Default.EventGridEventSchema);
+                return BinaryData.FromString(jsonString);
+            }).ToArray();
 
             // Get event count for logging (this will materialize the enumerable once)
-            var eventCount = events.Count();
+            var eventCount = eventsBinaryData.Length;
             _logger.LogInformation("Publishing {EventCount} events to topic '{TopicName}' with operation ID: {OperationId}",
                 eventCount, topicName, operationId);
 
             try
             {
-                // Send the events using raw HTTP POST to avoid EventGridPublisherClient AOT issues
-                var content = new StringContent(jsonPayload, Encoding.UTF8, "application/json");
-
-                // Get access token for Event Grid using Azure AD authentication
-                var tokenRequestContext = new TokenRequestContext(new[] { "https://eventgrid.azure.net/.default" });
-                var tokenResult = await credential.GetTokenAsync(tokenRequestContext, CancellationToken.None);
-
-                // Add Authorization header with Bearer token for Azure AD authentication
-                httpClient.DefaultRequestHeaders.Authorization =
-                    new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", tokenResult.Token);
-
-                var response = await httpClient.PostAsync(topic.Data.Endpoint, content);
-
-                if (!response.IsSuccessStatusCode)
-                {
-                    var responseContent = await response.Content.ReadAsStringAsync();
-                    throw new HttpRequestException(
-                        $"Event Grid returned status code {response.StatusCode}. Response: {responseContent}");
-                }
+                // Send events using EventGridPublisherClient with BinaryData (AOT-compatible)
+                await publisherClient.SendEventsAsync(eventsBinaryData);
             }
             catch (Exception publishEx)
             {
@@ -219,25 +186,25 @@ public async Task<EventPublishResult> PublishEventsAsync(
         }
     }
 
-    private static IEnumerable<EventData> ParseAndValidateEventData(string eventData, string eventSchema)
+    private static IEnumerable<EventGridEventSchema> ParseAndValidateEventData(string eventData, string eventSchema)
     {
         try
         {
             // Parse the JSON data
             var jsonDocument = JsonDocument.Parse(eventData);
 
-            IEnumerable<EventData> events;
+            IEnumerable<EventGridEventSchema> events;
 
             if (jsonDocument.RootElement.ValueKind == JsonValueKind.Array)
             {
                 // Handle array of events - use lazy evaluation
                 events = jsonDocument.RootElement.EnumerateArray()
-                    .Select(eventElement => CreateEventDataFromJsonElement(eventElement, eventSchema));
+                    .Select(eventElement => CreateEventGridEventSchemaFromJsonElement(eventElement, eventSchema));
             }
             else
             {
                 // Handle single event - return single item enumerable
-                events = new[] { CreateEventDataFromJsonElement(jsonDocument.RootElement, eventSchema) };
+                events = new[] { CreateEventGridEventSchemaFromJsonElement(jsonDocument.RootElement, eventSchema) };
             }
 
             // Force evaluation to validate all events before returning
@@ -255,48 +222,98 @@ private static IEnumerable<EventData> ParseAndValidateEventData(string eventData
         }
     }
 
-    private static EventData CreateEventDataFromJsonElement(JsonElement eventElement, string eventSchema)
+    private static EventGridEventSchema CreateEventGridEventSchemaFromJsonElement(JsonElement eventElement, string eventSchema)
     {
-        // Extract required properties
-        var eventType = eventElement.TryGetProperty("eventType", out var eventTypeProp) ? eventTypeProp.GetString() :
+        string? eventType, subject, dataVersion;
+        DateTimeOffset eventTime;
+        
+        // Extract event ID early for logging purposes
+        var id = eventElement.TryGetProperty("id", out var idProp) ? idProp.GetString() : Guid.NewGuid().ToString();
+        
+        if (eventSchema.Equals("CloudEvents", StringComparison.OrdinalIgnoreCase))
+        {
+            // CloudEvents spec handling (v1.0)
+            eventType = eventElement.TryGetProperty("type", out var typeProp) ? typeProp.GetString() : "CustomEvent";
+            
+            // CloudEvents uses "source" field, but we can fall back to "subject" for compatibility
+            subject = eventElement.TryGetProperty("source", out var sourceProp) ? sourceProp.GetString() :
+                     eventElement.TryGetProperty("subject", out var subjectProp) ? subjectProp.GetString() : "/default/subject";
+            
+            // CloudEvents uses "specversion" for schema version
+            dataVersion = eventElement.TryGetProperty("specversion", out var specProp) ? specProp.GetString() : "1.0";
+            
+            // CloudEvents uses "time" field
+            eventTime = eventElement.TryGetProperty("time", out var timeProp) && timeProp.TryGetDateTimeOffset(out var timeValue)
+                       ? timeValue : DateTimeOffset.UtcNow;
+            
+            // Handle datacontenttype - CloudEvents v1.0 spec field for content type of data payload
+            var dataContentType = eventElement.TryGetProperty("datacontenttype", out var dataContentTypeProp) 
+                ? dataContentTypeProp.GetString() 
+                : "application/json"; // Default per CloudEvents spec
+            
+            // Log and validate datacontenttype for debugging and monitoring purposes
+            
+            if (!string.Equals(dataContentType, "application/json", StringComparison.OrdinalIgnoreCase))
+            {
+                // Log when non-JSON content types are used - this helps with debugging
+                // Note: EventGrid will accept the event regardless of datacontenttype,
+                // but subscribers should handle non-JSON content types appropriately
+                // Common non-JSON types: application/xml, text/plain, application/octet-stream
+                
+                // For now, we'll just validate that it's a recognized MIME type format
+                if (string.IsNullOrWhiteSpace(dataContentType) || !dataContentType.Contains('/'))
+                {
+                    throw new ArgumentException($"Invalid datacontenttype '{dataContentType}' in CloudEvent with id '{id}'. Must be a valid MIME type (e.g., 'application/xml', 'text/plain').");
+                }
+            }
+        }
+        else if (eventSchema.Equals("EventGrid", StringComparison.OrdinalIgnoreCase))
+        {
+            // EventGrid spec handling
+            eventType = eventElement.TryGetProperty("eventType", out var eventTypeProp) ? eventTypeProp.GetString() : "CustomEvent";
+            subject = eventElement.TryGetProperty("subject", out var subjectProp) ? subjectProp.GetString() : "/default/subject";
+            dataVersion = eventElement.TryGetProperty("dataVersion", out var dataVersionProp) ? dataVersionProp.GetString() : "1.0";
+            eventTime = eventElement.TryGetProperty("eventTime", out var timeProp) && timeProp.TryGetDateTimeOffset(out var eventTimeValue)
+                       ? eventTimeValue : DateTimeOffset.UtcNow;
+        }
+        else // Custom schema
+        {
+            // For custom schema, try both CloudEvents and EventGrid field names for flexibility
+            eventType = eventElement.TryGetProperty("eventType", out var eventTypeProp) ? eventTypeProp.GetString() :
                        eventElement.TryGetProperty("type", out var typeProp) ? typeProp.GetString() : "CustomEvent";
 
-        var subject = eventElement.TryGetProperty("subject", out var subjectProp) ? subjectProp.GetString() : "/default/subject";
+            subject = eventElement.TryGetProperty("subject", out var subjectProp) ? subjectProp.GetString() :
+                     eventElement.TryGetProperty("source", out var sourceProp) ? sourceProp.GetString() : "/default/subject";
 
-        var dataVersion = eventElement.TryGetProperty("dataVersion", out var dataVersionProp) ? dataVersionProp.GetString() : "1.0";
+            dataVersion = eventElement.TryGetProperty("dataVersion", out var dataVersionProp) ? dataVersionProp.GetString() :
+                         eventElement.TryGetProperty("specversion", out var specProp) ? specProp.GetString() : "1.0";
 
-        // Extract data payload as raw JSON string for AOT compatibility
-        string? data = null;
+            eventTime = eventElement.TryGetProperty("eventTime", out var eventTimeProp) && eventTimeProp.TryGetDateTimeOffset(out var eventTimeValue) ? eventTimeValue :
+                       eventElement.TryGetProperty("time", out var timeProp) && timeProp.TryGetDateTimeOffset(out var timeValue) ? timeValue : DateTimeOffset.UtcNow;
+        }
+
+        // Extract data payload and parse as JsonNode for AOT compatibility
+        JsonNode? data = null;
         if (eventElement.TryGetProperty("data", out var dataProp))
         {
-            data = dataProp.GetRawText();
+            data = JsonNode.Parse(dataProp.GetRawText());
         }
 
-        var id = eventElement.TryGetProperty("id", out var idProp) ? idProp.GetString() : Guid.NewGuid().ToString();
-        var eventTime = eventElement.TryGetProperty("eventTime", out var timeProp) && timeProp.TryGetDateTimeOffset(out var eventTimeValue)
-                       ? eventTimeValue : DateTimeOffset.UtcNow;
+        // For CloudEvents schema, we've already captured datacontenttype above for validation/logging
+        // The EventGrid schema doesn't have a direct equivalent, so we don't persist it in the final event
 
-        // Create a simple event data structure
-        return new EventData(
-            Id: id ?? Guid.NewGuid().ToString(),
-            Subject: subject ?? "/default/subject",
-            EventType: eventType ?? "CustomEvent",
-            DataVersion: dataVersion ?? "1.0",
-            Data: data,
-            EventTime: eventTime,
-            Schema: eventSchema);
+        // Create EventGridEventSchema directly
+        return new EventGridEventSchema
+        {
+            Id = id ?? Guid.NewGuid().ToString(),
+            Subject = subject ?? "/default/subject",
+            EventType = eventType ?? "CustomEvent",
+            DataVersion = dataVersion ?? "1.0",
+            Data = data,
+            EventTime = eventTime
+        };
     }
 
-    // Simple record to hold event data for validation
-    private record EventData(
-        string Id,
-        string Subject,
-        string EventType,
-        string DataVersion,
-        string? Data,
-        DateTimeOffset EventTime,
-        string Schema);
-
     private async Task GetSubscriptionsForSpecificTopic(
         SubscriptionResource subscriptionResource,
         string? resourceGroup,