omnichannel-configserver-mcp/examples/demo/demo.py

#!/usr/bin/env python3
"""Сквозной демо-пример использования omnichannel-configserver-mcp.

Что делает:
  1) засевает демо-сервис в локальный config_server (агентский эндпоинт);
  2) запускает omnichannel-configserver-mcp по stdio;
  3) выполняет типичный сценарий эксплуатации/деплоя:
     server_info -> list_applications -> get_application -> set_env ->
     set_compose -> deploy -> get_task -> overview.

Зависимостей нет (только stdlib). Требуется поднятый стенд:
    docker compose -f examples/demo/docker-compose.yml up -d
и собранный бинарник:
    go build -o omnichannel-configserver-mcp .

Запуск: python3 examples/demo/demo.py
"""
import json
import os
import select
import subprocess
import sys
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(os.path.dirname(HERE))  # корень репозитория
BINARY = os.path.join(ROOT, "omnichannel-configserver-mcp")
CONFIG = os.path.join(HERE, "config.demo.json")

BASE_URL = "http://127.0.0.1:5005"
AGENT_SECRET = "demo-agent-secret"
DEMO_APP_ID = 1


def http_json(method, path, payload=None):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(BASE_URL + path, data=data, method=method,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        return json.loads(resp.read().decode())


def seed_demo_app():
    """Регистрирует демо-сервис так, как это делает агент ноды."""
    body = {
        "secret": AGENT_SECRET,
        "name": "demo_web",
        "host_ip": "10.0.0.10",
        "hostname": "demo-node",
        "path": "/opt/omnichannel/demo_web",
        "status": "OK",
        "compose_content": "services:\n  web:\n    image: nginx:1.25\n",
        "env_files": {".env": "FOO=bar\n"},
    }
    result = http_json("POST", "/api/register", body)
    print(f"засеян демо-сервис demo_web: app_id={result.get('app_id')}")


class MCP:
    """Минимальный MCP-клиент по stdio (JSON-RPC, newline-delimited)."""

    def __init__(self):
        env = dict(os.environ, OMNI_LOG_LEVEL="warn")
        self.proc = subprocess.Popen(
            [BINARY, "-config", CONFIG], cwd=ROOT, env=env,
            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, bufsize=1,
        )
        self.id = 0
        self._request("initialize", {
            "protocolVersion": "2025-06-18", "capabilities": {},
            "clientInfo": {"name": "demo", "version": "0"},
        })
        self._notify("notifications/initialized", {})

    def _id(self):
        self.id += 1
        return self.id

    def _send(self, obj):
        self.proc.stdin.write(json.dumps(obj) + "\n")
        self.proc.stdin.flush()

    def _recv(self, want_id, timeout=30):
        while True:
            ready, _, _ = select.select([self.proc.stdout], [], [], timeout)
            if not ready:
                raise TimeoutError(f"нет ответа на id={want_id}")
            line = self.proc.stdout.readline()
            if not line:
                raise RuntimeError("сервер закрыл stdout:\n" + self.proc.stderr.read())
            msg = json.loads(line)
            if msg.get("id") == want_id:
                return msg

    def _request(self, method, params):
        rid = self._id()
        self._send({"jsonrpc": "2.0", "id": rid, "method": method, "params": params})
        return self._recv(rid)

    def _notify(self, method, params):
        self._send({"jsonrpc": "2.0", "method": method, "params": params})

    def tools(self):
        return self._request("tools/list", {})["result"]["tools"]

    def call(self, name, arguments):
        resp = self._request("tools/call", {"name": name, "arguments": arguments})
        result = resp.get("result", {})
        text = "".join(c.get("text", "") for c in result.get("content", []))
        return text, result.get("isError", False)

    def close(self):
        self.proc.stdin.close()
        self.proc.terminate()


def show(title, name, arguments, mcp):
    text, is_error = mcp.call(name, arguments)
    print(f"\n=== {title} ({'ошибка' if is_error else 'ok'}) ===")
    print(text[:600])


def main():
    seed_demo_app()

    mcp = MCP()
    tools = mcp.tools()
    print(f"\nMCP-сервер отдал инструментов: {len(tools)}")

    show("Версия API и режим", "server_info", {}, mcp)
    show("Список сервисов", "list_applications", {}, mcp)
    show("Детали демо-сервиса", "get_application", {"app_id": DEMO_APP_ID}, mcp)

    show("Обновляем env", "set_env",
         {"app_id": DEMO_APP_ID, "filename": ".env", "content": "FOO=baz\n"}, mcp)
    show("Обновляем compose", "set_compose",
         {"app_id": DEMO_APP_ID, "content": "services:\n  web:\n    image: nginx:1.27\n"}, mcp)

    deploy_text, _ = mcp.call("deploy", {"app_id": DEMO_APP_ID})
    print("\n=== Запуск сервиса (deploy) ===")
    print(deploy_text[:400])

    try:
        task_id = json.loads(deploy_text)["task_id"]
        show("Статус задачи деплоя", "get_task", {"task_id": task_id}, mcp)
    except (ValueError, KeyError):
        print("(не удалось извлечь task_id — пропускаем get_task)")

    show("Сводка состояния стенда", "overview", {}, mcp)

    mcp.close()
    print("\nDemo завершён.")


if __name__ == "__main__":
    if not os.path.exists(BINARY):
        sys.exit(f"нет бинарника {BINARY}: сначала выполните `go build -o omnichannel-configserver-mcp .`")
    main()