MCP Server своими руками: протокол Model Context Protocol

1. Что такое MCP и зачем он нужен

MCP (Model Context Protocol) — открытый протокол от компании Anthropic, который позволяет AI-моделям (таким как Claude) подключаться к внешним источникам данных, инструментам и сервисам. Представьте себе USB-C для AI-агентов: единый стандарт, через который любая модель может взаимодействовать с любым источником данных.

До появления MCP каждый разработчик, желающий дать AI-агенту доступ к своей базе данных, API или файловой системе, был вынужден писать кастомную интеграцию под конкретную модель. Это означало разные форматы, разные протоколы и полное отсутствие переиспользуемости. С MCP эта проблема решена: вы пишете один сервер, и он работает с Claude Desktop, Cursor, Continue.dev и любым другим MCP-совместимым клиентом.

Когда вы задаёте вопрос Claude через Claude Desktop, модель сама определяет, что ей нужны дополнительные данные. Допустим, вы спрашиваете: «Какая сейчас погода в Лондоне?» Модель понимает, что у неё нет актуальных данных, видит зарегистрированный MCP-инструмент get_weather, вызывает его с аргументом {"city": "London"}, получает JSON-ответ с температурой и влажностью от вашего сервера, и формирует финальный ответ пользователю на естественном языке. Весь этот цикл происходит за доли секунды и полностью прозрачен для пользователя — он просто получает актуальный ответ.

Три ключевых примитива MCP, которые сервер предоставляет клиенту:

• Resources — данные, которые модель может читать (файлы, записи БД, ответы API). Это как GET-запросы в REST.
• Tools — действия, которые модель может выполнять (отправить email, создать issue в GitHub, выполнить SQL-запрос). Это как POST/PUT в REST.
• Prompts — шаблоны запросов, которые помогают пользователю быстро сформулировать задачу для AI. Это как сохранённые промпты.

На практике это означает: вы можете написать MCP-сервер, который подключается к вашей внутренней CRM, и Claude Desktop сможет искать клиентов, создавать записи и генерировать отчёты — без какого-либо кода на стороне самого Claude.

В этом руководстве мы пройдём полный путь: от установки SDK до работающего сервера с инструментами и ресурсами, который можно подключить к Claude Desktop за 5 минут.

2. Архитектура протокола MCP

MCP построен на архитектуре клиент-сервер с тремя основными участниками:

• Host — приложение, в котором работает AI-модель. Например, Claude Desktop. Host управляет жизненным циклом клиента.
• Client — устанавливает соединение с сервером по JSON-RPC 2.0. Каждый сервер получает свой экземпляр клиента.
• Server — ваш код на Python (или TypeScript, Kotlin), который реализует Resources, Tools и Prompts.

Транспортный уровень: по умолчанию MCP использует stdio — сервер запускается как дочерний процесс, а JSON-RPC сообщения передаются через стандартный ввод/вывод. Это обеспечивает изоляцию и безопасность. Для удалённых серверов поддерживается HTTP SSE (Server-Sent Events).

Протокол инициализации выглядит так:

1. Host запускает сервер как подпроцесс
2. Клиент отправляет initialize — сообщает о своих возможностях
3. Сервер отвечает initialize — сообщает о своих возможностях
4. Клиент отправляет initialized — уведомление о готовности
5. Начинается рабочий цикл: клиент запрашивает tools/list, вызывает tools/call, читает resources/read

3. Установка и первый MCP-сервер

Нам понадобится Python 3.10+ и официальный MCP SDK. Устанавливаем всё необходимое одной командой:

# Установка MCP SDK и зависимостей
pip install mcp httpx

# Или с uv (быстрее, рекомендуется)
uv pip install mcp httpx

Создадим минимальный MCP-сервер, который просто отвечает на запрос initialize и сообщает о своей готовности:

#!/usr/bin/env python3
# server.py — минимальный MCP-сервер

import asyncio
import logging

from mcp.server import Server
from mcp.server.stdio import stdio_server

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("my-mcp-server")

# Создаём экземпляр сервера с уникальным именем
server = Server("my-first-server")


@server.list_tools()
async def handle_list_tools() -> list:
    # Пока возвращаем пустой список — инструментов нет
    return []


async def main():
    async with stdio_server() as (read_stream, write_stream):
        logger.info("🚀 MCP-сервер запущен. Ожидаю подключения...")
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options(),
        )


if __name__ == "__main__":
    asyncio.run(main())

Этот сервер уже можно запустить:

python server.py
# INFO:my-mcp-server:🚀 MCP-сервер запущен. Ожидаю подключения...

Сервер слушает stdio — он ждёт JSON-RPC сообщений на стандартном вводе. Пока он делает только одно: отвечает на tools/list пустым списком. Но это уже полноценный MCP-сервер, который можно подключить к Claude Desktop (мы сделаем это в разделе 6).

4. Добавляем Tools — инструменты для AI-агента

Tools — это сердце MCP-сервера. Именно через инструменты AI-модель может выполнять действия: искать информацию, вызывать API, работать с файлами. Каждый инструмент описывается схемой (имя, описание, параметры) и имеет функцию-обработчик.

Добавим в наш сервер два практических инструмента: получение курса криптовалют через публичный API и калькулятор хеша файла. Вот полный код сервера с инструментами:

#!/usr/bin/env python3
# mcp_tools_server.py — MCP-сервер с инструментами

import asyncio
import hashlib
import logging
import json

import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-tools")

server = Server("crypto-tools-server")

# ── Описание инструментов ──

TOOLS = [
    Tool(
        name="get_crypto_price",
        description="Получить текущую цену криптовалюты в USD. Поддерживает BTC, ETH, SOL, DOGE и другие.",
        inputSchema={
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "Тикер криптовалюты (например, bitcoin, ethereum, solana)",
                }
            },
            "required": ["symbol"],
        },
    ),
    Tool(
        name="hash_file",
        description="Вычислить SHA-256 хеш строки или виртуального файла. Полезно для проверки целостности данных.",
        inputSchema={
            "type": "object",
            "properties": {
                "content": {
                    "type": "string",
                    "description": "Содержимое для хеширования",
                }
            },
            "required": ["content"],
        },
    ),
]


@server.list_tools()
async def handle_list_tools() -> list:
    return TOOLS


@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    logger.info(f"Вызван инструмент: {name} с аргументами {arguments}")

    if name == "get_crypto_price":
        symbol = arguments["symbol"].lower()
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"https://api.coingecko.com/api/v3/simple/price?ids={symbol}&vs_currencies=usd"
            )
            data = resp.json()

        if symbol in data:
            price = data[symbol]["usd"]
            result = f"💰 Цена {symbol.upper()}: ${price:,.2f} USD"
        else:
            result = f"❌ Криптовалюта '{symbol}' не найдена"

        return [TextContent(type="text", text=result)]

    elif name == "hash_file":
        content = arguments["content"]
        sha = hashlib.sha256(content.encode()).hexdigest()
        result = f"🔐 SHA-256: {sha}"

        return [TextContent(type="text", text=result)]

    else:
        raise ValueError(f"Неизвестный инструмент: {name}")


async def main():
    async with stdio_server() as (read_stream, write_stream):
        logger.info("🛠️  MCP Tools Server запущен (2 инструмента)")
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options(),
        )


if __name__ == "__main__":
    asyncio.run(main())

Ключевые моменты в этом коде:

• Tool из mcp.types — это Pydantic-модель с полями name, description и inputSchema. Схема описывается в формате JSON Schema.
• Декоратор @server.list_tools() регистрирует функцию, возвращающую список доступных инструментов.
• Декоратор @server.call_tool() принимает имя инструмента и аргументы, а возвращает список объектов TextContent.
• TextContent — стандартный способ вернуть текстовый результат. MCP также поддерживает ImageContent и EmbeddedResource.

5. Добавляем Resources — данные для чтения

Resources позволяют AI-модели читать данные из вашего сервера. Это может быть что угодно: содержимое файлов, записи из базы данных, ответы API, системная информация. В отличие от Tools (которые выполняют действия), Resources — это пассивные источники данных по URI-схеме.

Добавим в наш сервер два ресурса: системную информацию сервера и конфигурацию в формате JSON:

#!/usr/bin/env python3
# mcp_full_server.py — полный MCP-сервер с Tools и Resources

import asyncio
import hashlib
import logging
import json
import platform
import os
from datetime import datetime

import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent, Resource
from mcp.server.stdio import stdio_server

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-full")

server = Server("full-mcp-server")

# ── RESOURCES ──

@server.list_resources()
async def handle_list_resources() -> list:
    return [
        Resource(
            uri="system://info",
            name="Системная информация",
            description="ОС, архитектура, версия Python и текущее время сервера",
            mimeType="application/json",
        ),
        Resource(
            uri="config://server",
            name="Конфигурация сервера",
            description="Текущая конфигурация MCP-сервера: имя, окружение",
            mimeType="application/json",
        ),
    ]


@server.read_resource()
async def handle_read_resource(uri: str) -> str:
    logger.info(f"Запрошен ресурс: {uri}")

    if uri == "system://info":
        data = {
            "os": platform.system(),
            "arch": platform.machine(),
            "python_version": platform.python_version(),
            "server_time": datetime.now().isoformat(),
            "pid": os.getpid(),
        }
        return json.dumps(data, indent=2, ensure_ascii=False)

    elif uri == "config://server":
        data = {
            "server_name": "full-mcp-server",
            "version": "1.0.0",
            "environment": os.environ.get("MCP_ENV", "development"),
            "features": ["tools", "resources"],
        }
        return json.dumps(data, indent=2, ensure_ascii=False)

    else:
        raise ValueError(f"Ресурс не найден: {uri}")


# ── TOOLS ──

TOOLS = [
    Tool(
        name="get_crypto_price",
        description="Получить текущую цену криптовалюты в USD (BTC, ETH, SOL и др.)",
        inputSchema={
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "Тикер (bitcoin, ethereum, solana)"},
            },
            "required": ["symbol"],
        },
    ),
    Tool(
        name="hash_content",
        description="SHA-256 хеш строки. Полезно для проверки целостности.",
        inputSchema={
            "type": "object",
            "properties": {
                "content": {"type": "string", "description": "Строка для хеширования"},
            },
            "required": ["content"],
        },
    ),
]


@server.list_tools()
async def handle_list_tools() -> list:
    return TOOLS


@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if name == "get_crypto_price":
        symbol = arguments["symbol"].lower()
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(
                f"https://api.coingecko.com/api/v3/simple/price?ids={symbol}&vs_currencies=usd"
            )
            data = resp.json()
        price = data.get(symbol, {}).get("usd")
        if price:
            text = f"💰 {symbol.upper()}: ${price:,.2f} USD"
        else:
            text = f"❌ '{symbol}' не найдена. Попробуйте: bitcoin, ethereum, solana"
        return [TextContent(type="text", text=text)]

    elif name == "hash_content":
        content = arguments["content"]
        sha = hashlib.sha256(content.encode()).hexdigest()
        return [TextContent(type="text", text=f"🔐 SHA-256: {sha}")]

    raise ValueError(f"Неизвестный инструмент: {name}")


async def main():
    async with stdio_server() as (read_stream, write_stream):
        logger.info("🚀 Full MCP Server запущен: 2 resources + 2 tools")
        await server.run(read_stream, write_stream, server.create_initialization_options())


if __name__ == "__main__":
    asyncio.run(main())

Как видите, Resources и Tools прекрасно уживаются в одном сервере. Модель может сначала запросить system://info для контекста, а затем вызвать get_crypto_price для получения данных — всё через один и тот же JSON-RPC канал.

6. Подключение к Claude Desktop

Самый захватывающий момент — подключить наш сервер к Claude Desktop и увидеть, как AI-агент использует наши инструменты в реальном времени. Для этого нужно отредактировать конфигурационный файл Claude Desktop:

Путь к конфигурации:

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json
• Linux: ~/.config/Claude/claude_desktop_config.json

Добавляем наш сервер в секцию mcpServers:

{
  "mcpServers": {
    "crypto-tools": {
      "command": "python",
      "args": ["/absolute/path/to/mcp_full_server.py"],
      "env": {
        "MCP_ENV": "production"
      }
    }
  }
}

После сохранения конфигурации полностью перезапустите Claude Desktop. Если всё настроено правильно, вы увидите иконку молотка 🔨 в интерфейсе Claude — это означает, что MCP-инструменты подключены. Теперь в чате можно написать:

# Запросы к вашему MCP-серверу через Claude Desktop:
"Сколько сейчас стоит Bitcoin?"
"Посчитай SHA-256 для строки 'hello world'"
"Какая информация о системе на сервере?"

Claude автоматически определит, какой инструмент или ресурс использовать, вызовет его через MCP-клиент и включит результат в свой ответ. Вы не пишете код на стороне Claude — только сервер.

Отладка: если инструменты не появились, проверьте логи Claude Desktop. На macOS они находятся в ~/Library/Logs/Claude/. Ищите ошибки вида MCP server connection failed. Частая проблема — не установлен пакет mcp в том Python-окружении, которое используется командой из конфига.

Ещё один важный нюанс: Claude Desktop кэширует список инструментов после инициализации. Если вы изменили код сервера и добавили новый инструмент, недостаточно просто перезапустить серверный процесс — нужно полностью перезапустить само приложение Claude Desktop, чтобы произошла повторная инициализация MCP-соединения и клиент заново запросил список инструментов через tools/list. Это особенность архитектуры: Host управляет жизненным циклом клиента, и только полный перезапуск Host гарантирует новую сессию MCP.

Если вы планируете использовать сервер в production-окружении или делиться им с командой, рекомендую упаковать его как Python-пакет с точкой входа console_scripts в pyproject.toml. Тогда в конфиге Claude Desktop можно будет указать просто имя команды, а не полный путь к скрипту — это делает установку значительно проще для конечных пользователей.

7. Продвинутые паттерны и лучшие практики

7.1. Динамическая регистрация инструментов

В реальных проектах вы вряд ли захотите хардкодить список инструментов. Вот паттерн с классом-реестром, который позволяет добавлять инструменты из разных модулей:

import asyncio
from dataclasses import dataclass, field
from typing import Callable, Any
from mcp.types import Tool, TextContent
from mcp.server import Server

# Реестр инструментов
class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, tuple[Tool, Callable]] = {}

    def register(self, tool: Tool, handler: Callable):
        self._tools[tool.name] = (tool, handler)

    def list_tools(self) -> list[Tool]:
        return [t for t, _ in self._tools.values()]

    async def call_tool(self, name: str, args: dict) -> list:
        _, handler = self._tools[name]
        result = await handler(**args)
        return [TextContent(type="text", text=str(result))]

registry = ToolRegistry()

# Регистрируем инструмент из любого модуля
registry.register(
    Tool(
        name="multiply",
        description="Умножить два числа",
        inputSchema={
            "type": "object",
            "properties": {
                "a": {"type": "number"},
                "b": {"type": "number"},
            },
            "required": ["a", "b"],
        },
    ),
    lambda a, b: f"✅ {a} × {b} = {a * b}",
)

server = Server("dynamic-tools")

@server.list_tools()
async def list_tools():
    return registry.list_tools()

@server.call_tool()
async def call_tool(name, arguments):
    return await registry.call_tool(name, arguments)

7.2. HTTP/SSE транспорт для удалённых серверов

Если ваш MCP-сервер должен быть доступен по сети (а не только как дочерний процесс), используйте SSE-транспорт. Это полезно для командной работы, когда сервер развёрнут на удалённой машине:

# server_sse.py — MCP-сервер через HTTP SSE
import asyncio
import logging
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

server = Server("remote-server")
sse = SseServerTransport("/messages/")

async def handle_sse(request):
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        await server.run(
            streams[0], streams[1],
            server.create_initialization_options(),
        )

app = Starlette(routes=[
    Route("/sse", endpoint=handle_sse),
    Route("/messages/", endpoint=sse.handle_post_message, methods=["POST"]),
])

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8765)

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

Никогда не доверяйте аргументам, приходящим от AI-модели. Модель может передать неожиданные типы, пустые строки или отсутствующие ключи. Всегда валидируйте:

def safe_get_arg(arguments: dict, key: str, expected_type=str, default=None):
    value = arguments.get(key)
    if value is None:
        if default is not None:
            return default
        raise ValueError(f"Отсутствует обязательный аргумент: {key}")
    if not isinstance(value, expected_type):
        raise TypeError(f"Аргумент {key} должен быть {expected_type.__name__}, получен {type(value).__name__}")
    return value

7.4. Множественные MCP-серверы

Claude Desktop поддерживает одновременное подключение нескольких серверов. Вы можете разделить ответственность: один сервер для работы с базой данных, второй — для API-интеграций, третий — для файловой системы. Каждый сервер получает свой изолированный процесс и независимый канал stdio. Конфигурация для нескольких серверов выглядит так:

{
  "mcpServers": {
    "database": {
      "command": "python",
      "args": ["/path/to/db_server.py"]
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-filesystem", "/allowed/dir"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_..."
      }
    }
  }
}

8. Сравнение MCP с традиционным Function Calling

Если вы работали с OpenAI Function Calling или аналогичными механизмами, у вас может возникнуть вопрос: чем MCP отличается и зачем он нужен? Вот ключевые различия:

Вывод: MCP не заменяет Function Calling на уровне API-провайдера, а работает на уровень выше — как универсальный протокол подключения инструментов к AI-агентам. Вы пишете MCP-сервер один раз, а использовать его может любая MCP-совместимая модель. Это принципиально другая архитектура: не «модель вызывает функцию», а «модель через хост взаимодействует с сервером инструментов».

При этом важно понимать, что MCP и Function Calling могут сосуществовать. Например, вы можете использовать MCP для предоставления инструментов конечному пользователю в Claude Desktop, а параллельно реализовать Function Calling для программного вызова тех же инструментов через API. Более того, сам MCP-клиент внутри Claude Desktop, получив список инструментов от сервера, фактически транслирует их в формат, понятный модели — то есть MCP является прослойкой стандартизации, а не заменой низкоуровневого механизма вызова функций. Это значит, что ваши инвестиции в написание MCP-сервера окупаются дважды: вы получаете и готовую интеграцию с десктопными приложениями, и возможность программного использования через Function Calling API.