stdio, SSE, WebSocket — выбор транспорта для разных сценариев использования.
Транспорт — это способ связи. В этой теме изучим stdio, SSE и WebSocket транспорты для MCP, их преимущества, недостатки и сценарии использования.
MCP поддерживает три основных транспорта:
| Транспорт | Направление | Сценарий | Сложность |
|---|---|---|---|
| stdio | Двустороннее | Локальная разработка, desktop | Низкая |
| SSE | Однонаправленное (сервер→клиент) + HTTP POST | Удалённые серверы | Средняя |
| WebSocket | Двустороннее | Realtime, масштабируемые системы | Высокая |
| Критерий | stdio | SSE | WebSocket |
|---|---|---|---|
| Локальное использование | ✅ Отлично | ⚠️ Работает | ⚠️ Работает |
| Удалённое использование | ❌ Не работает | ✅ Хорошо | ✅ Отлично |
| Двусторонняя связь | ✅ Да | ⚠️ Частично | ✅ Да |
| Поддержка уведомлений | ❌ Нет | ✅ Да | ✅ Да |
| Сложность настройки | Низкая | Средняя | Высокая |
| Производительность | Средняя | Хорошая | Отличная |
| Брандмауэры | Н/Д | ✅ Проходят | ⚠️ Могут блокировать |
┌─────────────────┐ ┌─────────────────┐
│ MCP Client │ │ MCP Server │
│ (Host) │ │ (Process) │
└────────┬────────┘ └────────┬────────┘
│ │
│ stdout (чтение JSON-RPC ответов) │
│<──────────────────────────────────────────│
│ │
│ stdin (запись JSON-RPC запросов) │
│──────────────────────────────────────────>│
│ │
│ stderr (логи сервера) │
│<──────────────────────────────────────────│
│ │
{
"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()┌─────────────────┐ ┌─────────────────┐
│ 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 ответы) │
│<──────────────────────────────────────────│
│ │
#!/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)#!/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"
}
}
}┌─────────────────┐ ┌─────────────────┐
│ 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())#!/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)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 | Эффективность |
#!/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)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())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"}# Запуск сервера вручную
poetry run python server.py
# Отправка тестового запроса
echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}' | \
poetry run python server.py# Просмотр 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": {}}'# Использование 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, локальным моделям.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.