Почему чистый stdout защищает границу протокола MCP?

Граница MCP через stdio остается надежной только тогда, когда каждый байт в stdout принадлежит протоколу, а любой malformed или неоднозначный запрос прекращается до того, как достигает исполнителя действий. Парсер, который отклоняет мусор, но пропускает частично распознанный запрос к HTTP или SSH, дает сбой именно в опасной точке.

Команды часто воспринимают шум в stdout как досадную ошибку совместимости. Это слишком мягкая оценка. Если агент может запускать внешние действия, лишний баннер способен рассинхронизировать диалог, скрыть ответ об ошибке или убедить снисходительный клиент связать не тот ответ не с тем запросом. Правильный тест проверяет больше, чем чистое завершение процесса. Он доказывает, что некорректный ввод на уровне провода не вызывает внешнего эффекта.

Контракт stdout принимает по одному сообщению JSON-RPC в строке

Транспорт Model Context Protocol через stdio требует передавать сообщения JSON-RPC в stdout, разделяя их переводами строк и не смешивая с этим потоком посторонний вывод. В обратную сторону действует то же правило: клиент отправляет протокольные сообщения в stdin и не использует этот канал как трубу для журналов.

Это кажется очевидным, пока установщик пакета не напечатает уведомление, зависимость не выдаст предупреждение или кто-то не оставит временный print() в коде запуска. В обычном терминале такие строки безвредны. В протокольном потоке это байты, которые другая сторона должна интерпретировать. У нее нет надежного способа понять, является ли loading credentials баннером, malformed-результатом, фрагментом ответа или началом скомпрометированного обмена.

В транспортном контракте есть три части, которые тесты должны формулировать явно:

• Каждая полная строка должна декодироваться как UTF-8 и разбираться в одно значение JSON.
• Это значение должно иметь допустимую форму запроса, ответа или уведомления JSON-RPC для соответствующего направления передачи.
• Никакое действие не должно начинаться, пока полный запрос не пройдет проверки фрейминга, JSON, протокола, схемы и авторизации.

Первое условие обнаруживает шум. Второе выявляет объект, который оказывается корректным JSON, но не является сообщением MCP. Третье защищает от главной ошибки: нельзя принимать ранний этап парсинга за разрешение на выполнение.

Спецификация JSON-RPC 2.0 отделяет ошибку парсинга от некорректного запроса. Некорректный JSON может получить код ошибки -32700, если другая сторона еще способна отправить ответ в правильном фрейме. Значение JSON с неправильной структурой запроса считается некорректным запросом, обычно с кодом -32600. Эти коды помогают совместимой стороне диагностировать сбой. Но они не показывают, остался ли ваш исполнитель нетронутым. На этот вопрос тесты должны отвечать напрямую.

Баннер может повредить авторизованный ответ

Баннер при запуске способен сломать совершенно допустимое действие после успешной авторизации. Поэтому чистота stdout важна не только для входной проверки.

Представьте сервер, который принял запрос initialize и собирается вернуть результат инструмента. Зависимость выводит warning: configuration missing в stdout между началом и завершением жизненного цикла ответа. Строгий клиент отклоняет строку и разрывает соединение. Снисходительный клиент пропускает ее и продолжает работу. Строгий клиент теряет доступность. У снисходительного появляется политика парсинга, допускающая ничейные байты внутри чувствительного к безопасности обмена.

Не стоит считать снисходительный клиент удобным решением. Если клиент отбрасывает произвольные строки, ему приходится отвечать на длинный список вопросов, на которые он не может надежно ответить. Он отбросил диагностическое сообщение? Ответ на другой запрос? Обертка продублировала строку? Злоумышленник, способный влиять на дочерний процесс, внедрил текст, который изменил состояние клиента? По случайной последовательности байтов нельзя вывести намерение.

Храните диагностику в stderr. Настройте для stderr отдельные правила захвата, хранения и редактирования чувствительных данных, а управление процессом должно сохранять разделение потоков. Удивительно распространенная ошибка, проявляющаяся только в релизе, возникает, когда обертка объединяет оба потока, потому что во время разработки вывод в терминале выглядел удобнее. Такая обертка незаметно разрушает границу протокола.

Проверяйте исходящий трафик как необработанные байты, до того как библиотека клиента нормализует их. Если библиотека превращает некорректную последовательность в исключение и скрывает исходный транскрипт, сохраняйте его в выводе упавшего теста. Точная первая плохая строка экономит часы догадок.

Отказ должен доходить до исполнителя действий

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

Поместите рекордер действий сразу за последней границей проверки и авторизации. В реальной реализации этот стык может оборачивать функцию, которая открывает HTTP-соединение или вызывает SSH-помощник. В тесте используйте рекордер в памяти или локальный фальшивый сервис. Не направляйте тесты с некорректным вводом на настоящий endpoint и не рассчитывайте, что путь обработки ошибки вас спасет.

Эту разницу легко размыть, потому что обычный запрос проходит длинный путь. Он приходит как байты, превращается в JSON, затем в объект JSON-RPC, затем в вызов метода MCP, сопоставляется со схемой инструмента, получает решение по авторизации и наконец становится действием. Разработчик может создать запись аудита или объект запроса до завершения всех этих проверок. Это допустимо, только если ни одна из таких операций не может обратиться во внешний мир или использовать capability.

Полезный инвариант выглядит так: исполнитель принимает полностью типизированный и авторизованный объект действия, но не сырой JSON и не частично проверенный запрос. Если исполнитель принимает общий словарь, рано или поздно кто-нибудь вызовет его слишком рано. Месяцами все может проходить в happy path, потому что в обычной разработке malformed-фреймы встречаются редко.

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

Поместите тестовый стык ниже парсинга и выше выполнения

Самый маленький полезный тестовый стенд состоит из строгого валидатора провода и фальшивого исполнителя. Валидатор отвечает за байты и форму протокола. Фальшивый исполнитель записывает каждую попытку выполнить работу. Производственный адаптер может отличаться, но контракт между ними должен оставаться узким.

Этот пример на Python намеренно небольшой. Сохраните его как test_stdio_boundary.py, установите pytest и запустите pytest -q test_stdio_boundary.py. Замените Gateway адаптером своего шлюза, сохранив проверки вокруг Recorder.

import io
import json
import pytest

class Recorder:
    def __init__(self):
        self.calls = []

    def execute(self, action):
        self.calls.append(action)
        return {'ok': True}

class Gateway:
    def __init__(self, executor):
        self.executor = executor
        self.closed = False

    def reject(self, code, reason):
        self.closed = True
        return {'jsonrpc': '2.0', 'id': None,
                'error': {'code': code, 'message': reason}}

    def receive_line(self, raw_line):
        if self.closed:
            return None
        try:
            message = json.loads(raw_line)
        except json.JSONDecodeError:
            return self.reject(-32700, 'parse error')

        if not isinstance(message, dict):
            return self.reject(-32600, 'invalid request')
        if message.get('jsonrpc') != '2.0':
            return self.reject(-32600, 'invalid request')
        if message.get('method') != 'tools/call':
            return self.reject(-32601, 'method not found')
        if not isinstance(message.get('id'), (str, int)) or isinstance(message.get('id'), bool):
            return self.reject(-32600, 'invalid request')

        params = message.get('params')
        if not isinstance(params, dict):
            return self.reject(-32602, 'invalid params')
        if not isinstance(params.get('name'), str):
            return self.reject(-32602, 'invalid params')
        if not isinstance(params.get('arguments', {}), dict):
            return self.reject(-32602, 'invalid params')

        action = {'name': params['name'], 'arguments': params.get('arguments', {})}
        result = self.executor.execute(action)
        return {'jsonrpc': '2.0', 'id': message['id'], 'result': result}

def frame(value):
    return json.dumps(value, separators=(',', ':')) + '\n'


def test_bad_lines_never_execute():
    bad_lines = [
        'debug: entering tool handler\n',
        '<html>gateway unavailable</html>\n',
        '{not json}\n',
        'null\n',
        frame({'jsonrpc': '1.0', 'id': 4, 'method': 'tools/call', 'params': {}}),
        frame({'jsonrpc': '2.0', 'id': 4, 'method': 'tools/call', 'params': 'run'}),
    ]

    for raw_line in bad_lines:
        recorder = Recorder()
        gateway = Gateway(recorder)
        response = gateway.receive_line(raw_line)
        assert response['jsonrpc'] == '2.0'
        assert 'error' in response
        assert gateway.closed
        assert recorder.calls == []

def test_complete_valid_request_executes_once():
    recorder = Recorder()
    gateway = Gateway(recorder)
    request = frame({
        'jsonrpc': '2.0',
        'id': 9,
        'method': 'tools/call',
        'params': {'name': 'safe-test', 'arguments': {'value': 'green'}},
    })

    response = gateway.receive_line(request)
    assert response['result'] == {'ok': True}
    assert recorder.calls == [{'name': 'safe-test', 'arguments': {'value': 'green'}}]

Это не полноценная реализация MCP, и такой пример не должен ею становиться. Его задача состоит в том, чтобы сделать свойство «никаких действий» исполняемым. Адаптер может подавать реальные байты stdin в такой же тестовый стык и использовать настоящий валидатор запросов. Не переносите сокращенную обработку методов в production-сервер.

Обратите внимание на решение закрывать соединение после malformed или некорректной входной строки. Протокол не обязывает каждую реализацию использовать именно такую политику сессии. Это разумный вариант по умолчанию, когда шлюз управляет внешними побочными эффектами: поврежденная строка может означать, что другая сторона потеряла фрейминг. Если после корректно оформленного, но недопустимого запроса вы оставляете сессию открытой, отдельно протестируйте этот путь и докажите, что следующий допустимый запрос не унаследует состояние отклоненного.

Исполняемый тестовый стенд должен наблюдать байты и эффекты

Предыдущий модульный тест проверяет отказ на входе. Добавьте тест уровня процесса, чтобы обнаружить вывод, который появляется за пределами обычно проверяемого кода. Следующий помощник проверяет захваченный транскрипт stdout и не доверяет библиотеке клиента, которая разбирает его за вас.

import json
import subprocess

def assert_protocol_stdout(data):
    assert data.endswith(b'\n'), 'stdout ended without a complete frame'
    for number, raw_line in enumerate(data.splitlines(), start=1):
        try:
            line = raw_line.decode('utf-8')
            message = json.loads(line)
        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            raise AssertionError(
                f'non-protocol stdout on line {number}: {raw_line!r}'
            ) from error

        assert isinstance(message, dict), f'line {number} is not an object'
        assert message.get('jsonrpc') == '2.0', f'line {number} lacks JSON-RPC 2.0'
        is_notification = isinstance(message.get('method'), str) and 'id' not in message
        is_response = 'id' in message and ('result' in message or 'error' in message)
        assert is_notification or is_response, f'line {number} has no permitted shape'

def run_server(command, stdin_bytes):
    completed = subprocess.run(
        command,
        input=stdin_bytes,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    assert_protocol_stdout(completed.stdout)
    return completed

def test_release_command_has_clean_stdout():
    initialize = (
        b'{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",'
        b'\"params\":{\"protocolVersion\":\"2025-03-26\",'
        b'\"capabilities\":{},\"clientInfo\":{\"name\":\"boundary-test\",\"version\":\"1\"}}}\n'
    )
    completed = run_server(['./gateway-under-test'], initialize)
    assert completed.returncode == 0

Подставьте вместо ./gateway-under-test настоящую команду запуска. Если ей нужны среда выполнения, обертка пакета или переменные окружения, используйте те же production-условия. Тест должен проверять stdout даже при ненулевом коде завершения процесса. Упавший процесс все равно может вывести недопустимый баннер перед сбоем.

В сообщении об ошибке должна быть видна необработанная строка байтов. Полезный вывод выглядит так:

AssertionError: non-protocol stdout on line 1: b'loading optional extension\n'

Такой результат сразу показывает сопровождающему, где искать проблему. Общее сообщение вроде «сервер не инициализировался» отправляет людей искать ошибку в протокольном коде, хотя дефект может находиться в shell-скрипте, настройках журналирования или импорте зависимости.

Запускайте тестовый стенд в упакованном виде, а не только из checkout исходников. Упаковка меняет разрешение модулей, переменные окружения, права на выполнение и пути обработки ошибок. Именно там часто появляются случайные записи в stdout.

Тестируйте неудобные входные данные, а не вежливые ошибки

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

Используйте как минимум такие случаи:

• Обычная отладочная строка перед допустимым запросом, после которой на следующей строке идет этот запрос.
• Допустимый запрос, за которым на той же строке без разделителя перевода строки следует баннер.
• Усеченный JSON-объект, после которого наступает конец файла.
• Два полных JSON-объекта, соединенных без разделителя.
• Допустимый JSON-объект с правдоподобным методом, но malformed-параметрами.

Первый случай обнаруживает реализацию, которая молча пропускает плохие строки и продолжает работу. Второй выявляет ошибки фрейминга, когда построчный читатель может принять все за один поврежденный запрос. Усеченный случай проверяет код, который пытается восстановить ввод через буфер после ухода другой стороны. Соединенные объекты обнаруживают парсеры, настроенные принимать несколько JSON-значений верхнего уровня там, где транспорт разрешает один фрейм на строку.

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

Проверяйте также данные, корректные сами по себе, но опасные в контексте. Запрос с method, равным tools/call, и строковым params является допустимым JSON, но не вызовом. Запрос с числовым id, который ваш язык воспринимает как boolean, пересекает границу типов, которую вы не планировали пересекать. Запрос с неизвестными аргументами может быть отклонен схемой, но снисходительное объединение объектов способно случайно передать их в конструктор downstream-команды.

Шум при запуске обычно возникает по обычным причинам

Большая часть загрязнения stdout появляется из-за обычной работы по сопровождению, а не из-за попытки обойти протокол. Это не отменяет необходимости его обнаруживать.

Обертка команды может вывести уведомление о версии. Среда выполнения языка может записать предупреждение об устаревшей функции после изменения окружения. Разработчик может оставить отладочную строку в импорте пакета, который тесты не загружают. Обработчик сбоя может напечатать дружелюбное сообщение в stdout, потому что его создавали для терминального приложения. Супервизор может объединить stderr с stdout ради сбора журналов.

Рассматривайте каждый источник как отдельный релизный тест. Установите переменные окружения, включающие подробный вывод зависимостей. Запустите исполняемый файл из каталога без необязательной конфигурации. Вызовите восстанавливаемый сбой при запуске. Проверьте первый запрос после простоя. Затем каждый раз проверяйте необработанный транскрипт.

Не отключайте все журналы только ради прохождения теста. Так операционная проблема просто переместится в другое место. Направляйте диагностику в stderr, дайте операторам явный способ ее собирать и редактируйте чувствительные значения до того, как они покинут процесс. Корректный протокол и полезная диагностика совместимы, если потоки остаются разделенными.

Корректный JSON еще не доказывает безопасность запроса

Строгий парсер JSON защищает фрейминг. Он не решает, может ли запрос использовать учетные данные или обращаться к хосту.

Соблюдайте следующий порядок. Сначала транспорт читает один фрейм. Затем парсер создает одно значение JSON. После этого валидатор протокола проверяет форму JSON-RPC и метода MCP. Валидатор схемы проверяет аргументы инструмента. Только после этих этапов авторизация решает, можно ли выполнить запрошенное действие. Исполнитель должен получать типизированное действие вместе с этим решением, а не сырой метод и словарь параметров.

Такой порядок предотвращает незаметный сбой: создание HTTP-запроса во время проверки. Если последующая проверка отклонит вызов, но библиотека уже разрешила адрес хоста, открыла соединение или развернула шаблон команды, тест может сообщить об отказе, хотя граница уже выпустила действие наружу. Тест с фальшивым исполнителем обнаруживает очевидную форму проблемы. Локальный фальшивый HTTP-сервис или тестовый SSH-помощник могут выявить случайную сетевую активность в интеграционных тестах.

Отклоненный запрос может попасть в аудит, но не превращайте запись аудита в боковой канал, который вызывает слой действий. Записывайте отказ как отказ с причиной и отпечатком запроса, не раскрывающим секреты. Журнал внешних действий должен отличаться от записи о попытке вызова, которая не прошла авторизацию.

Одна случайная команда print может скрыть опасный сбой

Представьте шлюз, поддерживающий инструмент с именем deploy-preview. Его обработчик проверяет часть аргументов, начинает собирать исходящий запрос, а затем проверяет, может ли вызывающая сторона использовать выбранные учетные данные. Во время рефакторинга разработчик добавляет print в stdout, чтобы посмотреть на выбранную цель.

Строгий MCP-клиент видит этот вывод, не может разобрать его как JSON-RPC и отключается. Разработчик видит ошибку протокола и исправляет print. Это неудобно, но безопасно.

Снисходительный клиент пропускает строку, получает ответ об ошибке и сообщает оператору, что авторизация отклонила запрос. Тем временем обработчик уже передал собранный запрос повторяющему HTTP-помощнику до проверки авторизации. Удаленный сервис получает запрос без подходящих учетных данных, возможно возвращает ошибку и оставляет команде вводящий в заблуждение аудит: кажется, что агенту отказали, хотя сервис все же увидел активность.

Баннер не создал ошибку в порядке авторизации. Он показал, почему снисходительное восстановление транспорта и раннее создание действия плохо сочетаются. Решение не в более умном правиле пропуска. Перенесите решение по авторизации перед созданием запроса, используйте рекордер, чтобы это доказать, и сделайте stdout настолько строгим, чтобы случайный вывод сразу проваливал тест.

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

Разместите тест чистого stdout рядом с каждой упакованной командой шлюза и запускайте его в continuous integration. В сообщение об ошибке включайте первые нарушившие байты, команду запуска и stderr как отдельное вложение. Сопровождающий должен иметь возможность воспроизвести ошибку без восстановления диалога с агентом.

Храните небольшой набор malformed-входных фреймов под контролем версий. Добавляйте новый случай после каждого реального дефекта. Не принимайте новую странность только потому, что ее выдал один клиент. Если реализация отправляет malformed-трафик, исправьте реализацию или задокументируйте версионированную границу совместимости, которая не разрешает действия.

Sallyport помещает HTTP- и SSH-выполнение за встроенный stdio-shim sp mcp, поэтому этот тестовый стенд должен находиться на границе этого shim и проверять, что отклоненный фрейм не может вызвать внешнюю операцию. Такой же подход нужен любому MCP-серверу, который умеет больше, чем возвращать текст.

Критерий выпуска прост: stdout содержит только полные протокольные сообщения, а отклоненный ввод не оставляет следа в рекордере действий. Если не выполняется хотя бы одно из этих условий, сборка не готова пропускать трафик агентов.