Клиент, который проверяет ответ сервера

Теперь сервер имеет понятные инструменты и границы записи. Добавим маленький клиент, который запускает reader-процесс, делает обнаружение и вызывает поиск. После этого урока у обмена будет явный жизненный цикл: запуск, запрос, проверка ответа и завершение. Клиент не вызывает AI, не выдаёт operator-токены и не является универсальным MCP SDK. Он реализует только выбранный синхронный профиль нашей лаборатории.

client.py использует стандартные subprocess, threading и queue. Чтение ответа вынесено в поток, чтобы основной вызов мог иметь срок ожидания. Пример предназначен для самостоятельного запуска читателем, но при подготовке не выполнялся. Мы не подключали сервер к внешним хостам и не наблюдали реальный интерфейс модели. Договор программы объясняется по авторским исходникам и ограниченному набору доступных сообщений.

Дочерний процесс без shell

Клиент запускает server.py списком аргументов через текущий Python-интерпретатор. Shell не используется, поэтому строки запроса не превращаются в команды терминала. Рабочая папка задана явно, а окружение содержит MCP_ROLE=reader и UTF-8-профиль. Это удобная локальная настройка, но не изоляция процесса. Дочерний сервер всё ещё наследует права пользователя и другие переменные, которые хост должен отдельно оценить перед реальным подключением.

Стандартный ввод и вывод связаны с pipe. stderr наследуется отдельно и не попадает в JSON-канал ответа. Клиент сериализует запрос в одну строку и отправляет перевод строки. Затем flush делает сообщение доступным серверу без ожидания дополнительного буфера. Эти детали важны для наблюдаемого обмена: корректный JSON, оставшийся в пользовательском буфере, ещё не означает, что дочерний процесс получил запрос.

Каждый вызов получает новый целочисленный id. В нашем клиенте одновременно выполняется только один запрос, поэтому простой счётчик достаточен. Универсальный асинхронный клиент должен хранить карту ожидаемых ответов и обрабатывать уведомления отдельно. Наш профиль не поддерживает такую диспетчеризацию и не обещает работу с произвольным сервером. Ограничение лучше назвать явно, чем скрыть за интерфейсом похожим на полноценный SDK.

Проверка ответа

Поток читает строку с ограничением размера и передаёт разобранный объект в очередь. Основной вызов ожидает его не больше пяти секунд. Затем проверяются JSON-RPC, id и форма result. Несовпадение идентификатора означает, что ответ нельзя считать результатом текущего запроса, даже если внутри есть красивый текст. Это защищает предметный вывод от случайного смешения сообщений и помогает объяснить диагностическую ошибку в конкретной точке обмена.

Клиент принимает только resultType: complete. Он не поддерживает input_required и не выполняет вложенные запросы подтверждения. Наш сервер использует отдельный операторский интерфейс вместо такого маршрута, поэтому ограничение согласовано с проектом. При подключении другого сервера отказ на неизвестную форму является правильным поведением для этого клиента. Молчаливое принятие любых результатов могло бы представить незавершённый запрос как окончательные данные.

Протокольная ошибка поднимается как локальное исключение. Предметный isError инструмента, напротив, остаётся частью result и требует собственного разбора вызывающим приложением. В основном сценарии клиент печатает полный результат поиска, не превращая его в ответ языковой модели. Поэтому reader видит фактическую структуру, которую сервер должен вернуть при самостоятельном выполнении. Печатание не является доказательством корректности всех output-полей, и наш клиент не реализует универсальную проверку JSON Schema.

Срок ожидания и остановка

Если ответ не пришёл вовремя, ожидание очереди завершается ошибкой, а блок finally закрывает клиента. Это не означает, что сервер обязательно отменил уже выполненное действие. Для нашего reader-сценария побочного эффекта нет; для записи потеря ответа потребовала бы operation и replay из предыдущего урока. Срок ожидания клиента является локальной границей, а не подтверждением отсутствия commit на стороне другого процесса.

Завершение начинается с закрытия stdin. Сервер видит конец потока и выходит из цикла. Клиент ждёт ограниченное время, затем при необходимости завершает процесс и в крайнем случае принудительно убивает его. Такой порядок даёт возможность спокойной остановки, не оставляя бесконечно висящий дочерний процесс. Он также показывает, что lifetime stdio-процесса имеет отдельный договор: окончание одного ответа не обязано немедленно завершать сервер.

Поток чтения имеет ограниченную очередь. Это удерживает небольшую учебную программу от накопления произвольного числа сообщений. Но если сервер начнёт слать уведомления или ответы на параллельные запросы, наш клиент не станет корректным автоматически. Он рассчитан на объявленный профиль коротких синхронных операций. Для расширения потребуются разбор уведомлений, корреляция по id и отдельное управление отменой. Не добавляйте такие обещания только потому, что использован threading.

Ожидаемый маршрут начинается с server/discover. Клиент убеждается, что выбранная версия присутствует в supportedVersions, затем вызывает library.search для «постоянный адрес» и печатает результат. Роль reader фиксирована кодом, поэтому служебные документы не должны попадать в выдачу. Ни один этап не выполнялся при подготовке; фактический вывод отсутствует. Самостоятельное наблюдение читателя нужно привязать к конкретной папке состояния и своему Python.

Ответ с чужим идентификатором

Представьте, что запрос имеет id=2, а ответ содержит id=1. Даже если structuredContent выглядит правильным, клиент не должен считать его текущим результатом. Возможно, это задержанный ответ другой операции либо ошибка сервера. Наш синхронный профиль отклоняет несоответствие и завершает процесс в finally. Универсальный клиент мог бы направить сообщение другому ожиданию, но для этого нужна отдельная карта активных запросов, которой здесь нет.

Другой вариант — результат input_required. Наш клиент не умеет отвечать на такой маршрут и должен сообщить о неподдерживаемой форме. Нельзя автоматически принять текст промежуточного результата за законченный поиск. Такая проверка удерживает область совместимости и делает отказ объяснимым. Сервер нашей лаборатории возвращает complete, поэтому ограничение соответствует авторскому проекту, но не любому доступному MCP-подключению.

Если stdout неожиданно закрыт, поток передаёт ошибку в очередь. Это отличается от пустого поиска, который является успешным result с items. Разбор формы помогает приложению показать причину на правильном уровне: инструмент не нашёл источник либо процесс больше не отвечает. Оба случая не должны превращаться в одну выдуманную статью-ответ.

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

Полный файл client.py из состояния lesson-11. Полное самостоятельное состояние находится в архиве серии; фрагмент не заменяет остальные определения.

from pathlib import Path
import json
import os
import queue
import subprocess
import sys
import threading

ROOT=Path(__file__).resolve().parent
VERSION="2026-07-28"
MAX_LINE=65536

class LocalClient:
    def __init__(self):
        env=dict(os.environ, MCP_ROLE="reader", PYTHONUTF8="1")
        self.process=subprocess.Popen([sys.executable,str(ROOT/"server.py")],cwd=ROOT,env=env,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=None)
        self.inbox=queue.Queue(maxsize=8)
        self.next_id=0
        threading.Thread(target=self._reader,daemon=True).start()
    def _reader(self):
        try:
            while True:
                line=self.process.stdout.readline(MAX_LINE+1)
                if not line: raise RuntimeError("server closed output")
                if len(line)>MAX_LINE: raise RuntimeError("server output exceeds limit")
                self.inbox.put(json.loads(line),timeout=1)
        except Exception as exc:
            try: self.inbox.put(exc,timeout=1)
            except queue.Full: pass
    def call(self,method,**params):
        self.next_id+=1
        params["_meta"]={"io.modelcontextprotocol/protocolVersion":VERSION,"io.modelcontextprotocol/clientCapabilities":{},"io.modelcontextprotocol/clientInfo":{"name":"PWReader","version":"1.0.0"}}
        request={"jsonrpc":"2.0","id":self.next_id,"method":method,"params":params}
        self.process.stdin.write((json.dumps(request,ensure_ascii=False)+"\n").encode())
        self.process.stdin.flush()
        reply=self.inbox.get(timeout=5)
        if isinstance(reply,Exception): raise reply
        if not isinstance(reply,dict) or reply.get("jsonrpc")!="2.0" or type(reply.get("id")) is not int or reply["id"]!=self.next_id:
            raise RuntimeError("invalid or mismatched response")
        if ("error" in reply) == ("result" in reply): raise RuntimeError("response must have exactly result or error")
        if "error" in reply: raise RuntimeError(str(reply["error"]))
        result=reply.get("result")
        if not isinstance(result,dict) or result.get("resultType")!="complete":
            raise RuntimeError("unsupported result shape")
        return result
    def close(self):
        if self.process.stdin and not self.process.stdin.closed: self.process.stdin.close()
        try: self.process.wait(timeout=2)
        except subprocess.TimeoutExpired:
            self.process.terminate()
            try: self.process.wait(timeout=2)
            except subprocess.TimeoutExpired: self.process.kill(); self.process.wait()
        if self.process.stdout: self.process.stdout.close()

if __name__ == "__main__":
    client=LocalClient()
    try:
        discovery=client.call("server/discover")
        if VERSION not in discovery.get("supportedVersions",[]): raise RuntimeError("incompatible protocol")
        print(json.dumps(client.call("tools/call",name="library.search",arguments={"query":"постоянный адрес","limit":3}),ensure_ascii=False,indent=2))
    finally:
        client.close()

Статус примера: not_run. При подготовке не выполнялись учебные программы, тесты, вызовы AI или деплой. Описанные эффекты ожидаемые.

Оглавление серии · Полные исходники

Первичные документы: Источник 1, Источник 2.