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

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

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

принимает команды от клиентов через TCP;
хранит и обновляет состояние памяти агента (ключ‑значение);
обеспечивает простую API для чтения, записи и удаления данных.

Сервер написан на Python 3.8+ и использует стандартную библиотеку socket + threading.
Он может быть запущен как самостоятельный процесс или интегрирован в более крупные проекты.

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

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

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

Важно: проект использует только стандартную библиотеку Python.
Если вы хотите использовать виртуальное окружение:

python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r 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

Пример клиентского кода (client_example.py) демонстрирует:

Подключение к серверу.
Отправку команды SET key value.
Чтение значения через команду GET key.
Удаление ключа с помощью DEL key.

# client_example.py (упрощённый фрагмент)
import socket

def send_command(sock, cmd):
    sock.sendall((cmd + '\n').encode())
    return sock.recv(4096).decode().strip()

with socket.create_connection(('127.0.0.1', 9000)) as s:
    print(send_command(s, 'SET foo bar'))   # -> OK
    print(send_command(s, 'GET foo'))       # -> bar
    print(send_command(s, 'DEL foo'))       # -> DELETED

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

# Внутри вашего приложения
import socket

class MCPClient:
    def __init__(self, host='127.0.0.1', port=9000):
        self.sock = socket.create_connection((host, port))

    def set(self, key, value):
        return self._send(f'SET {key} {value}')

    def get(self, key):
        return self._send(f'GET {key}')

    def delete(self, key):
        return self._send(f'DEL {key}')

    def _send(self, cmd):
        self.sock.sendall((cmd + '\n').encode())
        return self.sock.recv(4096).decode().strip()

Тестирование

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

python -m unittest discover tests

Замечание: в текущей версии тесты отсутствуют, но структура готова к их добавлению.

Лицензия

MIT License – смотрите файл LICENSE.

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

4.5 KiB Raw Blame History Unescape Escape