29 KiB
MCP Transpaturi Personalizate - Ghid Avansat de Implementare
Protocolul Model Context (MCP) oferă flexibilitate în mecanismele de transport, permițând implementări personalizate pentru medii enterprise specializate. Acest ghid avansat explorează implementările personalizate de transport folosind Azure Event Grid și Azure Event Hubs ca exemple practice pentru construirea soluțiilor scalabile, native cloud MCP.
Introducere
Deși transporturile standard MCP (stdio și streaming HTTP) deservesc majoritatea scenariilor, mediile enterprise necesită adesea mecanisme specializate de transport pentru îmbunătățirea scalabilității, fiabilității și integrării cu infrastructura cloud existentă. Transpaturile personalizate permit MCP să valorifice serviciile de mesagerie native cloud pentru comunicare asincronă, arhitecturi bazate pe evenimente și procesare distribuită.
Această lecție explorează implementări avansate de transport bazate pe cea mai recentă specificație MCP (2025-11-25), serviciile de mesagerie Azure și modele consacrate de integrare enterprise.
Arhitectura Transportului MCP
Din Specificația MCP (2025-11-25):
- Transpaturi Standard: stdio (recomandat), streaming HTTP (pentru scenarii remote)
- Transpaturi Personalizate: Orice transport care implementează protocolul de schimb de mesaje MCP
- Formatul Mesajului: JSON-RPC 2.0 cu extensii specifice MCP
- Comunicare Bidirecțională: Comunicare full duplex necesară pentru notificări și răspunsuri
Obiective de Învățare
La finalul acestei lecții avansate, veți putea să:
- Înțelegeți Cerințele pentru Transpaturi Personalizate: Implementați protocolul MCP peste orice strat de transport păstrând conformitatea
- Construiți Transportul Azure Event Grid: Creați servere MCP bazate pe evenimente folosind Azure Event Grid pentru scalabilitate serverless
- Implementați Transportul Azure Event Hubs: Proiectați soluții MCP cu debit ridicat folosind Azure Event Hubs pentru streaming în timp real
- Aplicați Modele Enterprise: Integrați transpaturi personalizate cu infrastructura și modelele de securitate Azure existente
- Gestionați Fiabilitatea Transportului: Implementați durabilitate a mesajelor, ordonare și gestionarea erorilor pentru scenarii enterprise
- Optimizați Performanța: Proiectați soluții de transport pentru scalabilitate, latență și cerințe de debit
Cerințe pentru Transport
Cerințe de Bază din Specificația MCP (2025-11-25):
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"
Implementarea Transportului Azure Event Grid
Azure Event Grid oferă un serviciu serverless de rutare a evenimentelor ideal pentru arhitecturi MCP bazate pe evenimente. Această implementare demonstrează cum să construiți sisteme MCP scalabile, slab cuplate.
Prezentare Generală a Arhitecturii
graph TB
Client[MCP Client] --> EG[Azure Event Grid]
EG --> Server[MCP Server Function]
Server --> EG
EG --> Client
subgraph "Servicii Azure"
EG
Server
KV[Key Vault]
Monitor[Application Insights]
end
Implementare C# - Transport Event Grid
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();
}
}
Implementare TypeScript - Transport Event Grid
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]);
}
// Primire bazată pe evenimente prin Azure Functions
onMessage(handler: (message: McpMessage) => Promise<void>): void {
// Implementarea ar folosi declanșatorul Azure Functions Event Grid
// Aceasta este o interfață conceptuală pentru receptorul webhook
}
}
// Implementare Azure Functions
import { app, InvocationContext, EventGridEvent } from "@azure/functions";
app.eventGrid("mcpEventGridHandler", {
handler: async (event: EventGridEvent, context: InvocationContext) => {
try {
const mcpMessage = event.data as McpMessage;
// Procesează mesajul MCP
const response = await mcpServer.processMessage(mcpMessage);
// Trimite răspunsul prin Event Grid
await transport.sendMessage(response);
} catch (error) {
context.error("Error processing MCP message:", error);
throw error;
}
}
});
Implementare Python - Transport Event Grid
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
# Implementarea Azure Functions
import azure.functions as func
import logging
def main(event: func.EventGridEvent) -> None:
"""Azure Functions Event Grid trigger for MCP messages"""
try:
# Analizează mesajul MCP din evenimentul Event Grid
mcp_message = json.loads(event.get_body().decode('utf-8'))
# Procesează mesajul MCP
response = process_mcp_message(mcp_message)
# Trimite răspunsul înapoi prin Event Grid
# (Implementarea ar crea un nou client Event Grid)
except Exception as e:
logging.error(f"Error processing MCP Event Grid message: {e}")
raise
Implementarea Transportului Azure Event Hubs
Azure Event Hubs oferă capacități de streaming în timp real, cu debit ridicat, pentru scenarii MCP care necesită latență scăzută și volum mare de mesaje.
Prezentare Generală a Arhitecturii
graph TB
Client[MCP Client] --> EH[Azure Event Hubs]
EH --> Server[MCP Server]
Server --> EH
EH --> Client
subgraph "Funcționalități Event Hubs"
Partition[Partiționare]
Retention[Păstrarea Mesajelor]
Scaling[Scalare Automată]
end
EH --> Partition
EH --> Retention
EH --> Scaling
Implementare C# - Transport Event Hubs
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();
}
}
Implementare TypeScript - Transport Event Hubs
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);
// Actualizează punctul de control pentru livrare cel puțin o dată
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();
}
}
Implementare Python - Transport Event Hubs
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))
# Adaugă proprietăți specifice MCP
event_data.properties = {
"messageType": message.get("method", "response"),
"messageId": message.get("id"),
"timestamp": "2025-01-14T10:30:00Z" # Folosește marcajul de timp actual
}
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" # Pornește de la început
)
def _on_event_received(self, handler: Callable):
"""Internal event handler wrapper"""
async def handle_event(partition_context, event):
try:
# Parsează mesajul MCP din evenimentul Event Hubs
message_body = event.body_as_str(encoding='UTF-8')
mcp_message = json.loads(message_body)
# Procesează mesajul MCP
await handler(mcp_message)
# Actualizează punctul de verificare pentru livrare cel puțin o dată
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()
Modele Avansate de Transport
Durabilitatea și Fiabilitatea Mesajelor
// 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;
}
});
}
}
Integrarea Securității Transportului
// 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
);
}
}
Monitorizarea și Observabilitatea Transportului
// 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;
}
}
}
Scenarii de Integrare Enterprise
Scenariul 1: Procesare MCP Distribuită
Utilizarea Azure Event Grid pentru distribuirea cererilor MCP către mai multe noduri de procesare:
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
Scenariul 2: Streaming MCP în Timp Real
Utilizarea Azure Event Hubs pentru interacțiuni MCP de frecvență ridicată:
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
Scenariul 3: Arhitectură Hibridă de Transport
Combinarea mai multor transpaturi pentru scopuri diferite:
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);
}
}
Optimizarea Performanței
Gruparea Mesajelor pentru Event Grid
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);
}
}
}
Strategia de Partiționare pentru Event Hubs
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 });
}
}
Testarea Transpaturilor Personalizate
Testare Unită cu Double-uri de Test
[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
);
}
Testare de Integrare cu Azure Test Containers
[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"));
}
Cele Mai Bune Practici și Linii Ghid
Principii în Design-ul Transportului
- Idempotentă: Asigurați-vă că procesarea mesajelor este idempotentă pentru a gestiona duplicatele
- Gestionarea Erorilor: Implementați tratare cuprinzătoare a erorilor și cozi pentru mesaje nereușite
- Monitorizare: Adăugați telemetrie detaliată și verificări de sănătate
- Securitate: Folosiți identități gestionate și acces cu privilegiu minim
- Performanță: Proiectați pentru cerințele specifice de latență și debit
Recomandări Specifice Azure
- Folosiți Identitate Gestionată: Evitați șirurile de conexiune în producție
- Implementați Circuit Breakers: Protejați-vă împotriva indisponibilităților serviciilor Azure
- Monitorizați Costurile: Urmăriți volumul mesajelor și costurile de procesare
- Planificați pentru Scalare: Proiectați devreme strategiile de partiționare și scalare
- Testați Amănunțit: Utilizați Azure DevTest Labs pentru teste cuprinzătoare
Concluzie
Transpaturile personalizate MCP activează scenarii enterprise puternice folosind serviciile de mesagerie Azure. Prin implementarea transporturilor Event Grid sau Event Hubs, puteți construi soluții MCP scalabile și fiabile, care se integrează perfect cu infrastructura Azure existentă.
Exemplele prezentate demonstrează modele gata de producție pentru implementarea transporturilor personalizate, respectând conformitatea protocolului MCP și bunele practici Azure.
Resurse adiționale
- Specificatia MCP 2025-11-25
- Documentație Azure Event Grid
- Documentație Azure Event Hubs
- Azure Functions Event Grid Trigger
- Azure SDK pentru .NET
- Azure SDK pentru TypeScript
- Azure SDK pentru Python
Acest ghid se concentrează pe modele practice de implementare pentru sisteme MCP de producție. Validați întotdeauna implementările de transport în funcție de cerințele specifice și limitele serviciilor Azure. Standard Curent: Acest ghid reflectă cerințele de transport din Specificatia MCP 2025-11-25 și modele avansate de transport pentru medii enterprise.
Ce urmează
Declinare a responsabilității: Acest document a fost tradus folosind serviciul de traducere AI Co-op Translator. În timp ce ne străduim pentru acuratețe, vă rugăm să rețineți că traducerile automate pot conține erori sau inexactități. Documentul original în limba sa nativă trebuie considerat sursa autorizată. Pentru informații critice, se recomandă traducerea profesională realizată de un om. Nu ne asumăm responsabilitatea pentru eventualele neînțelegeri sau interpretări greșite care decurg din utilizarea acestei traduceri.