commit

app.py

@@ -0,0 +1,260 @@
+import os
+import re
+from dataclasses import dataclass
+from typing import Dict, Iterable, List, Optional, Tuple
+
+import gradio as gr
+import requests
+
+
+class ToolCallError(RuntimeError):
+    """Raised when a tool invocation fails even after retry."""
+
+
+@dataclass
+class StorageHit:
+    file_id: Optional[str]
+    file_name: Optional[str]
+    chunk_id: Optional[str]
+    score: Optional[float]
+    snippet: Optional[str]
+
+
+class OpenAIStorageClient:
+    """Minimal client for OpenAI Storage search/get endpoints."""
+
+    def __init__(self, api_key: str, base_url: Optional[str] = None, timeout: float = 15.0):
+        if not api_key:
+            raise ValueError("api_key is required")
+        self.api_key = api_key
+        base = base_url or os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1")
+        self.base_url = base.rstrip("/")
+        self.timeout = timeout
+
+    def _post(self, path: str, payload: Dict) -> Dict:
+        url = f"{self.base_url}{path}"
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+        try:
+            response = requests.post(url, json=payload, headers=headers, timeout=self.timeout)
+        except requests.RequestException as exc:
+            raise ToolCallError(str(exc)) from exc
+
+        if response.status_code >= 400:
+            raise ToolCallError(f"Tool call failed with status {response.status_code}: {response.text}")
+
+        try:
+            return response.json()
+        except ValueError as exc:
+            raise ToolCallError("Tool call returned non-JSON response") from exc
+
+    def storage_search(
+        self,
+        query: str,
+        top_k: int = 5,
+        filters: Optional[Dict] = None,
+        retries: int = 1,
+    ) -> List[StorageHit]:
+        payload: Dict[str, object] = {"query": query, "top_k": top_k}
+        if filters:
+            payload["filters"] = filters
+
+        attempts = 0
+        while True:
+            try:
+                raw = self._post("/tools/storage_search", payload)
+                hits = []
+                for item in raw.get("hits", []):
+                    hits.append(
+                        StorageHit(
+                            file_id=item.get("file_id"),
+                            file_name=item.get("file_name"),
+                            chunk_id=item.get("chunk_id"),
+                            score=item.get("score"),
+                            snippet=item.get("snippet"),
+                        )
+                    )
+                return hits
+            except ToolCallError:
+                attempts += 1
+                if attempts > retries:
+                    raise
+
+    def storage_get(
+        self,
+        *,
+        chunk_id: Optional[str] = None,
+        file_id: Optional[str] = None,
+        retries: int = 1,
+    ) -> Optional[str]:
+        if not chunk_id and not file_id:
+            raise ValueError("Either chunk_id or file_id must be provided.")
+
+        payload: Dict[str, object] = {}
+        if chunk_id:
+            payload["chunk_id"] = chunk_id
+        if file_id:
+            payload["file_id"] = file_id
+
+        attempts = 0
+        while True:
+            try:
+                raw = self._post("/tools/storage_get", payload)
+                content = raw.get("content")
+                if isinstance(content, str):
+                    return content
+                return None
+            except ToolCallError:
+                attempts += 1
+                if attempts > retries:
+                    raise
+
+
+def generate_search_queries(question: str) -> List[str]:
+    """Generate up to three search queries following the spec."""
+    normalized = question.strip()
+    if not normalized:
+        return []
+
+    queries: List[str] = [normalized]
+
+    ascii_tokens = re.findall(r"[A-Za-z0-9]+", normalized)
+    if ascii_tokens:
+        english_query = " ".join(ascii_tokens)
+        if english_query and english_query.lower() != normalized.lower():
+            queries.append(english_query)
+
+    japanese_clean = re.sub(r"[、。；；，]", " ", normalized)
+    japanese_clean = re.sub(r"\s+", " ", japanese_clean).strip()
+    if japanese_clean and japanese_clean not in queries:
+        queries.append(japanese_clean)
+
+    return queries[:3]
+
+
+def deduplicate_hits(hits_by_query: Iterable[Tuple[str, List[StorageHit]]]) -> List[StorageHit]:
+    """Merge and deduplicate hits by chunk/file IDs while preserving ordering."""
+    seen_keys = set()
+    merged: List[StorageHit] = []
+    for _, hits in hits_by_query:
+        for hit in hits:
+            key = (hit.chunk_id, hit.file_id)
+            if key in seen_keys:
+                continue
+            seen_keys.add(key)
+            merged.append(hit)
+    return merged
+
+
+def build_excerpt(text: str, limit: int = 280) -> str:
+    """Return a compact excerpt trimmed to the requested limit."""
+    collapsed = re.sub(r"\s+", " ", text).strip()
+    if len(collapsed) <= limit:
+        return collapsed
+    return collapsed[: limit - 1] + "…"
+
+
+def produce_answer(question: str, top_k: int) -> str:
+    """Main orchestration pipeline for the MCP responder."""
+    question = (question or "").strip()
+    if not question:
+        return "質問が入力されていません。"
+
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        return "OPENAI_API_KEY が設定されていません。環境変数を確認してください。"
+
+    client = OpenAIStorageClient(api_key=api_key)
+
+    queries = generate_search_queries(question)
+    if not queries:
+        return "該当データがありません。"
+
+    try:
+        hits_by_query = []
+        for query in queries:
+            hits = client.storage_search(query=query, top_k=top_k)
+            if hits:
+                hits_by_query.append((query, hits))
+
+        if not hits_by_query:
+            return "該当データがありません。"
+
+        merged_hits = deduplicate_hits(hits_by_query)
+        evidence_lines: List[str] = []
+
+        for hit in merged_hits[:3]:
+            content: Optional[str] = None
+            try:
+                if hit.chunk_id:
+                    content = client.storage_get(chunk_id=hit.chunk_id)
+                if not content and hit.file_id:
+                    content = client.storage_get(file_id=hit.file_id)
+            except ToolCallError:
+                # Propagate after exceeding retry budget according to spec.
+                return "システムエラーが発生しました。時間をおいて再度お試しください。"
+
+            source_text = content or hit.snippet
+            if not source_text:
+                continue
+
+            excerpt = build_excerpt(source_text)
+            if not excerpt:
+                continue
+
+            file_ref = hit.file_name or (hit.file_id or "unknown")
+            chunk_ref = hit.chunk_id or "chunk"
+            evidence_lines.append(f"- {excerpt} [{file_ref}#{chunk_ref}]")
+
+        if not evidence_lines:
+            return "該当データがありません。"
+
+        return "\n".join(evidence_lines)
+
+    except ToolCallError:
+        return "システムエラーが発生しました。時間をおいて再度お試しください。"
+
+
+def respond(question: str, top_k: int = 5) -> str:
+    """Wrapper for the Gradio UI."""
+    try:
+        return produce_answer(question, top_k=int(top_k))
+    except Exception:  # pragma: no cover - defensive fallback for UI
+        return "システムエラーが発生しました。時間をおいて再度お試しください。"
+
+
+with gr.Blocks() as demo:
+    gr.Markdown(
+        """
+        ## MCP Storage レスポンダ
+        OpenAI Storage (Files / Vector Store) に登録されたデータを検索し、根拠付きで回答します。
+        OPENAI_API_KEY を環境変数に設定してからご利用ください。
+        """
+    )
+
+    with gr.Row():
+        question_box = gr.Textbox(
+            label="質問",
+            placeholder="ここに質問を入力してください（例: 製品Aの対応OSは？）",
+            lines=4,
+        )
+    with gr.Row():
+        top_k_slider = gr.Slider(
+            minimum=1,
+            maximum=10,
+            value=5,
+            step=1,
+            label="検索件数 (top_k)",
+        )
+    with gr.Row():
+        output_box = gr.Textbox(label="回答", lines=10)
+    with gr.Row():
+        submit_button = gr.Button("検索")
+
+    submit_button.click(respond, inputs=[question_box, top_k_slider], outputs=output_box)
+
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,3 @@
+gradio==5.49.1
+openai>=1.51.0
+requests>=2.32.3