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

Понимание архитектуры MCP — ключ к созданию надёжных и масштабируемых серверов. В этой теме разберём протокол до винтика.

#1. Модель взаимодействия: Host ↔ Server

#Общая схема

┌─────────────────────────────────────────────────────────────────┐
│                         MCP Host                                │
│  ┌───────────────────────────────────────────────────────────┐  │
│  │                      LLM Engine                           │  │
│  │                    (Claude, GPT, ...)                     │  │
│  └──────────────────────────┬────────────────────────────────┘  │
│                             │                                   │
│  ┌──────────────────────────▼────────────────────────────────┐  │
│  │                    MCP Client                             │  │
│  │  ┌──────────────┬──────────────┬──────────────────────┐   │  │
│  │  │   Connection │   Message    │     Session          │   │  │
│  │  │   Manager    │   Router     │     Manager          │   │  │
│  │  └──────────────┴──────────────┴──────────────────────┘   │  │
│  └──────────────────────────┬────────────────────────────────┘  │
└─────────────────────────────┼───────────────────────────────────┘
                              │ MCP Protocol
                              │ (JSON-RPC 2.0 over Transport)
┌─────────────────────────────▼───────────────────────────────────┐
│                        MCP Server                               │
│  ┌───────────────────────────────────────────────────────────┐  │
│  │                   Protocol Layer                          │  │
│  │         (Request Handler, Response Builder)               │  │
│  └──────────────────────────┬────────────────────────────────┘  │
│                             │                                   │
│  ┌──────────────────────────▼────────────────────────────────┐  │
│  │                   Capability Providers                    │  │
│  │  ┌──────────────┬──────────────┬──────────────────────┐   │  │
│  │  │  Resources   │    Tools     │      Prompts         │   │  │
│  │  │  Provider    │   Provider   │     Provider         │   │  │
│  │  └──────────────┴──────────────┴──────────────────────┘   │  │
│  └───────────────────────────────────────────────────────────┘  │
│                             │                                   │
│  ┌──────────────────────────▼────────────────────────────────┐  │
│  │                  Backend Integrations                     │  │
│  │     (File System, Database, External APIs, ...)           │  │
│  └───────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

#Ключевые принципы

Client-Server архитектура: Host всегда инициирует соединение
Request-Response: Client отправляет запросы, Server отвечает
Stateless протокол: каждый запрос независим (но сессия сохраняет состояние)
JSON-RPC 2.0: формат сообщений стандартизирован

#2. Протокол: JSON-RPC 2.0

MCP использует JSON-RPC 2.0 — лёгкий протокол удалённого вызова процедур.

#Формат сообщения

Запрос (Request):

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

Ответ (Response):

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "search",
        "description": "Search knowledge base",
        "inputSchema": {
          "type": "object",
          "properties": {
            "query": {"type": "string"}
          },
          "required": ["query"]
        }
      }
    ]
  }
}

Ошибка (Error):

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32601,
    "message": "Method not found",
    "data": "Unknown method: tools/invalid"
  }
}

#Поля JSON-RPC

Поле	Тип	Обязательное	Описание
`jsonrpc`	string	Да	Версия протокола, всегда `"2.0"`
`id`	integer/string	Да*	Уникальный ID запроса (отсутствует для notifications)
`method`	string	Да	Имя метода, например `tools/list`
`params`	object/array	Нет	Параметры метода
`result`	object	Да*	Результат выполнения (в ответе)
`error`	object	Да*	Ошибка выполнения (в ответе при ошибке)

* — обязательно для ответа, но не одновременно

#Типы сообщений

1. Request (Запрос)

Требует ответа с тем же id
Пример: tools/list, resources/read

2. Notification (Уведомление)

Не требует ответа (нет поля id)
Пример: notifications/initialized

3. Response (Ответ)

Содержит result или error
id соответствует запросу

#3. Жизненный цикл сессии MCP

#Последовательность инициализации

┌─────────────┐                           ┌─────────────┐
│ MCP Client  │                           │ MCP Server  │
│   (Host)    │                           │   (Server)  │
└──────┬──────┘                           └──────┬──────┘
       │                                         │
       │  1. initialize (capabilities)           │
       │────────────────────────────────────────>│
       │                                         │
       │  2. initialize response                 │
       │     (server capabilities)               │
       │<────────────────────────────────────────│
       │                                         │
       │  3. notifications/initialized           │
       │────────────────────────────────────────>│
       │                                         │
       │  ◄─── Сессия активна, запросы ───►      │
       │                                         │
       │  4. tools/list, resources/read, ...     │
       │────────────────────────────────────────>│
       │                                         │
       │  5. response with data                  │
       │<────────────────────────────────────────│
       │                                         │
       │  6. close connection                    │
       │────────────────────────────────────────>│
       │                                         │

#Этапы сессии

#Этап 1: Подключение

Client устанавливает соединение через выбранный транспорт (stdio, SSE, WebSocket).

stdio пример:

# Запуск процесса сервера
python mcp_server.py

# stdin/stdout готовы к обмену сообщениями

SSE пример:

# Client подключается к SSE endpoint
GET http://localhost:8000/sse

# Сервер отправляет event с endpoint для сообщений
event: endpoint
data: http://localhost:8000/messages?session_id=abc123

#Этап 2: Инициализация

Client отправляет initialize запрос:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "roots": {
        "listChanged": true
      }
    },
    "clientInfo": {
      "name": "Claude Desktop",
      "version": "1.0.0"
    }
  }
}

Server отвечает своими возможностями:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "resources": {
        "subscribe": true,
        "listChanged": true
      },
      "tools": {
        "listChanged": true
      },
      "prompts": {
        "listChanged": true
      }
    },
    "serverInfo": {
      "name": "my-mcp-server",
      "version": "1.0.0"
    }
  }
}

#Этап 3: Подтверждение инициализации

Client отправляет уведомление:

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

Важно: После этого момента сессия считается активной, можно отправлять запросы.

#Этап 4: Рабочий обмен

Client отправляет запросы в любом порядке:

// Запрос списка инструментов
{"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}

// Запрос ресурса
{"jsonrpc": "2.0", "id": 3, "method": "resources/read", "params": {"uri": "file:///tmp/test.txt"}}

// Вызов инструмента
{"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "search", "arguments": {"query": "MCP"}}}

Server отвечает:

// Ответ tools/list
{"jsonrpc": "2.0", "id": 2, "result": {"tools": [...]}}

// Ответ resources/read
{"jsonrpc": "2.0", "id": 3, "result": {"contents": [{"uri": "file:///tmp/test.txt", "text": "Hello"}]}}

// Ответ tools/call
{"jsonrpc": "2.0", "id": 4, "result": {"content": [{"type": "text", "text": "Search results..."}]}}

#Этап 5: Завершение

Соединение закрывается одной из сторон:

stdio: процесс сервера завершается
SSE/WebSocket: отправка close фрейма

#4. Capabilities (Возможности)

Server сообщает Client о своих возможностях при инициализации.

#Типы capabilities

#Resources Capability

{
  "resources": {
    "subscribe": true,
    "listChanged": true
  }
}

Поле	Значение
`subscribe`	Сервер поддерживает подписку на изменения ресурсов
`listChanged`	Сервер отправляет уведомления при изменении списка ресурсов

#Tools Capability

{
  "tools": {
    "listChanged": true
  }
}

Поле	Значение
`listChanged`	Сервер отправляет уведомления при изменении списка инструментов

#Prompts Capability

{
  "prompts": {
    "listChanged": true
  }
}

Поле	Значение
`listChanged`	Сервер отправляет уведомления при изменении списка промптов

#Динамические изменения

Если сервер поддерживает listChanged, он может отправлять уведомления:

{
  "jsonrpc": "2.0",
  "method": "notifications/tools/list_changed"
}

Client после получения уведомления должен запросить обновлённый список через tools/list.

#5. Методы API

#Resources API

Метод	Описание
`resources/list`	Получить список доступных ресурсов
`resources/read`	Прочитать содержимое ресурса по URI
`resources/subscribe`	Подписаться на изменения ресурса
`resources/unsubscribe`	Отписаться от изменений ресурса

Пример resources/list:

// Запрос
{"jsonrpc": "2.0", "id": 1, "method": "resources/list", "params": {}}

// Ответ
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resources": [
      {
        "uri": "file:///etc/hosts",
        "name": "Hosts file",
        "description": "System hosts file",
        "mimeType": "text/plain"
      }
    ]
  }
}

Пример resources/read:

// Запрос
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "resources/read",
  "params": {"uri": "file:///etc/hosts"}
}

// Ответ
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "contents": [
      {
        "uri": "file:///etc/hosts",
        "mimeType": "text/plain",
        "text": "127.0.0.1 localhost\n"
      }
    ]
  }
}

#Tools API

Метод	Описание
`tools/list`	Получить список доступных инструментов
`tools/call`	Вызвать инструмент с аргументами

Пример tools/list:

// Запрос
{"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}

// Ответ
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "search_database",
        "description": "Search the knowledge base",
        "inputSchema": {
          "type": "object",
          "properties": {
            "query": {
              "type": "string",
              "description": "Search query"
            },
            "limit": {
              "type": "integer",
              "description": "Max results",
              "default": 10
            }
          },
          "required": ["query"]
        }
      }
    ]
  }
}

Пример tools/call:

// Запрос
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "search_database",
    "arguments": {"query": "MCP protocol", "limit": 5}
  }
}

// Ответ
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "Found 3 results:\n1. MCP Specification...\n2. MCP Python SDK...\n3. MCP Servers Catalog..."
      }
    ],
    "isError": false
  }
}

#Prompts API

Метод	Описание
`prompts/list`	Получить список доступных промптов
`prompts/get`	Получить промпт по имени с аргументами

Пример prompts/list:

// Запрос
{"jsonrpc": "2.0", "id": 1, "method": "prompts/list", "params": {}}

// Ответ
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "prompts": [
      {
        "name": "code_review",
        "description": "Request code review",
        "arguments": [
          {
            "name": "code",
            "description": "Code to review",
            "required": true
          },
          {
            "name": "language",
            "description": "Programming language",
            "required": false
          }
        ]
      }
    ]
  }
}

Пример prompts/get:

// Запрос
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "prompts/get",
  "params": {
    "name": "code_review",
    "arguments": {
      "code": "def hello():\n    print('Hello')",
      "language": "python"
    }
  }
}

// Ответ
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "description": "Code review request",
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "Please review this Python code:\n\ndef hello():\n    print('Hello')\n\nCheck for best practices and potential issues."
        }
      }
    ]
  }
}

#6. Обработка ошибок

#Стандартные коды ошибок JSON-RPC

Код	Значение
`-32700`	Parse error — невалидный JSON
`-32600`	Invalid Request — невалидная структура запроса
`-32601`	Method not found — метод не существует
`-32602`	Invalid params — невалидные параметры
`-32603`	Internal error — внутренняя ошибка сервера

#MCP специфичные ошибки

Server может возвращать собственные коды ошибок в диапазоне −32000 до −32099:

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32001,
    "message": "Resource not found",
    "data": {
      "uri": "file:///nonexistent.txt"
    }
  }
}

#Примеры обработки ошибок

Ресурс не найден:

{
  "jsonrpc": "2.0",
  "id": 5,
  "error": {
    "code": -32001,
    "message": "Resource not found",
    "data": {"uri": "file:///missing.txt"}
  }
}

Инструмент не найден:

{
  "jsonrpc": "2.0",
  "id": 6,
  "error": {
    "code": -32601,
    "message": "Tool not found: invalid_tool"
  }
}

Невалидные параметры:

{
  "jsonrpc": "2.0",
  "id": 7,
  "error": {
    "code": -32602,
    "message": "Invalid parameters",
    "data": "Missing required field: query"
  }
}

Внутренняя ошибка сервера:

{
  "jsonrpc": "2.0",
  "id": 8,
  "error": {
    "code": -32603,
    "message": "Internal error",
    "data": "Database connection failed: timeout after 30s"
  }
}

#7. Уведомления (Notifications)

Уведомления — это сообщения без id, не требующие ответа.

#Уведомления от Server к Client

tools/list_changed:

{
  "jsonrpc": "2.0",
  "method": "notifications/tools/list_changed"
}

resources/list_changed:

{
  "jsonrpc": "2.0",
  "method": "notifications/resources/list_changed"
}

prompts/list_changed:

{
  "jsonrpc": "2.0",
  "method": "notifications/prompts/list_changed"
}

resources/updated (при подписке):

{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "uri": "file:///logs/app.log"
  }
}

#Уведомления от Client к Server

initialized:

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

roots/list_changed:

{
  "jsonrpc": "2.0",
  "method": "notifications/roots/list_changed"
}

#8. Транспорты: детали реализации

#stdio Transport

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

┌─────────────┐         ┌─────────────┐
│ MCP Client  │         │ MCP Server  │
│   (Host)    │         │  (Process)  │
└──────┬──────┘         └──────┬──────┘
       │                       │
       │  stdout (read)        │
       │<──────────────────────│
       │                       │
       │  stdin (write)        │
       │──────────────────────>│
       │                       │

Пример запуска:

// Конфиг для Claude Desktop
{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"],
      "env": {
        "API_KEY": "secret"
      }
    }
  }
}

Преимущества:

Простота (не нужен сетевой стек)
Безопасность (нет сетевого интерфейса)
Автоматический жизненный цикл (сервер живёт пока нужен)

Недостатки:

Только локальное подключение
Ограниченная производительность (текстовый протокол)

#SSE Transport

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

┌─────────────┐                         ┌─────────────┐
│ MCP Client  │                         │ MCP Server  │
│   (Host)    │                         │  (FastAPI)  │
└──────┬──────┘                         └──────┬──────┘
       │                                       │
       │  GET /sse (SSE connection)            │
       │──────────────────────────────────────>│
       │                                       │
       │  event: endpoint                      │
       │  data: /messages?session_id=abc       │
       │<──────────────────────────────────────│
       │                                       │
       │  POST /messages?session_id=abc        │
       │  (JSON-RPC requests)                  │
       │──────────────────────────────────────>│
       │                                       │
       │  SSE events (JSON-RPC responses)      │
       │<──────────────────────────────────────│
       │                                       │

Пример сервера на FastAPI:

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
import asyncio

app = FastAPI()

# Хранилище сессий
sessions = {}

@app.get("/sse")
async def sse_endpoint(request: Request):
    session_id = generate_session_id()
    sessions[session_id] = asyncio.Queue()
    
    async def event_generator():
        while True:
            if await request.is_disconnected():
                break
            message = await sessions[session_id].get()
            yield {"data": message}
    
    return EventSourceResponse(event_generator())

@app.post("/messages")
async def messages_endpoint(session_id: str, request: Request):
    body = await request.json()
    response = await process_mcp_request(body)
    # Отправка ответа через SSE очередь
    await sessions[session_id].put(json.dumps(response))
    return {"status": "ok"}

Преимущества:

Работает через HTTP/HTTPS
Поддержка брандмауэрами
Простая реализация на любом фреймворке

Недостатки:

Однонаправленная связь (сервер → клиент для ответов)
Нужен отдельный endpoint для запросов

#WebSocket Transport

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

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

Пример сервера:

import asyncio
import websockets

async def handler(websocket):
    async for message in websocket:
        request = json.loads(message)
        response = await process_mcp_request(request)
        await websocket.send(json.dumps(response))

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

Преимущества:

Двусторонняя связь
Низкие задержки
Эффективно для realtime

Недостатки:

Сложнее в настройке (WebSocket сервер)
Некоторые брандмауэры блокируют WebSocket

#9. Сессии и состояние

#Stateless протокол с состоянием сессии

MCP протокол stateless — каждый запрос независим. Но сессия может хранить состояние:

Что хранится в сессии:

Подписки на ресурсы
Контекст выполнения (например, открытые транзакции)
Кэш данных (опционально)

Пример управления сессией:

class MCPSession:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.subscriptions: set[str] = set()  # Подписки на ресурсы
        self.context: dict = {}  # Контекст выполнения
    
    def subscribe(self, uri: str):
        self.subscriptions.add(uri)
    
    def unsubscribe(self, uri: str):
        self.subscriptions.discard(uri)
    
    def cleanup(self):
        # Отписка от всех ресурсов при завершении
        self.subscriptions.clear()
        self.context.clear()

#Управление жизненным циклом сессии

Создание сессии:

async def on_connect(transport):
    session_id = generate_uuid()
    session = MCPSession(session_id)
    sessions[session_id] = session
    return session

Завершение сессии:

async def on_disconnect(session):
    await session.cleanup()
    del sessions[session.id]

#10. Best Practices архитектуры

#1. Разделение ответственности

❌ ПЛОХО: Вся логика в одном классе
class MCPServer:
    async def handle_request(self, request):
        # 500 строк кода с обработкой всего
        
✅ ХОРОШО: Разделение по ответственности
class MCPServer:
    def __init__(self):
        self.resources_handler = ResourcesHandler()
        self.tools_handler = ToolsHandler()
        self.prompts_handler = PromptsHandler()
    
    async def handle_request(self, request):
        method = request.method
        if method.startswith("resources/"):
            return await self.resources_handler.handle(request)
        elif method.startswith("tools/"):
            return await self.tools_handler.handle(request)
        elif method.startswith("prompts/"):
            return await self.prompts_handler.handle(request)

#2. Валидация входных данных

❌ ПЛОХО: Нет валидации
async def search(query: str):
    return await db.search(query)

✅ ХОРОШО: Валидация схемы и бизнес-правил
async def search(query: str, limit: int = 10):
    # Валидация схемы (JSON Schema)
    if not isinstance(query, str):
        raise ValueError("query must be a string")
    
    # Бизнес-валидация
    if len(query) < 2:
        raise ValueError("query must be at least 2 characters")
    
    if limit > 100:
        raise ValueError("limit cannot exceed 100")
    
    return await db.search(query, limit)

#3. Обработка ошибок

❌ ПЛОХО: Проглатывание ошибок
async def get_resource(uri):
    try:
        return await fetch(uri)
    except:
        return None

✅ ХОРОШО: Явная обработка с информативными ошибками
async def get_resource(uri):
    try:
        return await fetch(uri)
    except ResourceNotFoundError as e:
        raise MCPError(
            code=-32001,
            message=f"Resource not found: {uri}",
            data={"uri": uri}
        )
    except PermissionError as e:
        raise MCPError(
            code=-32003,
            message=f"Access denied: {uri}"
        )
    except Exception as e:
        logger.exception(f"Internal error fetching {uri}")
        raise MCPError(
            code=-32603,
            message="Internal error",
            data=str(e)
        )

#4. Логирование

❌ ПЛОХО: Нет логирования
async def handle_request(request):
    response = await process(request)
    return response

✅ ХОРОШО: Структурированное логирование
import logging
import time

logger = logging.getLogger(__name__)

async def handle_request(request):
    start_time = time.time()
    request_id = generate_uuid()
    
    logger.info(
        "MCP request started",
        extra={
            "request_id": request_id,
            "method": request.method,
            "params": sanitize_params(request.params)
        }
    )
    
    try:
        response = await process(request)
        duration = time.time() - start_time
        
        logger.info(
            "MCP request completed",
            extra={
                "request_id": request_id,
                "method": request.method,
                "duration_ms": duration * 1000
            }
        )
        
        return response
    except Exception as e:
        logger.exception(
            "MCP request failed",
            extra={
                "request_id": request_id,
                "method": request.method
            }
        )
        raise

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

Концепция	Суть
JSON-RPC 2.0	Формат сообщений MCP
Сессия	Инициализация → Работа → Завершение
Capabilities	Сервер сообщает о возможностях при инициализации
Resources API	`resources/list`, `resources/read`, `resources/subscribe`
Tools API	`tools/list`, `tools/call`
Prompts API	`prompts/list`, `prompts/get`
Уведомления	Сообщения без `id`, не требуют ответа
Транспорты	stdio (локально), SSE/WebSocket (сеть)

Следующая тема: Установка и настройка — установка Python SDK, настройка окружения, подготовка инструментов разработки.

┌─────────────────────────────────────────────────────────────────┐ │ MCP Host │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ LLM Engine │ │ │ │ (Claude, GPT, ...) │ │ │ └──────────────────────────┬────────────────────────────────┘ │ │ │ │ │ ┌──────────────────────────▼────────────────────────────────┐ │ │ │ MCP Client │ │ │ │ ┌──────────────┬──────────────┬──────────────────────┐ │ │ │ │ │ Connection │ Message │ Session │ │ │ │ │ │ Manager │ Router │ Manager │ │ │ │ │ └──────────────┴──────────────┴──────────────────────┘ │ │ │ └──────────────────────────┬────────────────────────────────┘ │ └─────────────────────────────┼───────────────────────────────────┘ │ MCP Protocol │ (JSON-RPC 2.0 over Transport) ┌─────────────────────────────▼───────────────────────────────────┐ │ MCP Server │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ Protocol Layer │ │ │ │ (Request Handler, Response Builder) │ │ │ └──────────────────────────┬────────────────────────────────┘ │ │ │ │ │ ┌──────────────────────────▼────────────────────────────────┐ │ │ │ Capability Providers │ │ │ │ ┌──────────────┬──────────────┬──────────────────────┐ │ │ │ │ │ Resources │ Tools │ Prompts │ │ │ │ │ │ Provider │ Provider │ Provider │ │ │ │ │ └──────────────┴──────────────┴──────────────────────┘ │ │ │ └───────────────────────────────────────────────────────────┘ │ │ │ │ │ ┌──────────────────────────▼────────────────────────────────┐ │ │ │ Backend Integrations │ │ │ │ (File System, Database, External APIs, ...) │ │ │ └───────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘

Поле

Тип

Обязательное

Описание

jsonrpc

string

Да

Версия протокола, всегда "2.0"

id

integer/string

Да*

Уникальный ID запроса (отсутствует для notifications)

method

string

Да

Имя метода, например tools/list

params

object/array

Нет

Параметры метода

result

object

Да*

Результат выполнения (в ответе)

error

object

Да*

Ошибка выполнения (в ответе при ошибке)

┌─────────────┐ ┌─────────────┐ │ MCP Client │ │ MCP Server │ │ (Host) │ │ (Server) │ └──────┬──────┘ └──────┬──────┘ │ │ │ 1. initialize (capabilities) │ │────────────────────────────────────────>│ │ │ │ 2. initialize response │ │ (server capabilities) │ │<────────────────────────────────────────│ │ │ │ 3. notifications/initialized │ │────────────────────────────────────────>│ │ │ │ ◄─── Сессия активна, запросы ───► │ │ │ │ 4. tools/list, resources/read, ... │ │────────────────────────────────────────>│ │ │ │ 5. response with data │ │<────────────────────────────────────────│ │ │ │ 6. close connection │ │────────────────────────────────────────>│ │ │

Поле

Значение

subscribe

Сервер поддерживает подписку на изменения ресурсов

listChanged

Сервер отправляет уведомления при изменении списка ресурсов

Поле

Значение

listChanged

Сервер отправляет уведомления при изменении списка инструментов

Поле

Значение

listChanged

Сервер отправляет уведомления при изменении списка промптов

Метод

Описание

resources/list

Получить список доступных ресурсов

resources/read

Прочитать содержимое ресурса по URI

resources/subscribe

Подписаться на изменения ресурса

resources/unsubscribe

Отписаться от изменений ресурса

Метод

Описание

tools/list

Получить список доступных инструментов

tools/call

Вызвать инструмент с аргументами

Метод

Описание

prompts/list

Получить список доступных промптов

prompts/get

Получить промпт по имени с аргументами

Код

Значение

-32700

Parse error — невалидный JSON

-32600

Invalid Request — невалидная структура запроса

-32601

Method not found — метод не существует

-32602

Invalid params — невалидные параметры

-32603

Internal error — внутренняя ошибка сервера

┌─────────────┐ ┌─────────────┐ │ MCP Client │ │ MCP Server │ │ (Host) │ │ (Process) │ └──────┬──────┘ └──────┬──────┘ │ │ │ stdout (read) │ │<──────────────────────│ │ │ │ stdin (write) │ │──────────────────────>│ │ │

┌─────────────┐ ┌─────────────┐ │ MCP Client │ │ MCP Server │ │ (Host) │ │ (FastAPI) │ └──────┬──────┘ └──────┬──────┘ │ │ │ GET /sse (SSE connection) │ │──────────────────────────────────────>│ │ │ │ event: endpoint │ │ data: /messages?session_id=abc │ │<──────────────────────────────────────│ │ │ │ POST /messages?session_id=abc │ │ (JSON-RPC requests) │ │──────────────────────────────────────>│ │ │ │ SSE events (JSON-RPC responses) │ │<──────────────────────────────────────│ │ │

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

❌ ПЛОХО: Вся логика в одном классе class MCPServer: async def handle_request(self, request): # 500 строк кода с обработкой всего ✅ ХОРОШО: Разделение по ответственности class MCPServer: def __init__(self): self.resources_handler = ResourcesHandler() self.tools_handler = ToolsHandler() self.prompts_handler = PromptsHandler() async def handle_request(self, request): method = request.method if method.startswith("resources/"): return await self.resources_handler.handle(request) elif method.startswith("tools/"): return await self.tools_handler.handle(request) elif method.startswith("prompts/"): return await self.prompts_handler.handle(request)

Концепция

Суть

JSON-RPC 2.0

Формат сообщений MCP

Сессия

Инициализация → Работа → Завершение

Capabilities

Сервер сообщает о возможностях при инициализации

Resources API

resources/list, resources/read, resources/subscribe

Tools API

tools/list, tools/call

Prompts API

prompts/list, prompts/get

Уведомления

Сообщения без id, не требуют ответа

Транспорты

stdio (локально), SSE/WebSocket (сеть)