33 KiB
MCP Прилагођени Транспорти - Напредни Водич за Имплементацију
Протокол Контекста Модела (MCP) пружа флексибилност у транспортним механизмима, омогућавајући прилагођене имплементације за специјализована предузећа окружења. Овај напредни водич истражује прилагођене транспортне имплементације користећи Azure Event Grid и Azure Event Hubs као практичне примере за изградњу скалабилних, cloud-native MCP решења.
Увод
Док стандардни транспорти MCP-а (stdio и HTTP streaming) покривају већину случајева употребе, предузећа окружења често захтевају специјализоване транспортне механизме за побољшану скалабилност, поузданост и интеграцију са постојећом cloud инфраструктуром. Прилагођени транспорти омогућавају MCP-у да искористи cloud-native услуге порука за асинхрони комуникацију, догађајно вођене архитектуре и распределену обраду.
Ова лекција истражује напредне транспортне имплементације базиране на најновијој MCP спецификацији (2025-11-25), Azure услугама за поруке и утврђеним предузетничким интеграционим обрасцима.
Архитектура MCP Транспорта
Из MCP Спецификације (2025-11-25):
- Стандардни транспорти: stdio (препоручено), HTTP streaming (за удаљене сценарије)
- Прилагођени транспорти: Било који транспорт који имплементира MCP протокол размене порука
- Формат поруке: JSON-RPC 2.0 са MCP-специфичним проширењима
- Двосмерна комуникација: Потребна је пунодуплекс комуникација за обавештења и одговоре
Циљеви Учења
На крају ове напредне лекције, бићете у могућности да:
- Разумете Захтеве за Прилагођене Транспорте: Имплементирате MCP протокол преко било ког транспортног слоја уз одржавање усаглашености
- Изградите Azure Event Grid Транспорт: Креирате догађајно вођене MCP сервере користећи Azure Event Grid за serverless скалабилност
- Имплементирате Azure Event Hubs Транспорт: Дизајнирате високопропусна MCP решења користећи Azure Event Hubs за стреаминг у реалном времену
- Примените Предузетничке Обрасце: Интегришете прилагођене транспорте са постојећом Azure инфраструктуром и моделима безбедности
- Решавате Питања Поузданости Транспорта: Имплементирате издржљивост порука, поредак и руковање грешкама у предузетничким сценаријима
- Оптимизујете Перформансе: Дизајнирате транспортна решења за скалу, латенцију и захтеве пропусности
Захтеви за Транспорт
Основни Захтеви из 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"
Имплементација Azure Event Grid Транспорта
Azure Event Grid пружа серверлес услугу усмеравања догађаја идеалну за догађајно вођене MCP архитектуре. Ова имплементација показује како изградити скалабилне, лабаво повезане MCP системе.
Преглед Архитектуре
graph TB
Client[MCP Клијент] --> EG[Azure Event Grid]
EG --> Server[MCP Серверска Функција]
Server --> EG
EG --> Client
subgraph "Azure Услуге"
EG
Server
KV[Кеy Валут]
Monitor[Аппликатион Инсигхтс]
end
C# Имплементација - 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();
}
}
TypeScript Имплементација - 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]);
}
// Пријем заснован на догађајима преко Azure Functions
onMessage(handler: (message: McpMessage) => Promise<void>): void {
// Имплементација би користила Azure Functions Event Grid тригер
// Ово је концептуални интерфејс за прималац вебхука
}
}
// 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;
// Обрада MCP поруке
const response = await mcpServer.processMessage(mcpMessage);
// Слање одговора преко Event Grid
await transport.sendMessage(response);
} catch (error) {
context.error("Error processing MCP message:", error);
throw error;
}
}
});
Python Имплементација - 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
# Имплементација Azure функција
import azure.functions as func
import logging
def main(event: func.EventGridEvent) -> None:
"""Azure Functions Event Grid trigger for MCP messages"""
try:
# Парсирај MCP поруку из Event Grid догађаја
mcp_message = json.loads(event.get_body().decode('utf-8'))
# Обработи MCP поруку
response = process_mcp_message(mcp_message)
# Пошаљи одговор назад преко Event Grid-а
# (Имплементација би направила нови Event Grid клијент)
except Exception as e:
logging.error(f"Error processing MCP Event Grid message: {e}")
raise
Имплементација Azure Event Hubs Транспорта
Azure Event Hubs пружа високопропусни, стриминг у реалном времену за MCP сценарије који захтевају малу латенцију и велики обим порука.
Преглед Архитектуре
graph TB
Client[MCP Клијент] --> EH[Azure Event Hubs]
EH --> Server[MCP Сервер]
Server --> EH
EH --> Client
subgraph "Карактеристике Event Hubs"
Partition[Партиционисање]
Retention[Чување Порука]
Scaling[Аутоматско Помињање]
end
EH --> Partition
EH --> Retention
EH --> Scaling
C# Имплементација - 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();
}
}
TypeScript Имплементација - 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);
// Ажурирај контролну тачку за доставу најмање једном
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 Имплементација - 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))
# Додај својства специфична за MCP
event_data.properties = {
"messageType": message.get("method", "response"),
"messageId": message.get("id"),
"timestamp": "2025-01-14T10:30:00Z" # Користи стварно временско обележје
}
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" # Почни од почетка
)
def _on_event_received(self, handler: Callable):
"""Internal event handler wrapper"""
async def handle_event(partition_context, event):
try:
# Парсирај MCP поруку из Event Hubs догађаја
message_body = event.body_as_str(encoding='UTF-8')
mcp_message = json.loads(message_body)
# Обработи MCP поруку
await handler(mcp_message)
# Ажурирај контролну тачку за доставу бар једном
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()
Напредни Транспортни Образци
Издржљивост и Поузданост Порука
// 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;
}
});
}
}
Интеграција Безбедности Транспорта
// 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
);
}
}
Надгледање и Посматрачност Транспорта
// 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;
}
}
}
Сценарији Интеграције у Предузећу
Сценарио 1: Распоредељена MCP Обрада
Коришћење Azure Event Grid-а за расподелу MCP захтева преко више обрадних нодова:
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
Сценарио 2: MCP Стреаминг у реалном времену
Коришћење Azure Event Hubs за MCP интеракције високе фреквенције:
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
Сценарио 3: Хибридна Архитектура Транспорта
Комбиновање више транспорта за различите случајеве употребе:
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);
}
}
Оптимизација Перформанси
Пакетирање Порука за 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);
}
}
}
Стратегија Партиционисања за 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 });
}
}
Тестирање Прилагођених Транспорта
Јединично Тестирање уз Тест Даблс
[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
);
}
Интеграционо Тестирање са 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"));
}
Најбоље Практике и Упутства
Принципи Дизајна Транспорта
- Идемпотентност: Осигурајте да је обрада порука идемпотентна да бисте управљали дупликатима
- Руковање Грешкама: Имплементирајте свеобухватно руковање грешкама и редове неуспелих порука
- Надгледање: Додајте детаљну телеметрију и здравствене провере
- Безбедност: Користите управљане идентитете и приступ по принципу најмањих привилегија
- Перформансе: Дизајнирајте по специфичним захтевима за латенцију и пропусност
Преporuke Специфичне за Azure
- Користите Управљани Идентитет: Избегавајте конекционе стрингове у продукцији
- Имплементирајте Прекидаче Круга: Заштитите се од прекида услуга Azure
- Пратите Трошкове: Пратите обим порука и трошкове обраде
- Планирајте за Скалабилност: Рано дизајнирајте стратегије партиционисања и скалирања
- Тестирајте Темљно: Користите Azure DevTest Labs за свеобухватно тестирање
Закључак
Прилагођени MCP транспорти омогућавају моћне предузетничке сценарије коришћењем Azure услуга за поруке. Имплементирањем Event Grid или Event Hubs транспорта, можете изградити скалабилна, поуздана MCP решења која се беспрекорно интегришу са постојећом Azure инфраструктуром.
Примери који су дати илуструју производне образце за имплементацију прилагођених транспорти уз одржавање усаглашености са MCP протоколом и најбољим практикама Azure-а.
Додатни Ресурси
- MCP Спецификација 2025-11-25
- Azure Event Grid Документација
- Azure Event Hubs Документација
- Azure Functions Event Grid Trigger
- Azure SDK за .NET
- Azure SDK за TypeScript
- Azure SDK за Python
Овај водич се фокусира на практичне обрасце имплементације за производне MCP системе. Увек потврђујте транспортне имплементације у складу са својим специфичним захтевима и ограничењима Azure услуга. Тренутни Стандард: Овај водич одражава MCP Спецификацију 2025-11-25 захтеве за транспорт и напредне транспортне обрасце за предузетничка окружења.
Шта Следи
Изјава о одрицању одговорности: Овај документ је преведен коришћењем услуге за аутоматски превод Co-op Translator. Иако тежимо тачности, имајте у виду да аутоматски преводи могу садржати грешке или нетачности. Оригинални документ на његовом изворном језику треба сматрати ауторитативним извором. За критичне информације препоручује се професионални људски превод. Нисмо одговорни за било каква неспоразума или погрешна тумачења која произилазе из коришћења овог превода.