simbiot-harness/artifacts/shkor/connector.py

#!/usr/bin/env python3
"""Shkor Beauty Assistant — универсальный коннектор.

Один и тот же ассистент — три способа подключения для ЛЮБОГО агента:

  1) STDIO MCP-стиль (Claude/любой MCP-клиент):
       assistant → connector.py                (JSONL: {"method":"tools/list"} по строке)
     в конфиге агента:
       {"mcpServers": {"shkor": {"command": "python3", "args": [".../connector.py"], "transport": "stdio"}}}

  2) HTTP JSON API (агент = просто HTTP-клиент):
       connector.py --http 127.0.0.1:8791
       POST /tools/list | POST /call {"name": "...", "arguments": {...}}

  3) CLI разовый вызов (curl/крон/скрипт):
       connector.py --call analyze_skin --args '{"self_report":"стянутость и покраснение"}'

  4) Демо-чат (для живой проверки человеком):
       connector.py --demo

Форма запроса принимает MCP tools/call и простую {action, args} — см. assistant.handle().
"""
from __future__ import annotations

import argparse
import json
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from assistant import handle  # noqa: E402


# ── 1) STDIO: JSONL, одна строка = один запрос ───────────────────────────────
def serve_stdio() -> None:
    print(json.dumps({"ok": True, "transport": "stdio-jsonl",
                      "hint": "отправляйте JSON-строки, читайте ответ построчно"}, ensure_ascii=False), flush=True)
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            resp = handle(json.loads(line))
        except json.JSONDecodeError as e:
            resp = {"error": f"bad json: {e}"}
        print(json.dumps(resp, ensure_ascii=False), flush=True)


# ── 2) HTTP ──────────────────────────────────────────────────────────────────
def serve_http(addr: str) -> None:
    host, port = addr.split(":")
    port_i = int(port)

    class H(BaseHTTPRequestHandler):
        def _send(self, code: int, payload: dict) -> None:
            body = json.dumps(payload, ensure_ascii=False).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Access-Control-Allow-Origin", "*")  # витрина на другом домене: форма заявки → /call
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type")
            self.end_headers()
            self.wfile.write(body)

        def do_OPTIONS(self) -> None:  # preflight для формы витрины (fetch с другого origin)
            self.send_response(204)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type")
            self.send_header("Content-Length", "0")
            self.end_headers()

        def do_GET(self) -> None:  # health / открыть в браузере
            if self.path in ("/", "/health"):
                self._send(200, {"ok": True, "service": "shkor-assistant", "version": 1,
                                 "tools": "POST /tools/list", "call": "POST /call"})
            else:
                self._send(404, {"error": "not found", "routes": ["POST /tools/list", "POST /call"]})

        def do_POST(self) -> None:
            try:
                n = int(self.headers.get("Content-Length") or 0)
                data = json.loads(self.rfile.read(n) or b"{}")
            except Exception as e:  # noqa: BLE001
                self._send(400, {"error": f"bad request: {e}"})
                return
            if self.path == "/tools/list":
                self._send(200, handle({"method": "tools/list"}))
            elif self.path == "/call":
                self._send(200, handle({"method": "tools/call", "params": data}))
            else:
                self._send(404, {"error": "not found", "routes": ["POST /tools/list", "POST /call"]})

        def log_message(self, *a) -> None:  # тише в логи
            pass

    srv = ThreadingHTTPServer((host, port_i), H)
    print(f"shkor-assistant HTTP on http://{host}:{port_i}  (POST /tools/list, POST /call)", file=sys.stderr)
    srv.serve_forever()


# ── 4) Демо-чат ───────────────────────────────────────────────────────────────
DEMO = [
    ("analyze_skin", {"self_report": "кожа стянутая после умывания, лоб блестит к обеду, периодически покраснения"}),
    ("recommend", {"concerns": ["dryness", "redness"], "skin_type": "sensitive", "budget_max": 70}),
    ("place_order", {"items": [{"id": "p05", "qty": 1}, {"id": "p03", "qty": 1}], "delivery": "minsk_courier",
                     "name": "Тест", "phone": "+375XX0000000"}),
    ("usage_instructions", {"product_ids": ["p05", "p03"]}),
    ("check_order", {}),  # order_id подставится после place_order
]


def run_demo() -> None:
    last_order = None
    for action, args in DEMO:
        if action == "check_order":
            if not last_order:
                continue
            args = {"order_id": last_order}
        print(f"\n▶ {action} {json.dumps(args, ensure_ascii=False)}")
        resp = handle({"method": "tools/call", "params": {"name": action, "arguments": args}})
        print(json.dumps(resp.get("result", resp), ensure_ascii=False, indent=1)[:1200])
        if action == "place_order" and (resp.get("result") or {}).get("order_id"):
            last_order = resp["result"]["order_id"]


def main() -> None:
    ap = argparse.ArgumentParser(description="Shkor Beauty Assistant connector")
    ap.add_argument("--http", metavar="HOST:PORT", help="поднять HTTP API")
    ap.add_argument("--stdio", action="store_true", help="STDIO JSONL (MCP-стиль)")
    ap.add_argument("--call", metavar="TOOL", help="разовый вызов инструмента")
    ap.add_argument("--args", default="{}", help="JSON-аргументы для --call")
    ap.add_argument("--list-tools", action="store_true")
    ap.add_argument("--demo", action="store_true")
    a = ap.parse_args()

    if a.http:
        serve_http(a.http)
    elif a.stdio:
        serve_stdio()
    elif a.list_tools:
        print(json.dumps(handle({"method": "tools/list"}), ensure_ascii=False, indent=1))
    elif a.demo:
        run_demo()
    elif a.call:
        try:
            arguments = json.loads(a.args)
        except json.JSONDecodeError as e:
            sys.exit(f"bad --args json: {e}")
        print(json.dumps(handle({"method": "tools/call", "params": {"name": a.call, "arguments": arguments}}),
                         ensure_ascii=False, indent=1))
    else:
        ap.print_help()


if __name__ == "__main__":
    main()