mkdir -p ~/my-mcp-server/laya-mcp
cd ~/my-mcp-server/laya-mcp

python3 -m venv .venv
source .venv/bin/activate

pip install --upgrade pip
pip install "laya[serve,mcp]" transformers torch

import re, sys, time, torch
from typing import Any, Dict, Optional

_SAIHAI_MODEL = None
_SAIHAI_DEVICE = None

def is_japanese_text(text: str) -> bool:
    return bool(re.search(r"[\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FFF]", text)) if text else False

def get_saihai_model():
    global _SAIHAI_MODEL, _SAIHAI_DEVICE
    if _SAIHAI_MODEL is None:
        from transformers import AutoModel
        _SAIHAI_DEVICE = "mps" if torch.backends.mps.is_available() else "cpu"
        _SAIHAI_MODEL = AutoModel.from_pretrained(
            "hotchpotch/saihai-v1-310m",
            trust_remote_code=True,
            device=_SAIHAI_DEVICE,
            attn_implementation="sdpa",
        )
    return _SAIHAI_MODEL, _SAIHAI_DEVICE

def schema_to_saihai_questions(schema_dict: Dict[str, Any]) -> Dict[str, Any]:
    properties = schema_dict.get("properties", schema_dict)
    questions = {}
    for name, spec in properties.items():
        t = spec.get("type", "string")
        desc = spec.get("description", name)
        if t == "boolean":
            questions[name] = {"type": "noul", "instructions": f"{desc}の該当有無を判定する。"}
        elif t in ("enum", "string") and ("enum" in spec or "choices" in spec):
            choices = spec.get("enum", spec.get("choices", []))
            questions[name] = {"type": "choice", "instructions": f"{desc}を選択する。", "criteria": {str(c): f"{c}: 候補" for c in choices}}
        elif t == "integer" and "minimum" in spec and "maximum" in spec:
            questions[name] = {"type": "score", "instructions": f"{desc}のスコア評価。", "criteria": [f"レベル{i}" for i in range(spec["minimum"], spec["maximum"] + 1)]}
    return questions

def decide_with_saihai(context_str: str, schema_dict: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    model, device = get_saihai_model()
    if not model: return None
    try:
        t0 = time.perf_counter()
        res = model.systemone({"state": context_str, "questions": schema_to_saihai_questions(schema_dict)}, show_progress_bar=False)
        dur = (time.perf_counter() - t0) * 1000
        values = {k: v.get("choice") or v.get("noul") or v.get("score") for k, v in res.get("answers", {}).items()}
        return {"status": "success", "engine": "saihai-310m", "parallel_result": values, "parallel_latency": f"{dur:.1f}ms"}
    except Exception as e:
        return None

#!/usr/bin/env python3
import json, os, sys
from laya.mcp.server import server, _ensure_router, _wrap, _GUARDRAILS, _router_or_error
from laya.mcp.tools import laya_decide as official_laya_decide
from saihai_engine import decide_with_saihai, is_japanese_text

@server.tool(
    name="laya_decide",
    description="Plan A Hybrid Engine: routes Japanese to saihai-310m and English to Laya. " + _GUARDRAILS
)
def hybrid_laya_decide_tool(state: dict, schema: dict, model: str = "auto", min_confidence: float | None = None) -> str:
    text = " ".join(str(v) for v in state.values() if isinstance(v, str)) if isinstance(state, dict) else str(state)
    if is_japanese_text(text):
        res = decide_with_saihai(text, schema)
        if res is not None:
            return json.dumps(res, ensure_ascii=False)
    router = _router_or_error()
    return _wrap(official_laya_decide, state=state, schema=schema, model=model, min_confidence=min_confidence, router=router)

def main():
    _ensure_router()
    server.run()

if __name__ == "__main__":
    main()

"laya": {
  "command": "/Users//my-mcp-server/laya-mcp/.venv/bin/python",
  "args": ["/Users//my-mcp-server/laya-mcp/server.py"]
}