ExperionCrawler/mcp-server/server.py

#!/usr/bin/env python3
"""
ExperionCrawler RAG MCP Server
- 임베딩: Ollama nomic-embed-text (768-dim) — Roo Code 인덱스와 동일 모델
- 벡터 DB: Qdrant localhost:6333
- LLM: vLLM GLM-4.7-Flash localhost:8000/v1
- 사용처: Claude Code MCP / Roo Code MCP (동일 서버)
"""

from __future__ import annotations
import sys
import logging
import httpx
from functools import lru_cache
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.WARNING, stream=sys.stderr)

# ── 설정 ──────────────────────────────────────────────────────────────────────
QDRANT_URL      = "http://localhost:6333"
OLLAMA_URL      = "http://localhost:11434"
EMBED_MODEL     = "nomic-embed-text"        # 768-dim, Roo Code 인덱스와 동일
VLLM_BASE_URL   = "http://localhost:8000/v1"
VLLM_MODEL      = "glm-4.7-flash"

# Qdrant 컬렉션
COL_CODEBASE    = "ws-65f457145aee80b2"     # ExperionCrawler 소스코드
COL_OPC_DOCS    = "experion-opc-docs"       # Experion HS R530 OPC UA 공식 문서 (266 chunks)

mcp = FastMCP("iiot-rag")

# ── 임베딩 (Ollama) ───────────────────────────────────────────────────────────

def _embed(text: str) -> list[float]:
    """Ollama nomic-embed-text로 768-dim 벡터 생성."""
    with httpx.Client(timeout=30) as client:
        resp = client.post(
            f"{OLLAMA_URL}/api/embeddings",
            json={"model": EMBED_MODEL, "prompt": text},
        )
        resp.raise_for_status()
        return resp.json()["embedding"]

# ── LLM (vLLM / GLM-4.7-Flash) ───────────────────────────────────────────────

@lru_cache(maxsize=1)
def _llm():
    from openai import OpenAI
    return OpenAI(base_url=VLLM_BASE_URL, api_key="dummy")

# ── Qdrant 검색 헬퍼 ──────────────────────────────────────────────────────────

def _search(collection: str, query: str, top_k: int, threshold: float = 0.25) -> str:
    vec = _embed(query)
    with httpx.Client(timeout=20) as client:
        resp = client.post(
            f"{QDRANT_URL}/collections/{collection}/points/search",
            json={
                "vector": vec,
                "limit": top_k,
                "with_payload": True,
                "score_threshold": threshold,
            },
        )
        resp.raise_for_status()
        hits = resp.json().get("result", [])

    if not hits:
        return "관련 결과 없음."

    parts = []
    for h in hits:
        p = h.get("payload", {})
        file_path  = p.get("filePath", p.get("path", "unknown"))
        chunk      = p.get("codeChunk", p.get("content", p.get("text", "")))
        start_line = p.get("startLine", "")
        loc = f"{file_path}:{start_line}" if start_line else file_path
        parts.append(f"[score={h['score']:.3f}] {loc}\n```\n{chunk[:700]}\n```")

    return "\n\n---\n\n".join(parts)

# ── MCP 도구 ─────────────────────────────────────────────────────────────────

@mcp.tool()
def search_codebase(query: str, top_k: int = 6) -> str:
    """ExperionCrawler 프로젝트 소스코드 검색 (우리가 개발한 .NET 8 C# 코드).
    Experion HS R530 공식 문서가 아닌, ExperionCrawler 구현 코드를 검색함.

    사용 시점: ExperionCrawler 코드의 구현 방법, 버그, 구조를 알고 싶을 때.
    ⚠️  Experion HS R530 제품 동작/설정/스펙을 알고 싶으면 search_r530_docs 사용.

    Args:
        query: 검색어 (예: "OPC UA 구독 시작", "히스토리 스냅샷", "TextToSql 서비스")
        top_k: 반환 결과 수 (기본 6)
    """
    return _search(COL_CODEBASE, query, top_k)


@mcp.tool()
def search_r530_docs(query: str, top_k: int = 5) -> str:
    """Honeywell Experion HS R530 공식 제품 문서 검색.
    ExperionCrawler 코드가 아닌, Honeywell 공식 HTM 문서를 검색함.

    사용 시점: Experion HS R530의 OPC UA 설정, 인증서, 보안 정책, 포인트 주소 형식,
    채널/컨트롤러 속성, 문제해결 등 제품 스펙과 동작을 알고 싶을 때.
    ⚠️  ExperionCrawler 구현 코드를 찾으려면 search_codebase 사용.

    Args:
        query: 검색어 (예: "certificate configuration", "endpoint security policy", "point address syntax")
        top_k: 반환 결과 수 (기본 5)
    """
    return _search(COL_OPC_DOCS, query, top_k)


@mcp.tool()
def ask_iiot_llm(question: str, context: str = "") -> str:
    """GLM-4.7-Flash에게 IIoT/OPC UA 질문 (컨텍스트 없이 LLM 직접 질문).

    사용 시점: search_codebase 또는 search_r530_docs 결과를 context로 넘겨
    종합 분석·답변이 필요할 때. 또는 일반 IIoT/OPC UA 개념 질문.

    Args:
        question: 질문 내용
        context:  (선택) search_codebase 또는 search_r530_docs 검색 결과
    """
    system = (
        "당신은 IIoT(산업용 IoT), OPC UA, Honeywell Experion PKS/HS R530 전문가입니다.\n"
        "컨텍스트가 제공된 경우 컨텍스트를 우선 근거로 삼아 한국어로 답변합니다.\n"
        "컨텍스트 출처가 'Experion HS R530 공식 문서'인지 'ExperionCrawler 코드'인지 명확히 구분하여 설명합니다."
    )
    user_msg = f"컨텍스트:\n{context}\n\n질문: {question}" if context else question
    resp = _llm().chat.completions.create(
        model=VLLM_MODEL,
        messages=[
            {"role": "system", "content": system},
            {"role": "user",   "content": user_msg},
        ],
        max_tokens=2048,
        temperature=0.1,
    )
    return resp.choices[0].message.content or "(응답 없음)"


@mcp.tool()
def rag_query(question: str, search_code: bool = False, search_docs: bool = True) -> str:
    """검색 → GLM-4.7-Flash 답변 생성 (통합 RAG).

    기본값: Experion HS R530 공식 문서만 검색 (search_docs=True, search_code=False).
    ExperionCrawler 코드도 함께 보려면 search_code=True 추가.

    사용 시점: Experion HS R530 제품 질문이나 ExperionCrawler 코드 질문에
    검색+LLM 답변을 한 번에 얻고 싶을 때.

    Args:
        question:    질문
        search_docs: Experion HS R530 공식 문서 검색 여부 (기본 True)
        search_code: ExperionCrawler 소스코드 검색 여부 (기본 False)
    """
    context_parts: list[str] = []
    if search_docs:
        context_parts.append(f"=== Experion HS R530 공식 문서 ===\n{_search(COL_OPC_DOCS, question, 4)}")
    if search_code:
        context_parts.append(f"=== ExperionCrawler 구현 코드 ===\n{_search(COL_CODEBASE, question, 3)}")

    return ask_iiot_llm(question, "\n\n".join(context_parts))


if __name__ == "__main__":
    mcp.run(transport="stdio")