Создание статeless, безопасного MCP протокола

Основные Принципы Дизайна MCP

В этом руководстве мы создаем чистую, продвинутую демонстрацию современного дизайна MCP, сосредотачиваясь на трех основных идеях: stateless коммуникации, строгой валидации на уровне SDK и асинхронных, долгосрочных операциях. Мы реализуем минимальный протокол, похожий на MCP, используя структурированные конверты, подписанные запросы и инструменты с валидацией Pydantic, чтобы показать, как агенты и сервисы могут безопасно взаимодействовать без зависимости от постоянных сессий.

Настройка Основных Утилит

Мы настраиваем основные утилиты, необходимые для всей системы, включая помощники по времени, генерацию UUID, сериализацию в каноническом JSON и криптографическую подпись. Мы гарантируем, что все запросы и ответы могут быть детерминированно подписаны и проверены с использованием HMAC.

import asyncio, time, json, uuid, hmac, hashlib
from dataclasses import dataclass
from typing import Any, Dict, Optional, Literal, List
from pydantic import BaseModel, Field, ValidationError, ConfigDict
 
def _now_ms():
   return int(time.time() * 1000)
 
def _uuid():
   return str(uuid.uuid4())
 
def _canonical_json(obj):
   return json.dumps(obj, separators=(',', ':'), sort_keys=True).encode()
 
def _hmac_hex(secret, payload):
   return hmac.new(secret, _canonical_json(payload), hashlib.sha256).hexdigest()

Определение MCP Конверта и Ответа

Мы определяем структурированные форматы MCP конвертов и ответов, которые соблюдаются при каждом взаимодействии. Мы применяем строгие схемы с использованием Pydantic, чтобы гарантировать, что ошибочные или неожиданные поля отклоняются заранее.

class MCPEnvelope(BaseModel):
   model_config = ConfigDict(extra='forbid')
   v: Literal['mcp/0.1'] = 'mcp/0.1'
   request_id: str = Field(default_factory=_uuid)
   ts_ms: int = Field(default_factory=_now_ms)
   client_id: str
   server_id: str
   tool: str
   args: Dict[str, Any] = Field(default_factory=dict)
   nonce: str = Field(default_factory=_uuid)
   signature: str
 
class MCPResponse(BaseModel):
   model_config = ConfigDict(extra='forbid')
   v: Literal['mcp/0.1'] = 'mcp/0.1'
   request_id: str
   ts_ms: int = Field(default_factory=_now_ms)
   ok: bool
   server_id: str
   status: Literal['ok', 'accepted', 'running', 'done', 'error']
   result: Optional[Dict[str, Any]] = None
   error: Optional[str] = None
   signature: str

Валидация Моделей Входа и Выхода

Мы объявляем проверенные модели входа и выхода для каждого инструмента, доступного на сервере. Мы используем ограничения Pydantic, чтобы четко выразить, что каждый инструмент принимает и возвращает.

class ServerIdentityOut(BaseModel):
   model_config = ConfigDict(extra='forbid')
   server_id: str
   fingerprint: str
   capabilities: Dict[str, Any]
 
class BatchSumIn(BaseModel):
   model_config = ConfigDict(extra='forbid')
   numbers: List[float] = Field(min_length=1)
 
class BatchSumOut(BaseModel):
   model_config = ConfigDict(extra='forbid')
   count: int
   total: float
 
class StartLongTaskIn(BaseModel):
   model_config = ConfigDict(extra='forbid')
   seconds: int = Field(ge=1, le=20)
   payload: Dict[str, Any] = Field(default_factory=dict)
 
class PollJobIn(BaseModel):
   model_config = ConfigDict(extra='forbid')
   job_id: str

Реализация Stateless MCP Сервер

Мы реализуем stateless MCP сервер вместе с логикой управления асинхронными задачами. Мы обрабатываем верификацию запросов, распределение инструментов и выполнение долгосрочных заданий без зависимости от состояния сессии.

class MCPServer:
   def __init__(self, server_id, secret):
       self.server_id = server_id
       self.secret = secret
       self.jobs = {}
       self.tasks = {}
 
   async def handle(self, env_dict, client_secret):
       env = MCPEnvelope(**env_dict)
       payload = env.model_dump()
       sig = payload.pop('signature')
       if _hmac_hex(client_secret, payload) != sig:
           return {'error': 'bad signature'}
 
       # Обработка различных инструментов

Создание Stateless Клиента

Мы создаем легковесный stateless клиент, который подписывает каждый запрос и взаимодействует с сервером через структурированные конверты.

class MCPClient:
   def __init__(self, client_id, secret, server):
       self.client_id = client_id
       self.secret = secret
       self.server = server
 
   async def call(self, tool, args=None):
       env = MCPEnvelope(
           client_id=self.client_id,
           server_id=self.server.server_id,
           tool=tool,
           args=args or {},
           signature='',
       ).model_dump()
       env['signature'] = _hmac_hex(self.secret, {k: v for k, v in env.items() if k != 'signature'})
       return await self.server.handle(env, self.secret)

Заключение

Мы показали, как MCP развивается от простого интерфейса вызова инструментов до надежного протокола, подходящего для реальных систем. Вместе эти шаблоны демонстрируют, как современные системы в стиле MCP поддерживают надежные, готовые к предприятию рабочие процессы агентов, оставаясь при этом простыми, прозрачными и легкими для расширения.