Транспортные протоколы

Транспорт — это способ связи. В этой теме изучим stdio, SSE и WebSocket транспорты для MCP, их преимущества, недостатки и сценарии использования.

#1. Обзор транспортов MCP

MCP поддерживает три основных транспорта:

Транспорт	Направление	Сценарий	Сложность
stdio	Двустороннее	Локальная разработка, desktop	Низкая
SSE	Однонаправленное (сервер→клиент) + HTTP POST	Удалённые серверы	Средняя
WebSocket	Двустороннее	Realtime, масштабируемые системы	Высокая

#Сравнительная таблица

Критерий	stdio	SSE	WebSocket
Локальное использование	✅ Отлично	⚠️ Работает	⚠️ Работает
Удалённое использование	❌ Не работает	✅ Хорошо	✅ Отлично
Двусторонняя связь	✅ Да	⚠️ Частично	✅ Да
Поддержка уведомлений	❌ Нет	✅ Да	✅ Да
Сложность настройки	Низкая	Средняя	Высокая
Производительность	Средняя	Хорошая	Отличная
Брандмауэры	Н/Д	✅ Проходят	⚠️ Могут блокировать

#2. stdio Transport

#Архитектура

┌─────────────────┐                         ┌─────────────────┐
│   MCP Client    │                         │   MCP Server    │
│     (Host)      │                         │    (Process)    │
└────────┬────────┘                         └────────┬────────┘
         │                                           │
         │  stdout (чтение JSON-RPC ответов)         │
         │<──────────────────────────────────────────│
         │                                           │
         │  stdin (запись JSON-RPC запросов)         │
         │──────────────────────────────────────────>│
         │                                           │
         │  stderr (логи сервера)                    │
         │<──────────────────────────────────────────│
         │                                           │

#Конфигурация для Claude Desktop

{
  "mcpServers": {
    "local-server": {
      "command": "poetry",
      "args": ["run", "python", "src/server.py"],
      "cwd": "/path/to/project",
      "env": {
        "DEBUG": "true",
        "API_KEY": "secret"
      }
    }
  }
}

#Реализация сервера

#!/usr/bin/env python3
"""MCP сервер со stdio транспортом."""

from mcp.server.fastmcp import FastMCP
import logging
import sys

# Логирование в stderr, чтобы не ломать stdout
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    stream=sys.stderr
)

mcp = FastMCP("stdio Demo")


@mcp.tool()
def hello(name: str) -> str:
    """Приветствие."""
    return f"Hello, {name}!"


if __name__ == "__main__":
    # run() по умолчанию использует stdio
    mcp.run()

#Преимущества stdio

✅ Простота: Не требует сетевого стека
✅ Безопасность: Нет сетевого интерфейса
✅ Автоматический жизненный цикл: Сервер живёт, пока нужен клиенту
✅ Отладка: Легко запустить в терминале

#Недостатки stdio

❌ Только локально: Нельзя подключить удалённо
❌ Нет уведомлений: Сервер не может отправить push-уведомление
❌ Один клиент: Только одно подключение одновременно

#3. SSE (Server-Sent Events) Transport

#Архитектура

┌─────────────────┐                         ┌─────────────────┐
│   MCP Client    │                         │   MCP Server    │
│     (Host)      │                         │    (FastAPI)    │
└────────┬────────┘                         └────────┬────────┘
         │                                           │
         │  GET /sse (SSE connection для ответов)    │
         │──────────────────────────────────────────>│
         │                                           │
         │  event: endpoint                          │
         │  data: /messages?session_id=abc123        │
         │<──────────────────────────────────────────│
         │                                           │
         │  POST /messages?session_id=abc123         │
         │  (JSON-RPC запросы через HTTP)            │
         │──────────────────────────────────────────>│
         │                                           │
         │  SSE events (JSON-RPC ответы)             │
         │<──────────────────────────────────────────│
         │                                           │

#Реализация сервера на FastAPI

#!/usr/bin/env python3
"""MCP сервер с SSE транспортом."""

from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from fastapi import FastAPI, Request
from fastapi.responses import Response
from starlette.types import Receive, Send
import uvicorn

# Создаём MCP сервер
mcp = FastMCP("SSE Demo")


@mcp.tool()
def hello(name: str) -> str:
    """Приветствие."""
    return f"Hello, {name} from SSE server!"


# Создаём SSE транспорт
sse = SseServerTransport("/messages")

# Получаем ASGI приложение
app = FastAPI()


@app.get("/sse")
async def sse_endpoint(request: Request):
    """SSE endpoint для подключения клиентов."""
    async with sse.connect_sse(request) as streams:
        await mcp.run(
            streams[0],  # read stream
            streams[1],  # write stream
            mcp.create_initialization_options()
        )


@app.post("/messages")
async def messages_endpoint(request: Request):
    """Endpoint для получения запросов от клиентов."""
    await sse.handle_post_message(request.scope, request.receive, request._send)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

#Упрощённая реализация с mcp[cli]

#!/usr/bin/env python3
"""Упрощённый SSE сервер."""

from mcp.server.fastmcp import FastMCP
import asyncio

mcp = FastMCP("Simple SSE Demo")


@mcp.tool()
def get_time() -> str:
    """Текущее время."""
    from datetime import datetime
    return datetime.now().isoformat()


if __name__ == "__main__":
    # Запуск с SSE транспортом
    mcp.run(transport="sse")

#Конфигурация клиента

{
  "mcpServers": {
    "remote-sse-server": {
      "url": "http://example.com:8000/sse"
    }
  }
}

#Преимущества SSE

✅ HTTP-совместимость: Работает через стандартные порты (80/443)
✅ Простота реализации: Легче WebSocket
✅ Поддержка уведомлений: Сервер может отправлять push-уведомления
✅ Брандмауэры: Проходит через большинство брандмауэров

#Недостатки SSE

❌ Однонаправленность: Для запросов нужен отдельный HTTP endpoint
❌ Нет бинарных данных: Только текст
❌ HTTP/1.1: Менее эффективен, чем WebSocket

#4. WebSocket Transport

#Архитектура

┌─────────────────┐                         ┌─────────────────┐
│   MCP Client    │                         │   MCP Server    │
│     (Host)      │                         │   (WebSocket)   │
└────────┬────────┘                         └────────┬────────┘
         │                                           │
         │  WebSocket Connect /ws                    │
         │──────────────────────────────────────────>│
         │                                           │
         │  <двусторонний обмен сообщениями>         │
         │<─────────────────────────────────────────>│
         │                                           │

#Реализация сервера

#!/usr/bin/env python3
"""MCP сервер с WebSocket транспортом."""

from mcp.server.fastmcp import FastMCP
from mcp.server.websocket import WebSocketServerTransport
import asyncio
import websockets
import json

mcp = FastMCP("WebSocket Demo")


@mcp.tool()
def echo(message: str) -> str:
    """Эхо сообщение."""
    return f"Echo: {message}"


@mcp.tool()
def add(a: int, b: int) -> int:
    """Сложение чисел."""
    return a + b


async def handler(websocket):
    """Обработчик WebSocket подключений."""
    transport = WebSocketServerTransport()
    
    async with transport.connect(websocket) as streams:
        await mcp.run(
            streams[0],  # read stream
            streams[1],  # write stream
            mcp.create_initialization_options()
        )


async def main():
    """Запуск сервера."""
    async with websockets.serve(handler, "localhost", 8765):
        print("MCP WebSocket server running on ws://localhost:8765")
        await asyncio.Future()  # run forever


if __name__ == "__main__":
    asyncio.run(main())

#Реализация с FastAPI + WebSockets

#!/usr/bin/env python3
"""MCP сервер с FastAPI + WebSocket."""

from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI, WebSocket
from mcp.server.websocket import WebSocketServerTransport
import asyncio

mcp = FastMCP("FastAPI WebSocket Demo")


@mcp.tool()
def search(query: str) -> list:
    """Поиск."""
    return [{"title": f"Result for: {query}", "url": "https://example.com"}]


app = FastAPI()
transport = WebSocketServerTransport()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket endpoint для MCP."""
    await websocket.accept()
    
    async with transport.connect(websocket) as streams:
        await mcp.run(
            streams[0],
            streams[1],
            mcp.create_initialization_options()
        )


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8765)

#Преимущества WebSocket

✅ Двусторонняя связь: Полноценное duplex соединение
✅ Низкие задержки: Минимальный overhead
✅ Бинарные данные: Поддержка бинарных сообщений
✅ Эффективность: Лучше производительность при высокой нагрузке

#Недостатки WebSocket

❌ Сложность: Требует WebSocket сервер
❌ Брандмауэры: Могут блокировать некоторые сети
❌ Прокси: Не все прокси поддерживают WebSocket

#5. Выбор транспорта: Decision Matrix

#Вопросы для выбора

1. Где будет работать сервер?
   ├── Локально (на машине пользователя)
   │   └── → stdio
   │
   └── Удалённо (в облаке/на сервере)
       └── → Переходим к вопросу 2

2. Нужны ли push-уведомления от сервера?
   ├── Нет
   │   └── → HTTP POST (простой REST)
   │
   └── Да
       └── → Переходим к вопросу 3

3. Какая сеть между клиентом и сервером?
   ├── Корпоративная с строгими брандмауэрами
   │   └── → SSE (работает через HTTP/HTTPS)
   │
   └── Контролируемая/открытая сеть
       └── → WebSocket (лучшая производительность)

#Сценарии использования

Сценарий	Рекомендуемый транспорт	Обоснование
Desktop приложение (Claude Desktop)	stdio	Простота, безопасность
Локальная разработка	stdio	Быстрая отладка
Облачный сервер (один клиент)	SSE	Простота реализации
Облачный сервер (много клиентов)	WebSocket	Масштабируемость
Корпоративная сеть	SSE	Проходит брандмауэры
Realtime приложение	WebSocket	Низкие задержки
Мобильное приложение	WebSocket	Эффективность

#6. Продвинутые темы

#Мультиплексирование подключений

#!/usr/bin/env python3
"""SSE сервер с поддержкой нескольких клиентов."""

from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from fastapi import FastAPI, Request
import uuid

mcp = FastMCP("Multi-Client SSE")


@mcp.tool()
def get_server_info() -> str:
    """Информация о сервере."""
    return f"Server running with {len(sse.sessions)} active sessions"


app = FastAPI()
sse = SseServerTransport("/messages")


@app.get("/sse")
async def sse_connect(request: Request):
    """Подключение нового клиента."""
    session_id = str(uuid.uuid4())
    async with sse.connect_sse(request, session_id) as streams:
        await mcp.run(
            streams[0],
            streams[1],
            mcp.create_initialization_options()
        )


@app.post("/messages/{session_id}")
async def message_handler(session_id: str, request: Request):
    """Обработка сообщений от клиентов."""
    await sse.handle_post_message(session_id, request.scope, request.receive, request._send)

#Graceful Shutdown

import signal
import asyncio


class GracefulShutdown:
    """Обработчик корректного завершения."""
    
    def __init__(self):
        self.shutdown_event = asyncio.Event()
        
    def setup(self):
        """Настройка обработчиков сигналов."""
        for sig in (signal.SIGTERM, signal.SIGINT):
            signal.signal(sig, self._signal_handler)
    
    def _signal_handler(self, signum, frame):
        """Обработчик сигналов."""
        print(f"\nReceived signal {signum}, shutting down gracefully...")
        self.shutdown_event.set()
    
    async def wait_for_shutdown(self):
        """Ожидание сигнала завершения."""
        await self.shutdown_event.wait()


# Использование
async def main():
    shutdown = GracefulShutdown()
    shutdown.setup()
    
    # Запуск сервера
    server_task = asyncio.create_task(start_server())
    
    # Ожидание завершения
    await shutdown.wait_for_shutdown()
    
    # Корректное завершение
    server_task.cancel()
    await asyncio.gather(server_task, return_exceptions=True)


asyncio.run(main())

#Health Check endpoint

from fastapi import FastAPI, HTTPException
from datetime import datetime

app = FastAPI()


@app.get("/health")
async def health_check():
    """Проверка здоровья сервера."""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0"
    }


@app.get("/ready")
async def readiness_check():
    """Проверка готовности обрабатывать запросы."""
    # Проверка подключений к БД, внешним сервисам
    if not database.is_connected():
        raise HTTPException(status_code=503, detail="Database not connected")
    
    return {"status": "ready"}

#7. Отладка транспортов

#Отладка stdio

# Запуск сервера вручную
poetry run python server.py

# Отправка тестового запроса
echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}' | \
  poetry run python server.py

#Отладка SSE

# Просмотр SSE событий
curl -N http://localhost:8000/sse

# Отправка POST запроса
curl -X POST http://localhost:8000/messages \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}'

#Отладка WebSocket

# Использование websocat (установить: brew install websocat)
websocat ws://localhost:8765

# Отправка сообщения
echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}' | \
  websocat ws://localhost:8765

#Ключевые выводы

Транспорт	Когда использовать
stdio	Локальная разработка, desktop приложения
SSE	Удалённые серверы, корпоративные сети, простые сценарии
WebSocket	Realtime приложения, высокая нагрузка, контролируемые сети

Следующая тема: Интеграция с LLM — подключение MCP к Claude, GPT, локальным моделям.

Транспорт

Направление

Сценарий

Сложность

stdio

Двустороннее

Локальная разработка, desktop

Низкая

SSE

Однонаправленное (сервер→клиент) + HTTP POST

Удалённые серверы

Средняя

WebSocket

Двустороннее

Realtime, масштабируемые системы

Высокая

Критерий

stdio

SSE

WebSocket

Локальное использование

✅ Отлично

⚠️ Работает

Удалённое использование

❌ Не работает

✅ Хорошо

✅ Отлично

Двусторонняя связь

✅ Да

⚠️ Частично

✅ Да

Поддержка уведомлений

❌ Нет

✅ Да

Сложность настройки

Низкая

Средняя

Высокая

Производительность

Средняя

Хорошая

Отличная

Брандмауэры

Н/Д

✅ Проходят

⚠️ Могут блокировать

┌─────────────────┐ ┌─────────────────┐ │ MCP Client │ │ MCP Server │ │ (Host) │ │ (Process) │ └────────┬────────┘ └────────┬────────┘ │ │ │ stdout (чтение JSON-RPC ответов) │ │<──────────────────────────────────────────│ │ │ │ stdin (запись JSON-RPC запросов) │ │──────────────────────────────────────────>│ │ │ │ stderr (логи сервера) │ │<──────────────────────────────────────────│ │ │

┌─────────────────┐ ┌─────────────────┐ │ MCP Client │ │ MCP Server │ │ (Host) │ │ (FastAPI) │ └────────┬────────┘ └────────┬────────┘ │ │ │ GET /sse (SSE connection для ответов) │ │──────────────────────────────────────────>│ │ │ │ event: endpoint │ │ data: /messages?session_id=abc123 │ │<──────────────────────────────────────────│ │ │ │ POST /messages?session_id=abc123 │ │ (JSON-RPC запросы через HTTP) │ │──────────────────────────────────────────>│ │ │ │ SSE events (JSON-RPC ответы) │ │<──────────────────────────────────────────│ │ │

┌─────────────────┐ ┌─────────────────┐ │ MCP Client │ │ MCP Server │ │ (Host) │ │ (WebSocket) │ └────────┬────────┘ └────────┬────────┘ │ │ │ WebSocket Connect /ws │ │──────────────────────────────────────────>│ │ │ │ <двусторонний обмен сообщениями> │ │<─────────────────────────────────────────>│ │ │

1. Где будет работать сервер? ├── Локально (на машине пользователя) │ └── → stdio │ └── Удалённо (в облаке/на сервере) └── → Переходим к вопросу 2 2. Нужны ли push-уведомления от сервера? ├── Нет │ └── → HTTP POST (простой REST) │ └── Да └── → Переходим к вопросу 3 3. Какая сеть между клиентом и сервером? ├── Корпоративная с строгими брандмауэрами │ └── → SSE (работает через HTTP/HTTPS) │ └── Контролируемая/открытая сеть └── → WebSocket (лучшая производительность)

Сценарий

Рекомендуемый транспорт

Обоснование

Desktop приложение (Claude Desktop)

stdio

Простота, безопасность

Локальная разработка

stdio

Быстрая отладка

Облачный сервер (один клиент)

SSE

Простота реализации

Облачный сервер (много клиентов)

WebSocket

Масштабируемость

Корпоративная сеть

SSE

Проходит брандмауэры

Realtime приложение

WebSocket

Низкие задержки

Мобильное приложение

WebSocket

Эффективность

Транспорт

Когда использовать

stdio

Локальная разработка, desktop приложения

SSE

Удалённые серверы, корпоративные сети, простые сценарии

WebSocket

Realtime приложения, высокая нагрузка, контролируемые сети