845 lines
28 KiB
Markdown
845 lines
28 KiB
Markdown
# MCP Custom Transports - Advanced Implementation Guide
|
|
|
|
The Model Context Protocol (MCP) provides flexibility in transport mechanisms, allowing custom implementations for specialized enterprise environments. This advanced guide explores custom transport implementations using Azure Event Grid and Azure Event Hubs as practical examples for building scalable, cloud-native MCP solutions.
|
|
|
|
> **Looking ahead:** this guide is written against **MCP Specification 2025-11-25**, where session ordering must be preserved per session (see Message Protocol below). The `2026-07-28` release candidate removes the protocol-level session entirely and requires `Mcp-Method`/`Mcp-Name` headers so gateways and custom transports can route per-request instead of per-session. See [What's Changing in MCP: The 2026-07-28 Release Candidate](../../01-CoreConcepts/mcp-2026-07-28-release-candidate.md).
|
|
|
|
## Introduction
|
|
|
|
While MCP's standard transports (stdio and HTTP streaming) serve most use cases, enterprise environments often require specialized transport mechanisms for improved scalability, reliability, and integration with existing cloud infrastructure. Custom transports enable MCP to leverage cloud-native messaging services for asynchronous communication, event-driven architectures, and distributed processing.
|
|
|
|
This lesson explores advanced transport implementations based on the latest MCP specification (2025-11-25), Azure messaging services, and established enterprise integration patterns.
|
|
|
|
### **MCP Transport Architecture**
|
|
|
|
**From MCP Specification (2025-11-25):**
|
|
|
|
- **Standard Transports**: stdio (recommended), HTTP streaming (for remote scenarios)
|
|
- **Custom Transports**: Any transport that implements the MCP message exchange protocol
|
|
- **Message Format**: JSON-RPC 2.0 with MCP-specific extensions
|
|
- **Bidirectional Communication**: Full duplex communication required for notifications and responses
|
|
|
|
## Learning Objectives
|
|
|
|
By the end of this advanced lesson, you will be able to:
|
|
|
|
- **Understand Custom Transport Requirements**: Implement MCP protocol over any transport layer while maintaining compliance
|
|
- **Build Azure Event Grid Transport**: Create event-driven MCP servers using Azure Event Grid for serverless scalability
|
|
- **Implement Azure Event Hubs Transport**: Design high-throughput MCP solutions using Azure Event Hubs for real-time streaming
|
|
- **Apply Enterprise Patterns**: Integrate custom transports with existing Azure infrastructure and security models
|
|
- **Handle Transport Reliability**: Implement message durability, ordering, and error handling for enterprise scenarios
|
|
- **Optimize Performance**: Design transport solutions for scale, latency, and throughput requirements
|
|
|
|
## **Transport Requirements**
|
|
|
|
### **Core Requirements from MCP Specification (2025-11-25):**
|
|
|
|
```yaml
|
|
Message Protocol:
|
|
format: "JSON-RPC 2.0 with MCP extensions"
|
|
bidirectional: "Full duplex communication required"
|
|
ordering: "Message ordering must be preserved per session"
|
|
|
|
Transport Layer:
|
|
reliability: "Transport MUST handle connection failures gracefully"
|
|
security: "Transport MUST support secure communication"
|
|
identification: "Each session MUST have unique identifier"
|
|
|
|
Custom Transport:
|
|
compliance: "MUST implement complete MCP message exchange"
|
|
extensibility: "MAY add transport-specific features"
|
|
interoperability: "MUST maintain protocol compatibility"
|
|
```
|
|
|
|
## **Azure Event Grid Transport Implementation**
|
|
|
|
Azure Event Grid provides a serverless event routing service ideal for event-driven MCP architectures. This implementation demonstrates how to build scalable, loosely-coupled MCP systems.
|
|
|
|
### **Architecture Overview**
|
|
|
|
```mermaid
|
|
graph TB
|
|
Client[MCP Client] --> EG[Azure Event Grid]
|
|
EG --> Server[MCP Server Function]
|
|
Server --> EG
|
|
EG --> Client
|
|
|
|
subgraph "Azure Services"
|
|
EG
|
|
Server
|
|
KV[Key Vault]
|
|
Monitor[Application Insights]
|
|
end
|
|
```
|
|
|
|
### **C# Implementation - Event Grid Transport**
|
|
|
|
```csharp
|
|
using Azure.Messaging.EventGrid;
|
|
using Microsoft.Extensions.Azure;
|
|
using System.Text.Json;
|
|
|
|
public class EventGridMcpTransport : IMcpTransport
|
|
{
|
|
private readonly EventGridPublisherClient _publisher;
|
|
private readonly string _topicEndpoint;
|
|
private readonly string _clientId;
|
|
|
|
public EventGridMcpTransport(string topicEndpoint, string accessKey, string clientId)
|
|
{
|
|
_publisher = new EventGridPublisherClient(
|
|
new Uri(topicEndpoint),
|
|
new AzureKeyCredential(accessKey));
|
|
_topicEndpoint = topicEndpoint;
|
|
_clientId = clientId;
|
|
}
|
|
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
var eventGridEvent = new EventGridEvent(
|
|
subject: $"mcp/{_clientId}",
|
|
eventType: "MCP.MessageReceived",
|
|
dataVersion: "1.0",
|
|
data: JsonSerializer.Serialize(message))
|
|
{
|
|
Id = Guid.NewGuid().ToString(),
|
|
EventTime = DateTimeOffset.UtcNow
|
|
};
|
|
|
|
await _publisher.SendEventAsync(eventGridEvent);
|
|
}
|
|
|
|
public async Task<McpMessage> ReceiveMessageAsync(CancellationToken cancellationToken)
|
|
{
|
|
// Event Grid is push-based, so implement webhook receiver
|
|
// This would typically be handled by Azure Functions trigger
|
|
throw new NotImplementedException("Use EventGridTrigger in Azure Functions");
|
|
}
|
|
}
|
|
|
|
// Azure Function for receiving Event Grid events
|
|
[FunctionName("McpEventGridReceiver")]
|
|
public async Task<IActionResult> HandleEventGridMessage(
|
|
[EventGridTrigger] EventGridEvent eventGridEvent,
|
|
ILogger log)
|
|
{
|
|
try
|
|
{
|
|
var mcpMessage = JsonSerializer.Deserialize<McpMessage>(
|
|
eventGridEvent.Data.ToString());
|
|
|
|
// Process MCP message
|
|
var response = await _mcpServer.ProcessMessageAsync(mcpMessage);
|
|
|
|
// Send response back via Event Grid
|
|
await _transport.SendMessageAsync(response);
|
|
|
|
return new OkResult();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
log.LogError(ex, "Error processing Event Grid MCP message");
|
|
return new BadRequestResult();
|
|
}
|
|
}
|
|
```
|
|
|
|
### **TypeScript Implementation - Event Grid Transport**
|
|
|
|
```typescript
|
|
import { EventGridPublisherClient, AzureKeyCredential } from "@azure/eventgrid";
|
|
import { McpTransport, McpMessage } from "./mcp-types";
|
|
|
|
export class EventGridMcpTransport implements McpTransport {
|
|
private publisher: EventGridPublisherClient;
|
|
private clientId: string;
|
|
|
|
constructor(
|
|
private topicEndpoint: string,
|
|
private accessKey: string,
|
|
clientId: string
|
|
) {
|
|
this.publisher = new EventGridPublisherClient(
|
|
topicEndpoint,
|
|
new AzureKeyCredential(accessKey)
|
|
);
|
|
this.clientId = clientId;
|
|
}
|
|
|
|
async sendMessage(message: McpMessage): Promise<void> {
|
|
const event = {
|
|
id: crypto.randomUUID(),
|
|
source: `mcp-client-${this.clientId}`,
|
|
type: "MCP.MessageReceived",
|
|
time: new Date(),
|
|
data: message
|
|
};
|
|
|
|
await this.publisher.sendEvents([event]);
|
|
}
|
|
|
|
// Event-driven receive via Azure Functions
|
|
onMessage(handler: (message: McpMessage) => Promise<void>): void {
|
|
// Implementation would use Azure Functions Event Grid trigger
|
|
// This is a conceptual interface for the webhook receiver
|
|
}
|
|
}
|
|
|
|
// Azure Functions implementation
|
|
import { app, InvocationContext, EventGridEvent } from "@azure/functions";
|
|
|
|
app.eventGrid("mcpEventGridHandler", {
|
|
handler: async (event: EventGridEvent, context: InvocationContext) => {
|
|
try {
|
|
const mcpMessage = event.data as McpMessage;
|
|
|
|
// Process MCP message
|
|
const response = await mcpServer.processMessage(mcpMessage);
|
|
|
|
// Send response via Event Grid
|
|
await transport.sendMessage(response);
|
|
|
|
} catch (error) {
|
|
context.error("Error processing MCP message:", error);
|
|
throw error;
|
|
}
|
|
}
|
|
});
|
|
```
|
|
|
|
### **Python Implementation - Event Grid Transport**
|
|
|
|
```python
|
|
from azure.eventgrid import EventGridPublisherClient, EventGridEvent
|
|
from azure.core.credentials import AzureKeyCredential
|
|
import asyncio
|
|
import json
|
|
from typing import Callable, Optional
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
class EventGridMcpTransport:
|
|
def __init__(self, topic_endpoint: str, access_key: str, client_id: str):
|
|
self.client = EventGridPublisherClient(
|
|
topic_endpoint,
|
|
AzureKeyCredential(access_key)
|
|
)
|
|
self.client_id = client_id
|
|
self.message_handler: Optional[Callable] = None
|
|
|
|
async def send_message(self, message: dict) -> None:
|
|
"""Send MCP message via Event Grid"""
|
|
event = EventGridEvent(
|
|
data=message,
|
|
subject=f"mcp/{self.client_id}",
|
|
event_type="MCP.MessageReceived",
|
|
data_version="1.0"
|
|
)
|
|
|
|
await self.client.send(event)
|
|
|
|
def on_message(self, handler: Callable[[dict], None]) -> None:
|
|
"""Register message handler for incoming events"""
|
|
self.message_handler = handler
|
|
|
|
# Azure Functions implementation
|
|
import azure.functions as func
|
|
import logging
|
|
|
|
def main(event: func.EventGridEvent) -> None:
|
|
"""Azure Functions Event Grid trigger for MCP messages"""
|
|
try:
|
|
# Parse MCP message from Event Grid event
|
|
mcp_message = json.loads(event.get_body().decode('utf-8'))
|
|
|
|
# Process MCP message
|
|
response = process_mcp_message(mcp_message)
|
|
|
|
# Send response back via Event Grid
|
|
# (Implementation would create new Event Grid client)
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error processing MCP Event Grid message: {e}")
|
|
raise
|
|
```
|
|
|
|
## **Azure Event Hubs Transport Implementation**
|
|
|
|
Azure Event Hubs provides high-throughput, real-time streaming capabilities for MCP scenarios requiring low latency and high message volume.
|
|
|
|
### **Architecture Overview**
|
|
|
|
```mermaid
|
|
graph TB
|
|
Client[MCP Client] --> EH[Azure Event Hubs]
|
|
EH --> Server[MCP Server]
|
|
Server --> EH
|
|
EH --> Client
|
|
|
|
subgraph "Event Hubs Features"
|
|
Partition[Partitioning]
|
|
Retention[Message Retention]
|
|
Scaling[Auto Scaling]
|
|
end
|
|
|
|
EH --> Partition
|
|
EH --> Retention
|
|
EH --> Scaling
|
|
```
|
|
|
|
### **C# Implementation - Event Hubs Transport**
|
|
|
|
```csharp
|
|
using Azure.Messaging.EventHubs;
|
|
using Azure.Messaging.EventHubs.Producer;
|
|
using Azure.Messaging.EventHubs.Consumer;
|
|
using System.Text;
|
|
|
|
public class EventHubsMcpTransport : IMcpTransport, IDisposable
|
|
{
|
|
private readonly EventHubProducerClient _producer;
|
|
private readonly EventHubConsumerClient _consumer;
|
|
private readonly string _consumerGroup;
|
|
private readonly CancellationTokenSource _cancellationTokenSource;
|
|
|
|
public EventHubsMcpTransport(
|
|
string connectionString,
|
|
string eventHubName,
|
|
string consumerGroup = "$Default")
|
|
{
|
|
_producer = new EventHubProducerClient(connectionString, eventHubName);
|
|
_consumer = new EventHubConsumerClient(
|
|
consumerGroup,
|
|
connectionString,
|
|
eventHubName);
|
|
_consumerGroup = consumerGroup;
|
|
_cancellationTokenSource = new CancellationTokenSource();
|
|
}
|
|
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
var messageBody = JsonSerializer.Serialize(message);
|
|
var eventData = new EventData(Encoding.UTF8.GetBytes(messageBody));
|
|
|
|
// Add MCP-specific properties
|
|
eventData.Properties.Add("MessageType", message.Method ?? "response");
|
|
eventData.Properties.Add("MessageId", message.Id);
|
|
eventData.Properties.Add("Timestamp", DateTimeOffset.UtcNow);
|
|
|
|
await _producer.SendAsync(new[] { eventData });
|
|
}
|
|
|
|
public async Task StartReceivingAsync(
|
|
Func<McpMessage, Task> messageHandler)
|
|
{
|
|
await foreach (PartitionEvent partitionEvent in _consumer.ReadEventsAsync(
|
|
_cancellationTokenSource.Token))
|
|
{
|
|
try
|
|
{
|
|
var messageBody = Encoding.UTF8.GetString(
|
|
partitionEvent.Data.EventBody.ToArray());
|
|
var mcpMessage = JsonSerializer.Deserialize<McpMessage>(messageBody);
|
|
|
|
await messageHandler(mcpMessage);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// Handle deserialization or processing errors
|
|
Console.WriteLine($"Error processing message: {ex.Message}");
|
|
}
|
|
}
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
_cancellationTokenSource?.Cancel();
|
|
_producer?.DisposeAsync().AsTask().Wait();
|
|
_consumer?.DisposeAsync().AsTask().Wait();
|
|
_cancellationTokenSource?.Dispose();
|
|
}
|
|
}
|
|
```
|
|
|
|
### **TypeScript Implementation - Event Hubs Transport**
|
|
|
|
```typescript
|
|
import {
|
|
EventHubProducerClient,
|
|
EventHubConsumerClient,
|
|
EventData
|
|
} from "@azure/event-hubs";
|
|
|
|
export class EventHubsMcpTransport implements McpTransport {
|
|
private producer: EventHubProducerClient;
|
|
private consumer: EventHubConsumerClient;
|
|
private isReceiving = false;
|
|
|
|
constructor(
|
|
private connectionString: string,
|
|
private eventHubName: string,
|
|
private consumerGroup: string = "$Default"
|
|
) {
|
|
this.producer = new EventHubProducerClient(
|
|
connectionString,
|
|
eventHubName
|
|
);
|
|
this.consumer = new EventHubConsumerClient(
|
|
consumerGroup,
|
|
connectionString,
|
|
eventHubName
|
|
);
|
|
}
|
|
|
|
async sendMessage(message: McpMessage): Promise<void> {
|
|
const eventData: EventData = {
|
|
body: JSON.stringify(message),
|
|
properties: {
|
|
messageType: message.method || "response",
|
|
messageId: message.id,
|
|
timestamp: new Date().toISOString()
|
|
}
|
|
};
|
|
|
|
await this.producer.sendBatch([eventData]);
|
|
}
|
|
|
|
async startReceiving(
|
|
messageHandler: (message: McpMessage) => Promise<void>
|
|
): Promise<void> {
|
|
if (this.isReceiving) return;
|
|
|
|
this.isReceiving = true;
|
|
|
|
const subscription = this.consumer.subscribe({
|
|
processEvents: async (events, context) => {
|
|
for (const event of events) {
|
|
try {
|
|
const messageBody = event.body as string;
|
|
const mcpMessage: McpMessage = JSON.parse(messageBody);
|
|
|
|
await messageHandler(mcpMessage);
|
|
|
|
// Update checkpoint for at-least-once delivery
|
|
await context.updateCheckpoint(event);
|
|
} catch (error) {
|
|
console.error("Error processing Event Hubs message:", error);
|
|
}
|
|
}
|
|
},
|
|
processError: async (err, context) => {
|
|
console.error("Event Hubs error:", err);
|
|
}
|
|
});
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
this.isReceiving = false;
|
|
await this.producer.close();
|
|
await this.consumer.close();
|
|
}
|
|
}
|
|
```
|
|
|
|
### **Python Implementation - Event Hubs Transport**
|
|
|
|
```python
|
|
from azure.eventhub import EventHubProducerClient, EventHubConsumerClient
|
|
from azure.eventhub import EventData
|
|
import json
|
|
import asyncio
|
|
from typing import Callable, Dict, Any
|
|
import logging
|
|
|
|
class EventHubsMcpTransport:
|
|
def __init__(
|
|
self,
|
|
connection_string: str,
|
|
eventhub_name: str,
|
|
consumer_group: str = "$Default"
|
|
):
|
|
self.producer = EventHubProducerClient.from_connection_string(
|
|
connection_string,
|
|
eventhub_name=eventhub_name
|
|
)
|
|
self.consumer = EventHubConsumerClient.from_connection_string(
|
|
connection_string,
|
|
consumer_group=consumer_group,
|
|
eventhub_name=eventhub_name
|
|
)
|
|
self.is_receiving = False
|
|
|
|
async def send_message(self, message: Dict[str, Any]) -> None:
|
|
"""Send MCP message via Event Hubs"""
|
|
event_data = EventData(json.dumps(message))
|
|
|
|
# Add MCP-specific properties
|
|
event_data.properties = {
|
|
"messageType": message.get("method", "response"),
|
|
"messageId": message.get("id"),
|
|
"timestamp": "2025-01-14T10:30:00Z" # Use actual timestamp
|
|
}
|
|
|
|
async with self.producer:
|
|
event_data_batch = await self.producer.create_batch()
|
|
event_data_batch.add(event_data)
|
|
await self.producer.send_batch(event_data_batch)
|
|
|
|
async def start_receiving(
|
|
self,
|
|
message_handler: Callable[[Dict[str, Any]], None]
|
|
) -> None:
|
|
"""Start receiving MCP messages from Event Hubs"""
|
|
if self.is_receiving:
|
|
return
|
|
|
|
self.is_receiving = True
|
|
|
|
async with self.consumer:
|
|
await self.consumer.receive(
|
|
on_event=self._on_event_received(message_handler),
|
|
starting_position="-1" # Start from beginning
|
|
)
|
|
|
|
def _on_event_received(self, handler: Callable):
|
|
"""Internal event handler wrapper"""
|
|
async def handle_event(partition_context, event):
|
|
try:
|
|
# Parse MCP message from Event Hubs event
|
|
message_body = event.body_as_str(encoding='UTF-8')
|
|
mcp_message = json.loads(message_body)
|
|
|
|
# Process MCP message
|
|
await handler(mcp_message)
|
|
|
|
# Update checkpoint for at-least-once delivery
|
|
await partition_context.update_checkpoint(event)
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error processing Event Hubs message: {e}")
|
|
|
|
return handle_event
|
|
|
|
async def close(self) -> None:
|
|
"""Clean up transport resources"""
|
|
self.is_receiving = False
|
|
await self.producer.close()
|
|
await self.consumer.close()
|
|
```
|
|
|
|
## **Advanced Transport Patterns**
|
|
|
|
### **Message Durability and Reliability**
|
|
|
|
```csharp
|
|
// Implementing message durability with retry logic
|
|
public class ReliableTransportWrapper : IMcpTransport
|
|
{
|
|
private readonly IMcpTransport _innerTransport;
|
|
private readonly RetryPolicy _retryPolicy;
|
|
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
await _retryPolicy.ExecuteAsync(async () =>
|
|
{
|
|
try
|
|
{
|
|
await _innerTransport.SendMessageAsync(message);
|
|
}
|
|
catch (TransportException ex) when (ex.IsRetryable)
|
|
{
|
|
// Log and retry
|
|
throw;
|
|
}
|
|
});
|
|
}
|
|
}
|
|
```
|
|
|
|
### **Transport Security Integration**
|
|
|
|
```csharp
|
|
// Integrating Azure Key Vault for transport security
|
|
public class SecureTransportFactory
|
|
{
|
|
private readonly SecretClient _keyVaultClient;
|
|
|
|
public async Task<IMcpTransport> CreateEventGridTransportAsync()
|
|
{
|
|
var accessKey = await _keyVaultClient.GetSecretAsync("EventGridAccessKey");
|
|
var topicEndpoint = await _keyVaultClient.GetSecretAsync("EventGridTopic");
|
|
|
|
return new EventGridMcpTransport(
|
|
topicEndpoint.Value.Value,
|
|
accessKey.Value.Value,
|
|
Environment.MachineName
|
|
);
|
|
}
|
|
}
|
|
```
|
|
|
|
### **Transport Monitoring and Observability**
|
|
|
|
```csharp
|
|
// Adding telemetry to custom transports
|
|
public class ObservableTransport : IMcpTransport
|
|
{
|
|
private readonly IMcpTransport _transport;
|
|
private readonly ILogger _logger;
|
|
private readonly TelemetryClient _telemetryClient;
|
|
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
using var activity = Activity.StartActivity("MCP.Transport.Send");
|
|
activity?.SetTag("transport.type", "EventGrid");
|
|
activity?.SetTag("message.method", message.Method);
|
|
|
|
var stopwatch = Stopwatch.StartNew();
|
|
|
|
try
|
|
{
|
|
await _transport.SendMessageAsync(message);
|
|
|
|
_telemetryClient.TrackDependency(
|
|
"EventGrid",
|
|
"SendMessage",
|
|
DateTime.UtcNow.Subtract(stopwatch.Elapsed),
|
|
stopwatch.Elapsed,
|
|
true
|
|
);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_telemetryClient.TrackException(ex);
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
## **Enterprise Integration Scenarios**
|
|
|
|
### **Scenario 1: Distributed MCP Processing**
|
|
|
|
Using Azure Event Grid for distributing MCP requests across multiple processing nodes:
|
|
|
|
```yaml
|
|
Architecture:
|
|
- MCP Client sends requests to Event Grid topic
|
|
- Multiple Azure Functions subscribe to process different tool types
|
|
- Results aggregated and returned via separate response topic
|
|
|
|
Benefits:
|
|
- Horizontal scaling based on message volume
|
|
- Fault tolerance through redundant processors
|
|
- Cost optimization with serverless compute
|
|
```
|
|
|
|
### **Scenario 2: Real-time MCP Streaming**
|
|
|
|
Using Azure Event Hubs for high-frequency MCP interactions:
|
|
|
|
```yaml
|
|
Architecture:
|
|
- MCP Client streams continuous requests via Event Hubs
|
|
- Stream Analytics processes and routes messages
|
|
- Multiple consumers handle different aspect of processing
|
|
|
|
Benefits:
|
|
- Low latency for real-time scenarios
|
|
- High throughput for batch processing
|
|
- Built-in partitioning for parallel processing
|
|
```
|
|
|
|
### **Scenario 3: Hybrid Transport Architecture**
|
|
|
|
Combining multiple transports for different use cases:
|
|
|
|
```csharp
|
|
public class HybridMcpTransport : IMcpTransport
|
|
{
|
|
private readonly IMcpTransport _realtimeTransport; // Event Hubs
|
|
private readonly IMcpTransport _batchTransport; // Event Grid
|
|
private readonly IMcpTransport _fallbackTransport; // HTTP Streaming
|
|
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
// Route based on message characteristics
|
|
var transport = message.Method switch
|
|
{
|
|
"tools/call" when IsRealtime(message) => _realtimeTransport,
|
|
"resources/read" when IsBatch(message) => _batchTransport,
|
|
_ => _fallbackTransport
|
|
};
|
|
|
|
await transport.SendMessageAsync(message);
|
|
}
|
|
}
|
|
```
|
|
|
|
## **Performance Optimization**
|
|
|
|
### **Message Batching for Event Grid**
|
|
|
|
```csharp
|
|
public class BatchingEventGridTransport : IMcpTransport
|
|
{
|
|
private readonly List<McpMessage> _messageBuffer = new();
|
|
private readonly Timer _flushTimer;
|
|
private const int MaxBatchSize = 100;
|
|
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
lock (_messageBuffer)
|
|
{
|
|
_messageBuffer.Add(message);
|
|
|
|
if (_messageBuffer.Count >= MaxBatchSize)
|
|
{
|
|
_ = Task.Run(FlushMessages);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task FlushMessages()
|
|
{
|
|
List<McpMessage> toSend;
|
|
lock (_messageBuffer)
|
|
{
|
|
toSend = new List<McpMessage>(_messageBuffer);
|
|
_messageBuffer.Clear();
|
|
}
|
|
|
|
if (toSend.Any())
|
|
{
|
|
var events = toSend.Select(CreateEventGridEvent);
|
|
await _publisher.SendEventsAsync(events);
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
### **Partitioning Strategy for Event Hubs**
|
|
|
|
```csharp
|
|
public class PartitionedEventHubsTransport : IMcpTransport
|
|
{
|
|
public async Task SendMessageAsync(McpMessage message)
|
|
{
|
|
// Partition by client ID for session affinity
|
|
var partitionKey = ExtractClientId(message);
|
|
|
|
var eventData = new EventData(JsonSerializer.SerializeToUtf8Bytes(message))
|
|
{
|
|
PartitionKey = partitionKey
|
|
};
|
|
|
|
await _producer.SendAsync(new[] { eventData });
|
|
}
|
|
}
|
|
```
|
|
|
|
## **Testing Custom Transports**
|
|
|
|
### **Unit Testing with Test Doubles**
|
|
|
|
```csharp
|
|
[Test]
|
|
public async Task EventGridTransport_SendMessage_PublishesCorrectEvent()
|
|
{
|
|
// Arrange
|
|
var mockPublisher = new Mock<EventGridPublisherClient>();
|
|
var transport = new EventGridMcpTransport(mockPublisher.Object);
|
|
var message = new McpMessage { Method = "tools/list", Id = "test-123" };
|
|
|
|
// Act
|
|
await transport.SendMessageAsync(message);
|
|
|
|
// Assert
|
|
mockPublisher.Verify(
|
|
x => x.SendEventAsync(
|
|
It.Is<EventGridEvent>(e =>
|
|
e.EventType == "MCP.MessageReceived" &&
|
|
e.Subject == "mcp/test-client"
|
|
)
|
|
),
|
|
Times.Once
|
|
);
|
|
}
|
|
```
|
|
|
|
### **Integration Testing with Azure Test Containers**
|
|
|
|
```csharp
|
|
[Test]
|
|
public async Task EventHubsTransport_IntegrationTest()
|
|
{
|
|
// Using Testcontainers for integration testing
|
|
var eventHubsContainer = new EventHubsContainer()
|
|
.WithEventHub("test-hub");
|
|
|
|
await eventHubsContainer.StartAsync();
|
|
|
|
var transport = new EventHubsMcpTransport(
|
|
eventHubsContainer.GetConnectionString(),
|
|
"test-hub"
|
|
);
|
|
|
|
// Test message round-trip
|
|
var sentMessage = new McpMessage { Method = "test", Id = "123" };
|
|
McpMessage receivedMessage = null;
|
|
|
|
await transport.StartReceivingAsync(msg => {
|
|
receivedMessage = msg;
|
|
return Task.CompletedTask;
|
|
});
|
|
|
|
await transport.SendMessageAsync(sentMessage);
|
|
await Task.Delay(1000); // Allow for message processing
|
|
|
|
Assert.That(receivedMessage?.Id, Is.EqualTo("123"));
|
|
}
|
|
```
|
|
|
|
## **Best Practices and Guidelines**
|
|
|
|
### **Transport Design Principles**
|
|
|
|
1. **Idempotency**: Ensure message processing is idempotent to handle duplicates
|
|
2. **Error Handling**: Implement comprehensive error handling and dead letter queues
|
|
3. **Monitoring**: Add detailed telemetry and health checks
|
|
4. **Security**: Use managed identities and least privilege access
|
|
5. **Performance**: Design for your specific latency and throughput requirements
|
|
|
|
### **Azure-Specific Recommendations**
|
|
|
|
1. **Use Managed Identity**: Avoid connection strings in production
|
|
2. **Implement Circuit Breakers**: Protect against Azure service outages
|
|
3. **Monitor Costs**: Track message volume and processing costs
|
|
4. **Plan for Scale**: Design partitioning and scaling strategies early
|
|
5. **Test Thoroughly**: Use Azure DevTest Labs for comprehensive testing
|
|
|
|
## **Conclusion**
|
|
|
|
Custom MCP transports enable powerful enterprise scenarios using Azure's messaging services. By implementing Event Grid or Event Hubs transports, you can build scalable, reliable MCP solutions that integrate seamlessly with existing Azure infrastructure.
|
|
|
|
The examples provided demonstrate production-ready patterns for implementing custom transports while maintaining MCP protocol compliance and Azure best practices.
|
|
|
|
## **Additional Resources**
|
|
|
|
- [MCP Specification 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/)
|
|
- [Azure Event Grid Documentation](https://docs.microsoft.com/azure/event-grid/)
|
|
- [Azure Event Hubs Documentation](https://docs.microsoft.com/azure/event-hubs/)
|
|
- [Azure Functions Event Grid Trigger](https://docs.microsoft.com/azure/azure-functions/functions-bindings-event-grid)
|
|
- [Azure SDK for .NET](https://github.com/Azure/azure-sdk-for-net)
|
|
- [Azure SDK for TypeScript](https://github.com/Azure/azure-sdk-for-js)
|
|
- [Azure SDK for Python](https://github.com/Azure/azure-sdk-for-python)
|
|
|
|
---
|
|
|
|
> *This guide focuses on practical implementation patterns for production MCP systems. Always validate transport implementations against your specific requirements and Azure service limits.*
|
|
> **Current Standard**: This guide reflects [MCP Specification 2025-11-25](https://modelcontextprotocol.io/specification/2025-11-25/) transport requirements and advanced transport patterns for enterprise environments.
|
|
|
|
|
|
## What's Next
|
|
- [6. Community Contributions](../../06-CommunityContributions/README.md) |