28 KiB
MCP Vlastní Transporty - Pokročilý Průvodce Implementací
Model Context Protocol (MCP) poskytuje flexibilitu v transportních mechanismech, umožňující vlastní implementace pro specializované podnikové prostředí. Tento pokročilý průvodce zkoumá vlastní implementace transportů s využitím Azure Event Grid a Azure Event Hubs jako praktické příklady pro budování škálovatelných, cloud-native MCP řešení.
Úvod
Zatímco standardní transporty MCP (stdio a HTTP streaming) slouží většině případů použití, podnikové prostředí často vyžadují specializované transportní mechanismy pro zlepšení škálovatelnosti, spolehlivosti a integraci s existující cloudovou infrastrukturou. Vlastní transporty umožňují MCP využívat cloud-native messaging služby pro asynchronní komunikaci, event-driven architektury a distribuované zpracování.
Tato lekce zkoumá pokročilé implementace transportů založené na nejnovější specifikaci MCP (2025-11-25), Azure messaging službách a zavedených podnikovém integračních vzorech.
Architektura MCP Transportů
Z MCP Specifikace (2025-11-25):
- Standardní Transporty: stdio (doporučeno), HTTP streaming (pro vzdálené scénáře)
- Vlastní Transporty: Jakýkoli transport, který implementuje protokol výměny zpráv MCP
- Formát Zpráv: JSON-RPC 2.0 s rozšířeními specifickými pro MCP
- Obousměrná Komunikace: Plně duplexní komunikace požadována pro notifikace a odpovědi
Výukové Cíle
Po skončení této pokročilé lekce budete schopni:
- Porozumět Požadavkům na Vlastní Transporty: Implementovat MCP protokol přes libovolnou transportní vrstvu při zachování souladu
- Vytvořit Azure Event Grid Transport: Vytvářet event-driven MCP servery pomocí Azure Event Grid pro serverless škálovatelnost
- Implementovat Azure Event Hubs Transport: Navrhnout MCP řešení s vysokou propustností využívající Azure Event Hubs pro real-time streaming
- Aplikovat Podnikové Vzory: Integrovat vlastní transporty se stávající Azure infrastrukturou a bezpečnostními modely
- Zpracovat Spolehlivost Transportu: Implementovat trvanlivost zpráv, pořadí a zpracování chyb pro podnikové scénáře
- Optimalizovat Výkon: Navrhovat transportní řešení podle požadavků škálovatelnosti, latence a propustnosti
Požadavky na Transport
Jádrové Požadavky ze Specifikace 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"
Implementace Transportu Azure Event Grid
Azure Event Grid poskytuje serverless službu směrování událostí ideální pro event-driven MCP architektury. Tato implementace demonstruje, jak vybudovat škálovatelné, volně spojované MCP systémy.
Přehled Architektury
graph TB
Client[MCP Klient] --> EG[Azure Event Grid]
EG --> Server[MCP Server Funkce]
Server --> EG
EG --> Client
subgraph "Azure Služby"
EG
Server
KV[Key Vault]
Monitor[Application Insights]
end
Implementace v C# - Event Grid Transport
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();
}
}
Implementace v TypeScript - Event Grid Transport
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]);
}
// Příjem řízený událostmi pomocí Azure Functions
onMessage(handler: (message: McpMessage) => Promise<void>): void {
// Implementace by použila trigger Azure Functions Event Grid
// Toto je koncepční rozhraní pro přijímač webhooku
}
}
// Implementace 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;
// Zpracovat zprávu MCP
const response = await mcpServer.processMessage(mcpMessage);
// Odeslat odpověď přes Event Grid
await transport.sendMessage(response);
} catch (error) {
context.error("Error processing MCP message:", error);
throw error;
}
}
});
Implementace v Python - Event Grid Transport
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
# Implementace Azure Functions
import azure.functions as func
import logging
def main(event: func.EventGridEvent) -> None:
"""Azure Functions Event Grid trigger for MCP messages"""
try:
# Analyzovat zprávu MCP z události Event Grid
mcp_message = json.loads(event.get_body().decode('utf-8'))
# Zpracovat zprávu MCP
response = process_mcp_message(mcp_message)
# Odeslat odpověď zpět přes Event Grid
# (Implementace by vytvořila nového klienta Event Grid)
except Exception as e:
logging.error(f"Error processing MCP Event Grid message: {e}")
raise
Implementace Transportu Azure Event Hubs
Azure Event Hubs poskytují vysokopropustné schopnosti real-time streamingu pro MCP scénáře vyžadující nízkou latenci a vysoký objem zpráv.
Přehled Architektury
graph TB
Client[MCP Klient] --> EH[Azure Event Hubs]
EH --> Server[MCP Server]
Server --> EH
EH --> Client
subgraph "Funkce Event Hubs"
Partition[Rozdělování]
Retention[Uchovávání zpráv]
Scaling[Automatické škálování]
end
EH --> Partition
EH --> Retention
EH --> Scaling
Implementace v C# - Event Hubs Transport
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();
}
}
Implementace v TypeScript - Event Hubs Transport
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);
// Aktualizovat kontrolní bod pro doručení alespoň jednou
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();
}
}
Implementace v Python - Event Hubs Transport
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))
# Přidat vlastnosti specifické pro MCP
event_data.properties = {
"messageType": message.get("method", "response"),
"messageId": message.get("id"),
"timestamp": "2025-01-14T10:30:00Z" # Použít skutečný časový údaj
}
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" # Začít od začátku
)
def _on_event_received(self, handler: Callable):
"""Internal event handler wrapper"""
async def handle_event(partition_context, event):
try:
# Analyzovat zprávu MCP z události Event Hubs
message_body = event.body_as_str(encoding='UTF-8')
mcp_message = json.loads(message_body)
# Zpracovat zprávu MCP
await handler(mcp_message)
# Aktualizovat kontrolní bod pro doručení alespoň jednou
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()
Pokročilé Vzory Transportu
Trvanlivost a Spolehlivost Zpráv
// 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;
}
});
}
}
Integrace Bezpečnosti Transportu
// 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
);
}
}
Monitorování a Observabilita Transportu
// 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;
}
}
}
Podnikové Integrační Scénáře
Scénář 1: Distribuované Zpracování MCP
Použití Azure Event Grid pro distribuci MCP požadavků přes více zpracovatelských uzlů:
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
Scénář 2: Real-time MCP Streaming
Použití Azure Event Hubs pro vysokofrekvenční MCP interakce:
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
Scénář 3: Hybridní Transportní Architektura
Kombinace více transportů pro různé případy použití:
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);
}
}
Optimalizace Výkonu
Batchování Zpráv pro 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);
}
}
}
Strategie Particionování pro 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 });
}
}
Testování Vlastních Transportů
Jednotkové Testování s Testovacími Doubly
[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
);
}
Integrační Testování s 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"));
}
Nejlepší Praktiky a Směrnice
Zásady Návrhu Transportu
- Idempotentnost: Zajistit idempotentní zpracování zpráv pro zvládnutí duplikátů
- Zpracování Chyb: Implementovat komplexní zpracování chyb a fronty mrtvých zpráv
- Monitorování: Přidat detailní telemetrii a kontroly zdraví
- Bezpečnost: Používat spravované identity a přístup s nejmenšími oprávněními
- Výkon: Navrhnout dle specifických požadavků na latenci a propustnost
Doporučení Specifická pro Azure
- Používat Spravovanou Identitu: Vyvarovat se connection stringů v produkci
- Implementovat Circuit Breakers: Ochrana proti výpadkům služeb Azure
- Monitorovat Náklady: Sledovat objem zpráv a náklady na zpracování
- Plánovat pro Škálování: Raně navrhnout strategie particionování a škálování
- Provádět Důkladné Testování: Využít Azure DevTest Labs pro komplexní testování
Závěr
Vlastní MCP transporty umožňují silné podnikové scénáře s využitím Azure messaging služeb. Implementací transportů Event Grid nebo Event Hubs můžete vybudovat škálovatelná, spolehlivá MCP řešení, která se bezproblémově integrují s existující Azure infrastrukturou.
Předložené příklady demonstrují produkčně připravené vzory pro implementaci vlastních transportů při zachování souladu s protokolem MCP a nejlepšími praktikami Azure.
Další Zdroje
- Specifikace MCP 2025-11-25
- Dokumentace Azure Event Grid
- Dokumentace Azure Event Hubs
- Azure Functions Event Grid Trigger
- Azure SDK pro .NET
- Azure SDK pro TypeScript
- Azure SDK pro Python
Tento průvodce se zaměřuje na praktické implementační vzory pro produkční MCP systémy. Vždy validujte implementace transportů podle vašich specifických požadavků a limitů služeb Azure. Současný Standard: Tento průvodce odráží Specifikaci MCP 2025-11-25 požadavky na transporty a pokročilé transportní vzory pro podnikové prostředí.
Co Dále
Prohlášení o omezení odpovědnosti: Tento dokument byl přeložen pomocí AI překladatelské služby Co-op Translator. Přestože usilujeme o co největší přesnost, mějte prosím na paměti, že automatizované překlady mohou obsahovat chyby nebo nepřesnosti. Originální dokument v jeho mateřském jazyce by měl být považován za autoritativní zdroj. Pro kritické informace se doporučuje profesionální lidský překlad. Nejsme odpovědní za jakékoli nedorozumění nebo nesprávné interpretace vzniklé použitím tohoto překladu.