Files

T

wehub-resource-sync c275ba2868 chore: import upstream snapshot with attribution

2026-07-13 13:31:35 +08:00

34 KiB

Raw Permalink Blame History

MCP Персонализирани Транспорти - Ръководство за Напреднала Имплементация

Протоколът за Моделен Контекст (MCP) предоставя гъвкавост в транспортните механизми, позволявайки персонализирани имплементации за специализирани корпоративни среди. Това напреднало ръководство разглежда имплементации на персонализирани транспорти, използвайки Azure Event Grid и Azure Event Hubs като практически примери за изграждане на мащабируеми, облачно-родени MCP решения.

Въведение

Докато стандартните транспорти на MCP (stdio и HTTP стрийминг) обслужват повечето случаи, корпоративните среди често изискват специализирани транспортни механизми за подобрена мащабируемост, надеждност и интеграция със съществуващата облачна инфраструктура. Персонализираните транспорти позволяват на MCP да използва облачно-родени съобщителни услуги за асинхронна комуникация, събитийно ориентирани архитектури и разпределена обработка.

Този урок разглежда напреднали имплементации на транспорти, базирани на последната MCP спецификация (2025-11-25), Azure съобщителни услуги и установени корпоративни интеграционни модели.

Архитектура на MCP транспорти

От MCP спецификация (2025-11-25):

Стандартни транспорти: stdio (препоръчително), HTTP стрийминг (за отдалечени сценарии)
Персонализирани транспорти: Всеки транспорт, който имплементира MCP протокола за обмен на съобщения
Формат на съобщенията: JSON-RPC 2.0 с MCP-специфични разширения
Двунасочна комуникация: Изисква се пълен дуплекс за уведомления и отговори

Обучителни Цели

Към края на този напреднал урок ще можете да:

Разберете изискванията за персонализирани транспорти: Имплементирате MCP протокола върху всеки транспортен слой, като поддържате съвместимост
Изградите Azure Event Grid транспорт: Създадете събитийно-ориентирани MCP сървъри, използващи Azure Event Grid за сървърна мащабируемост
Имплементирате 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[Key Vault]
        Monitor[Application Insights]
    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 тригер
        // Това е концептуален интерфейс за получаване на webhook
    }
}

// Имплементация на 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 Functions
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"));
}

Добри практики и насоки

Принципи за дизайн на транспорти

Идempotентност: Осигурявайте обработка на съобщения, която понася дублирани съобщения
Обработка на грешки: Имплементирайте всеобхватна обработка на грешки и опашки за мъртви писма
Мониторинг: Добавяйте подробна телеметрия и проверки на здравето
Сигурност: Използвайте управлявани идентичности и достъп с най-малко привилегии
Производителност: Проектирайте за зададените от вас изисквания за латентност и пропускателна способност

Препоръки специфични за Azure

Използвайте управлявани идентичности: Избягвайте низове за връзка в продукция
Имплементирайте прекъсвачи на вериги (Circuit Breakers): Защита срещу прекъсвания на Azure услуги
Мониторирайте разходите: Следете обема на съобщения и разходите за обработка
Планирайте за мащаб: Проектирайте стратегии за партициониране и мащабиране рано
Тествайте задълбочено: Използвайте Azure DevTest Labs за пълноценно тестване

Заключение

Персонализираните MCP транспорти позволяват мощни корпоративни сценарии, използвайки съобщителните услуги на Azure. Чрез имплементиране на Event Grid или Event Hubs транспорти можете да изградите мащабируеми, надеждни MCP решения, които се интегрират безпроблемно със съществуващата Azure инфраструктура.

Представените примери демонстрират модели, готови за продукция, за реализиране на персонализирани транспорти, като същевременно поддържат съвместимост с MCP протокола и най-добрите практики на Azure.

Допълнителни ресурси

Това ръководство е съсредоточено върху практически модели за имплементация за производствени MCP системи. Винаги валидирайте имплементациите на транспорта според вашите специфични изисквания и лимити на Azure услуги. Текущ стандарт: Това ръководство отразява изискванията за транспорти и напредналите транспортни модели за корпоративни среди от MCP Спецификация 2025-11-25.

Какво следва

6. Обществени приноси

Отказ от отговорност: Този документ е преведен с помощта на AI преводачески услуга Co-op Translator. Въпреки че се стремим към точност, моля имайте предвид, че автоматизираните преводи могат да съдържат грешки или неточности. Оригиналният документ на неговия роден език трябва да се счита за авторитетен източник. За критична информация се препоръчва професионален човешки превод. Ние не носим отговорност за каквито и да е недоразумения или неправилни тълкувания, произтичащи от използването на този превод.

34 KiB Raw Permalink Blame History Unescape Escape

MCP Персонализирани Транспорти - Ръководство за Напреднала Имплементация

Въведение

Архитектура на MCP транспорти

Обучителни Цели

Изисквания за транспорта

Основни изисквания от MCP спецификация (2025-11-25):

Имплементация на Azure Event Grid транспорт

Преглед на Архитектурата

Имплементация на C# - Event Grid транспорт

Имплементация на TypeScript - Event Grid транспорт

Имплементация на Python - Event Grid транспорт

Имплементация на Azure Event Hubs транспорт

Преглед на Архитектурата

Имплементация на C# - Event Hubs транспорт

Имплементация на TypeScript - Event Hubs транспорт

Имплементация на Python - Event Hubs транспорт

Напреднали транспортни модели

Издръжливост и надеждност на съобщенията

Интеграция на транспортна сигурност

Мониторинг и наблюдаемост на транспорта

Корпоративни интеграционни сценарии

Сценарий 1: Разпределена MCP обработка

Сценарий 2: Реално време MCP стрийминг

Сценарий 3: Хибридна транспортна архитектура

Оптимизация на производителността

Групиране на съобщения за Event Grid

Стратегия за партициониране в Event Hubs

Тестване на персонализирани транспорти

Модулно тестване с тестови двойници

Интеграционно тестване с Azure Test Containers

Добри практики и насоки

Принципи за дизайн на транспорти

Препоръки специфични за Azure

Заключение

Допълнителни ресурси

Какво следва

34 KiB

Raw Permalink Blame History