dz/solutions/69f8e929da860fb4533faa2a_MCP_сервер_для_управления_памятью_агента

MCP‑сервер для управления памятью агента

Описание проекта

MCP (Memory Control Protocol) – это протокол, позволяющий удалённо управлять памятью агентов в распределённых системах.
Данный репозиторий содержит реализацию MCP‑сервера, который:

• принимает команды от клиентов по TCP/IP;
• обрабатывает запросы на чтение/запись памяти агента;
• поддерживает простую аутентификацию и логирование.

Сервер написан на Python 3.8+ и использует асинхронную библиотеку asyncio для одновременной обработки множества соединений.

Структура проекта

mcp_server/
├── mcp_server.py          # Основной серверный скрипт
├── client_example.py      # Пример клиента, демонстрирующий работу с сервером
├── requirements.txt       # Список зависимостей
└── README.md              # Это файл

Установка зависимостей

# Создайте виртуальное окружение (рекомендуется)
python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate

# Установите зависимости
pip install -r requirements.txt

requirements.txt содержит только стандартные библиотеки, но в случае необходимости можно добавить сторонние пакеты.

Запуск сервера

python mcp_server.py --host 0.0.0.0 --port 9000

Параметры:

Параметр	Описание	По умолчанию
--host	IP‑адрес, на котором сервер будет слушать подключения	127.0.0.1
--port	TCP‑порт для входящих соединений	9000

Сервер выводит в консоль информацию о подключениях и выполненных командах.

Запуск примера клиента

python client_example.py --host 127.0.0.1 --port 9000

Клиент демонстрирует:

1. Подключение к серверу;
2. Отправку команды SET key value;
3. Чтение значения через команду GET key;
4. Вывод результата в консоль.

Пример использования

# client_example.py
import asyncio
import argparse

async def main(host, port):
    reader, writer = await asyncio.open_connection(host, port)

    # Установим значение
    writer.write(b"SET counter 42\n")
    await writer.drain()
    print(await reader.readline())  # OK

    # Получим значение
    writer.write(b"GET counter\n")
    await writer.drain()
    print(await reader.readline())  # 42

    writer.close()
    await writer.wait_closed()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="MCP client example")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=9000)
    args = parser.parse_args()
    asyncio.run(main(args.host, args.port))

Тесты

Для запуска тестов (если они добавлены) используйте:

pytest tests/

Лицензия

MIT License – см. файл LICENSE.

---

Если возникнут вопросы или предложения по улучшению – открывайте issue!