Первый сервер и тестовый клиент

Ситуация

Нужен самый маленький сервер, который уже полезен: читает последние строки журнала «Заметок». И нужен способ его вызвать без модели — чтобы проверять работу, не тратя денег и не гадая, что «решила» модель.

Главная мысль

MCP-сервер — обычная программа. Хост запускает её и разговаривает через stdin и stdout. Поэтому stdout принадлежит протоколу: всё, что вы напечатаете туда сами, сломает обмен. Сообщения для человека — в stderr.

Подготовка

Нужен Python 3.12. В Ubuntu 24.04 он есть; модуль виртуальных окружений ставится пакетом python3.12-venv.

mkdir -p ~/mcp-course && cd ~/mcp-course
python3 -m venv .venv
. .venv/bin/activate
pip install mcp==2.3.0
python -c "from importlib.metadata import version; print(version('mcp'))"

Последняя строка должна напечатать 2.3.0. Версия закреплена: в SDK 2.x поменялись имена (например, прежний FastMCP теперь MCPServer), и примеры курса проверены именно на 2.3.0.

Скачайте файлы ученика: make_lab.py, check.py и requirements.txt. Распакуйте эти три файла в ~/mcp-course, затем создайте учебную папку:

python3 make_lab.py

Скрипт создаёт ~/mcp-lab: журнал, заметки и файлы для проверок границ. Если папка уже есть, он откажет и ничего не изменит; --fresh переименует старую и создаст новую. В учебной папке лежит метка .mcp-lab — наш сервер будет отказываться работать без неё. Это защита от случайного запуска на настоящих данных.

Сервер

Создайте ~/mcp-course/diag_server.py:

"""Учебный MCP-сервер диагностики приложения «Заметки»."""
import os
import sys
from pathlib import Path

from mcp.server.mcpserver import MCPServer

LAB = Path(os.environ.get("MCP_LAB_DIR", "~/mcp-lab")).expanduser()
if not (LAB / ".mcp-lab").is_file():
    # stdout занят протоколом, поэтому сообщения — только в stderr.
    print(f"Отказ: {LAB} не учебная папка (нет метки .mcp-lab). Создайте её: python3 make_lab.py", file=sys.stderr)
    sys.exit(1)

LOG = LAB / "app" / "logs" / "app.log"
NOTES = LAB / "app" / "data" / "notes"

# Записка дежурного как инструкции для агента — удобно: агент сразу знает, где журнал.
# (К этому решению курс вернётся в главе 6.)
mcp = MCPServer("notes-diag", version="0.2.0", instructions=(NOTES / "README-ops.md").read_text(encoding="utf-8"))


@mcp.tool()
def log_tail(lines: int = 20) -> str:
    """Последние строки журнала приложения «Заметки»."""
    text = LOG.read_text(encoding="utf-8", errors="replace")
    return "\n".join(text.splitlines()[-lines:])


if __name__ == "__main__":
    mcp.run()

Что здесь происходит:

• MCPServer — класс официального SDK, который берёт на себя протокол: объявление инструментов, проверку запросов, ответы. В документации SDK его импортируют короче — from mcp.server import MCPServer; это тот же класс, работают оба способа.
• @mcp.tool() превращает функцию в инструмент. Имя — имя функции, описание — строка документации, аргументы — параметры функции с их типами.
• mcp.run() по умолчанию работает через stdio.
• Переменная MCP_LAB_DIR позволяет указать другую учебную папку; check.py пользуется ею, чтобы проверять сервер на временных копиях.

Запустите сервер руками: python diag_server.py. Он молчит — ждёт сообщений в stdin. Остановите его Ctrl+C.

Клиент без модели

Создайте ~/mcp-course/first_client.py:

"""Первый клиент — тестовый, без модели: вызывает инструменты по списку."""
import asyncio
import sys

from mcp import Client, StdioServerParameters


async def main() -> None:
    server = StdioServerParameters(command=sys.executable, args=["diag_server.py"])
    async with Client(server) as client:
        print("Протокол:", client.protocol_version)
        print("Сервер:", client.server_info.name, client.server_info.version)
        tools = await client.list_tools()
        print("Инструменты:", [t.name for t in tools.tools])
        result = await client.call_tool("log_tail", {"lines": 3})
        print(result.content[0].text)


asyncio.run(main())

python first_client.py

Ожидаемый вывод (время в строках журнала совпадёт — журнал создаётся одинаковым):

Протокол: 2026-07-28
Сервер: notes-diag 0.2.0
Инструменты: ['log_tail']
2026-10-16T09:59:48 INFO notes.http request method=GET path=/static/app.css status=200 ua="curl/8.5.0"
2026-10-16T09:59:54 INFO notes.http request method=GET path=/notes/42 status=200 ua="curl/8.5.0"
2026-10-16T10:00:00 WARN notes.http bad bytes in query q=��

Client сам запустил сервер как процесс, спросил его server/discover, выбрал версию 2026-07-28 и дальше отправлял её в каждом запросе. Это тестовый клиент: он не решает, какой инструмент вызвать, — вызывает тот, что вы написали. Модель в такой схеме была бы внутри хоста и только советовала бы, какой вызов сделать.

Обратите внимание: клиент передаёт серверу не всё своё окружение, а только HOME, LOGNAME, PATH, SHELL, TERM, USER. Ваши ключи из переменных окружения серверу по умолчанию не достаются; если серверу что-то нужно, это передают явно (StdioServerParameters(env=...)).

Сырые сообщения

Посмотрим, что ходит по stdin и stdout. Каждое сообщение — одна строка JSON:

M='"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}'
( printf '%s\n' \
  "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"log_tail\",\"arguments\":{\"lines\":1},$M}}" \
  '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' \
  "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"tools/list\",\"params\":{\"_meta\":{\"io.modelcontextprotocol/protocolVersion\":\"2025-06-18\",\"io.modelcontextprotocol/clientCapabilities\":{}}}}"
  sleep 2 ) | python diag_server.py

Ответы (переносы строк добавлены для чтения):

{"jsonrpc":"2.0","id":2,"error":{"code":-32602,"message":"params._meta must be an object carrying the required
 'io.modelcontextprotocol/protocolVersion' and 'io.modelcontextprotocol/clientCapabilities' envelope keys"}}
{"jsonrpc":"2.0","id":3,"error":{"code":-32022,"message":"Unsupported protocol version",
 "data":{"supported":["2026-07-28"],"requested":"2025-06-18"}}}
{"jsonrpc":"2.0","id":1,"result":{"content":[{"text":"2026-10-16T10:00:00 WARN notes.http bad bytes in query q=��",
 "type":"text"}],"isError":false,"resultType":"complete","structuredContent":{"result":"…"},
 "_meta":{"io.modelcontextprotocol/serverInfo":{"name":"notes-diag","version":"0.2.0"}}}}

Три урока из трёх строк:

1. Запрос без версии в _meta отклонён: сервер не помнит соединение, каждый запрос должен представиться сам.
2. Версия старой ревизии в _meta отклонена ошибкой «Unsupported protocol version» со списком поддерживаемых. Старые ревизии этот SDK понимает только через прежнее рукопожатие initialize — в курсе его нет.
3. Ответы пришли не по порядку: сервер обрабатывает запросы параллельно и связывает ответ с запросом по id.

sleep 2 держит stdin открытым: когда stdin закрывается, сервер завершается, и незаконченные запросы пропадают. Хост поступает так же: он держит процесс сервера, пока тот нужен.

Самостоятельно

1. Добавьте инструмент log_levels без аргументов: сколько строк каждого уровня в журнале. Верните строку, где на каждой строке УРОВЕНЬ ЧИСЛО, например ERROR 13. Уровень — второе слово строки журнала.
2. Допишите в first_client.py вызов log_levels и печать ответа.
3. Напечатайте что-нибудь в stdout из инструмента (print("debug")) и запустите клиент. Клиент SDK 2.3.0 получит строку debug вместо сообщения JSON, запишет в stderr ошибку разбора (Invalid JSON … input_value='debug') и продолжит. Другой хост вправе оборвать соединение, а строка, похожая на JSON, может быть принята за ответ. Верните как было и напечатайте то же в stderr (print(..., file=sys.stderr)) — ошибки разбора больше нет.

Проверка

python check.py --chapter 2 diag_server.py

check.py — тестовый клиент курса без модели. Он создаёт временную учебную папку, запускает ваш сервер и вызывает инструменты по списку. Ожидается:

Итог: выполнено 5 из 5 проверок, пройдено 5.

Код возврата 0 — все проверки пройдены. Сообщения сервера (stderr) он сохраняет в файл, путь к которому печатает последней строкой.

Если результат другой

• ModuleNotFoundError: No module named 'mcp.server.mcpserver' — установлена не та версия: pip show mcp.
• No module named 'mcp.server.fastmcp' в чужом примере — пример для SDK 1.x.
• В выводе клиента ошибка Invalid JSON или хост обрывает соединение — сервер печатает в stdout (шаг 3 упражнения).
• Отказ: … не учебная папка — запустите python3 make_lab.py.
• В ответе log_levels числа другие — проверьте, что считаете второе слово строки, и что учебная папка свежая.

Решение — answers/02-first-server.md; полный файл после главы — project/stages/02/.

Источники

• Архитектура MCP: server/discover, _meta, tools/list, tools/call.
• Пакет mcp 2.3.0: MCPServer (mcp/server/mcpserver/server.py), Client и StdioServerParameters (mcp/client/client.py, mcp/client/stdio.py), список версий протокола (mcp_types/version.py).

@mcp.tool()
def log_levels() -> str:
    """Сколько строк каждого уровня в журнале: по строке «УРОВЕНЬ ЧИСЛО»."""
    counts: dict[str, int] = {}
    for line in LOG.read_text(encoding="utf-8", errors="replace").splitlines():
        parts = line.split(" ", 2)
        level = parts[1] if len(parts) > 1 else "?"
        counts[level] = counts.get(level, 0) + 1
    return "\n".join(f"{level} {n}" for level, n in sorted(counts.items()))

        result = await client.call_tool("log_levels")
        print(result.content[0].text)

DEBUG 1
ERROR 13
INFO 2361
WARN 26

"""Учебный MCP-сервер диагностики приложения «Заметки» — состояние после главы 2."""
import os
import sys
from pathlib import Path

from mcp.server.mcpserver import MCPServer

LAB = Path(os.environ.get("MCP_LAB_DIR", "~/mcp-lab")).expanduser()
if not (LAB / ".mcp-lab").is_file():
    # stdout занят протоколом, поэтому сообщения — только в stderr.
    print(f"Отказ: {LAB} не учебная папка (нет метки .mcp-lab). Создайте её: python3 make_lab.py", file=sys.stderr)
    sys.exit(1)

LOG = LAB / "app" / "logs" / "app.log"
NOTES = LAB / "app" / "data" / "notes"

# Записка дежурного как инструкции для агента — удобно: агент сразу знает, где журнал.
# (К этому решению курс вернётся в главе 6.)
mcp = MCPServer("notes-diag", version="0.2.0", instructions=(NOTES / "README-ops.md").read_text(encoding="utf-8"))


@mcp.tool()
def log_tail(lines: int = 20) -> str:
    """Последние строки журнала приложения «Заметки»."""
    text = LOG.read_text(encoding="utf-8", errors="replace")
    return "\n".join(text.splitlines()[-lines:])


# Самостоятельный шаг главы 2.
@mcp.tool()
def log_levels() -> str:
    """Сколько строк каждого уровня в журнале: по строке «УРОВЕНЬ ЧИСЛО»."""
    counts: dict[str, int] = {}
    for line in LOG.read_text(encoding="utf-8", errors="replace").splitlines():
        parts = line.split(" ", 2)
        level = parts[1] if len(parts) > 1 else "?"
        counts[level] = counts.get(level, 0) + 1
    return "\n".join(f"{level} {n}" for level, n in sorted(counts.items()))


if __name__ == "__main__":
    mcp.run()

"""Первый клиент главы 2 — тестовый, без модели: вызывает инструменты по списку."""
import asyncio
import sys

from mcp import Client, StdioServerParameters


async def main() -> None:
    server = StdioServerParameters(command=sys.executable, args=["diag_server.py"])
    async with Client(server) as client:
        print("Протокол:", client.protocol_version)
        print("Сервер:", client.server_info.name, client.server_info.version)
        tools = await client.list_tools()
        print("Инструменты:", [t.name for t in tools.tools])
        result = await client.call_tool("log_tail", {"lines": 3})
        print(result.content[0].text)
        # Самостоятельный шаг: вызвать log_levels и напечатать ответ.
        result = await client.call_tool("log_levels")
        print(result.content[0].text)


asyncio.run(main())