Building a Watsonx.ai Chatbot RAG Server with MCP

Watsonx.ai チャットボット RAG サーバーを MCP で構築する

このチュートリアルでは、IBM Watsonx.ai、ベクトルインデックス作成用の ChromaDB、そして Model Context Protocol（MCP）Python SDK を使用して、フル機能の検索拡張生成（RAG）サーバーを構築し、公開する方法を学びます。最後まで進めると、PDF ドキュメントを処理するスタンドアロンサーバーと、それらに対してクエリを実行するクライアントが完成します。特定のデータに基づく大規模言語モデルの力を活用できます。最後に、Claude Desktop 内に直接統合する方法を紹介します。

Model Context Protocol（MCP）は、アプリケーションと LLM の間のインターフェースを標準化します。MCP を使用すると、コンテキストの提供、コードの実行、ユーザーとのやり取りの管理を分離できます。MCP Python SDK は MCP 仕様全体を実装しており、次のことができます。

• リソースを公開する： LLM にデータを提供します（GET エンドポイントに相当します）。
• ツールを定義する： アクションや計算を実行する機能を提供します（POST エンドポイントに相当します）。
• プロンプトを作成する： 再利用可能なテンプレート化されたやり取りを提供します。 このサーバーでは、クライアントが呼び出せる RAG「ツール」を実装します。

前提条件

• Python 3.8 以降がインストールされていること（コードでは古いバージョンと互換性のある typing.Union を使用していますが、一般に新しい Python を推奨します）。
• プロビジョニング済みの Watsonx.ai サービスを利用できる IBM Cloud アカウントがあること。API キーとプロジェクト ID が必要です。
• Python パッケージをインストールするための pip。

プロジェクト構成

watsonx-rag-mcp-server/
├── .env
├── requirements.txt
├── documents/         # Your PDF files live here
├── client.py
└── server.py

ステップ 1：プロジェクトと環境のセットアップ

まず、プロジェクト専用のディレクトリを作成し、そのディレクトリに移動します。

仮想環境の使用を強くおすすめします。

python -m venv .venv
source .venv/bin/activate # On Windows use `.venv\Scripts\activate`

次に、プロジェクトのルートに .env ファイルを作成し、Watsonx.ai の認証情報と設定を安全に保存します。プレースホルダーの値を実際の認証情報に置き換えて、次の行を追加します。

# .env
WATSONX_APIKEY=your_watsonx_api_key_here
WATSONX_URL=your_watsonx_instance_url_here
PROJECT_ID=your_watsonx_project_id_here

# Optional: Specify different folder/DB paths if needed
# DOCS_FOLDER=./my_manuals
# CHROMA_PERSIST_DIR=./vector_db_storage
# Optional: Specify a different Watsonx.ai model
# MODEL_ID=meta-llama/llama-3-70b-instruct

ステップ 2：ドキュメントのセットアップ

RAG サーバーが情報を検索するには、ドキュメントが必要です。ディレクトリ（デフォルトでは documents という名前で、.env の設定または server.py のデフォルト設定と一致します）を作成し、その中に PDF ファイルを配置します。

mkdir documents
# --- Add your PDF files (e.g., drone manuals) into this 'documents' folder ---
# Example: cp /path/to/your/manual.pdf documents/

ステップ 3：サーバーの実装（server.py）

検索拡張生成（RAG）チャットボットの内部の仕組みを詳しく見ていきましょう。この Python スクリプト server.py は、PDF ドキュメントの処理、ベクトルデータベースの管理、IBM の Watsonx.ai 大規模言語モデル（LLM）とのやり取り、そして MCP（Meta Compute Protocol）サーバーを介したチャット機能の公開を担います。

目標は、提供された PDF ドキュメント（この例ではドローンのマニュアルなど）の内容 に特化して 質問に回答できるシステムを作ることです。server.py がこれをどのように実現するのか、詳しく見ていきましょう。

1. 準備：インポートと設定

どのようなプロジェクトでもそうであるように、必要なツールをインポートし、設定を行うところから始めます。

# server.py
import os
import sys
import logging
from pathlib import Path
import re
import textwrap
from dotenv import load_dotenv
from typing import Union # <--- IMPORT Union for older Python versions

# MCP Imports
from mcp.server.fastmcp import FastMCP

# IBM Watsonx.ai SDK Imports
from ibm_watsonx_ai import APIClient, Credentials
from ibm_watsonx_ai.foundation_models import ModelInference
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames as GenParams

# Vector DB Imports
import chromadb
from pypdf import PdfReader

# --- Configuration and Setup ---

# Load .env variables
load_dotenv()

# Watsonx Credentials and Config
API_KEY      = os.getenv("WATSONX_APIKEY")
URL          = os.getenv("WATSONX_URL")
PROJECT_ID   = os.getenv("PROJECT_ID")
MODEL_ID     = os.getenv("MODEL_ID", "ibm/granite-13b-instruct-v2") # Default model

# Vector DB Config
DOCS_FOLDER = Path(os.getenv("DOCS_FOLDER", "documents")) # Where PDFs are stored
CHROMA_PERSIST_DIR = Path(os.getenv("CHROMA_PERSIST_DIR", "./chroma_db_data")) # Where DB data is saved
COLLECTION_NAME = "pdf_documents_chunked_watsonx" # DB Collection name
NUM_RESULTS_RAG = 3   # How many chunks to retrieve for context
CHUNK_SIZE = 700      # Target size for text chunks (in characters)
CHUNK_OVERLAP = 150   # Overlap between consecutive chunks

# --- Global Variables (Using typing.Union for compatibility) ---
# Holds the initialized ChromaDB collection
pdf_collection: Union[chromadb.Collection, None] = None # <--- FIX: Use Union
# Holds the persistent ChromaDB client instance
chroma_client: Union[chromadb.PersistentClient, None] = None # <--- FIX: Use Union

# --- Logging Configuration ---
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)

# --- Environment Variable Validation ---
for name, val in [
    ("WATSONX_APIKEY", API_KEY),
    ("WATSONX_URL", URL),
    ("PROJECT_ID", PROJECT_ID)
]:
    if not val:
        raise RuntimeError(f"{name} is not set. Please add it to your .env file.")

説明：

• インポート： 標準ライブラリ（os、sys、logging、pathlib、re、textwrap）、.env ファイルから機密性の高い認証情報を読み込むための dotenv、型ヒント（特に古い Python バージョンとの互換性を保つための Union）に使用する typing、FastMCP サーバークラス、Watsonx.ai SDK のコンポーネント（APIClient、Credentials、ModelInference）、ベクトルデータベース用の chromadb、PDF ファイルの読み取りに使用する pypdf をインポートします。
• 設定（load_dotenv、定数）： load_dotenv() を使用して、環境変数（API キーなど）を読み込みます。Watsonx.ai の詳細、ドキュメントとデータベースの永続化ディレクトリへのパス、データベースコレクション名、RAG プロセスを制御するパラメーター（結果の数、チャンクサイズ、オーバーラップ）の定数を定義します。定数を使用すると、コードが読みやすくなり、変更も容易になります。
• グローバル変数（pdf_collection、chroma_client）： これらは起動時に一度だけ初期化し、後で MCP ツール関数（chat_with_manual）からアクセスできるよう、グローバルに宣言します。Union[..., None] は、初期化前には特定の型または None が設定される可能性があることを示します。
• ロギングと検証： サーバーの動作を確認するため、基本的なロギングを設定します。特に重要なのは、必要な Watsonx.ai 認証情報が環境変数に設定されているか確認することです。設定されていない場合、スクリプトはエラーを出して停止します。

2. 知識の抽出：PDF の処理（pdf_to_text）

ドキュメントを検索する前に、まずテキストを抽出する必要があります。

def pdf_to_text(folder: Path) -> dict:
    """Extract text from every PDF in `folder`, preserving structure."""
    texts = {}
    if not folder.is_dir():
        logging.error("Folder not found: %s", folder)
        return texts
    logging.info("Scanning folder '%s' for PDFs...", folder)
    pdf_files = list(folder.glob("*.pdf"))
    if not pdf_files:
        logging.warning("No PDF files found in folder '%s'.", folder)
        return texts
    for pdf_path in pdf_files:
        try:
            logging.info("Attempting to read %s", pdf_path.name)
            reader = PdfReader(str(pdf_path))
            # Handle encrypted PDFs (basic attempt)
            if reader.is_encrypted:
                logging.warning("PDF is encrypted: %s. Attempting default decryption.", pdf_path.name)
                try: reader.decrypt(''); logging.info("Successfully decrypted %s.", pdf_path.name)
                except Exception as de: logging.error("Failed decrypt %s: %s.", pdf_path.name, de); continue
            # Extract text page by page and join
            content = "\n".join(page.extract_text() or "" for page in reader.pages)
            # Basic cleaning: reduce excessive newlines
            content = re.sub(r'\n{3,}', '\n\n', content.strip())
            if content:
                texts[pdf_path.name] = content
                logging.info("Extracted %d chars from %s", len(content), pdf_path.name)
            else:
                logging.warning("Extracted no text from %s.", pdf_path.name)
        except Exception as e:
            logging.error("Error reading %s: %s", pdf_path.name, e)
    if not texts:
        logging.warning("No text extracted from any PDF.")
    return texts

説明：

• pdf_to_text 関数は、入力として DOCS_FOLDER のパスを受け取ります。
• フォルダー内から、.pdf で終わるファイルをすべて検索します。
• 各 PDF を開くために、pypdf.PdfReader を使用します。
• 暗号化の基本的な確認を行い、空のパスワードでの復号を試みます（デフォルトの保護設定でよく使用されます）。
• 各ページを順に処理し、page.extract_text() を使ってテキストを抽出した後、すべてのページのテキストを改行で結合します。
• シンプルな正規表現（re.sub）で、過剰な空行を取り除きます。
• 各 PDF から抽出したテキストは、ファイル名とテキスト内容を対応付けた辞書に格納されます。
• ファイルの読み取りに関する問題を処理するエラー処理も含まれています。

3. テキストの分割：チャンク化（chunk_text）

LLM が一度に処理できるテキスト量には制限があります（コンテキストウィンドウ）。また、小さく焦点を絞ったテキストのまとまりにすると、特定の情報を検索しやすくなります。そこでチャンク化を行います。

def chunk_text(text: str, chunk_size: int, chunk_overlap: int) -> list[str]:
    """Recursively splits text into chunks of a target size with overlap."""
    if not text: return []
    # Prioritized list of separators to split by
    separators = ["\n\n", "\n", ". ", " ", ""] # From paragraphs down to characters

    def split(text_to_split, current_separators):
        # Base case: If text is small enough or no more separators to try
        if not text_to_split or len(text_to_split) <= chunk_size:
             return [text_to_split] if text_to_split and text_to_split.strip() else []

        current_separator = current_separators[0]
        next_separators = current_separators[1:]

        # Handle splitting by character if all else fails
        if current_separator == "":
            # Simple character-based splitting with overlap
            chunks = []
            start = 0
            while start < len(text_to_split):
                end = start + chunk_size
                chunk = text_to_split[start:end]
                if chunk.strip():
                    chunks.append(chunk)
                # Move start, considering overlap
                start += chunk_size - chunk_overlap
                if start >= len(text_to_split): break # Prevent infinite loop
            return chunks

        # Try splitting by the current separator
        splits = [s for s in text_to_split.split(current_separator) if s.strip()]
        if not splits: # If this separator didn't work, try the next one
             return split(text_to_split, next_separators) if next_separators else split(text_to_split, [""])


        final_chunks = []
        current_chunk_parts = []
        current_length = 0
        separator_len = len(current_separator)

        # Try to combine smaller splits into chunks respecting chunk_size
        for i, part in enumerate(splits):
            part_len = len(part)
            potential_length = current_length + part_len + (separator_len if current_chunk_parts else 0)

            if potential_length <= chunk_size:
                # Add to current chunk
                current_chunk_parts.append(part)
                current_length = potential_length
            else:
                # Current chunk is full, or this part is too large
                if current_chunk_parts:
                    # Finalize the current chunk
                    chunk = current_separator.join(current_chunk_parts)
                    if chunk.strip(): final_chunks.append(chunk)

                    # --- Overlap Logic (Simplified Approach) ---
                    # For simplicity, we restart the next chunk with the current part.
                    # More complex overlap (like carrying over the end of the last chunk)
                    # can be implemented but adds complexity. A simple restart ensures
                    # the current part isn't lost.
                    current_chunk_parts = [part]
                    current_length = part_len

                    # If the current part *itself* is too big, split it further
                    if part_len > chunk_size:
                        final_chunks.extend(split(part, next_separators if next_separators else [""]))
                        current_chunk_parts = [] # Reset as it was handled recursively
                        current_length = 0

                else: # current_chunk_parts is empty: this 'part' alone is > chunk_size
                    final_chunks.extend(split(part, next_separators if next_separators else [""]))
                    current_chunk_parts = [] # Ensure reset
                    current_length = 0

            # Add the last assembled chunk if any parts remain
            if i == len(splits) - 1 and current_chunk_parts:
                chunk = current_separator.join(current_chunk_parts)
                if chunk.strip(): final_chunks.append(chunk)

        return [c for c in final_chunks if c.strip()] # Final cleanup

    # Start the splitting process with the initial text and separators
    return split(text, separators)

説明：

• chunk_text 関数は、再帰的な方法でテキストを分割します。
• まず意味のある区切り（段落 \n\n、行 \n、文 . ）で分割し、それができない場合にスペースや個々の文字で分割します。
• 目標の CHUNK_SIZE に近いチャンクを作成しようとします。
• CHUNK_OVERLAP によって、チャンク間で文脈が唐突に途切れないようにし、チャンクの境界をまたぐ関連情報を検索プロセスで見つけやすくします。文がチャンク A の末尾で始まり、チャンク B の冒頭で終わる場合、オーバーラップによってその文を捉えやすくなります。
• 再帰処理では、単一の「部分」（たとえば、改行を含まない非常に長い段落）が CHUNK_SIZE を超える場合に、それをさらに細かく分割します。

4. ナレッジベースの構築：ベクトル DB の作成（build_new_vector_db）

この関数は、抽出してチャンクに分割したテキストを ChromaDB ベクトルデータベースにインデックス登録するプロセスを管理します。データベースが存在しないか、空の場合にのみ実行されます。

def build_new_vector_db(client: chromadb.PersistentClient, docs_folder: Path) -> bool:
    """Extracts, chunks, and indexes PDFs into the specified collection."""
    global pdf_collection # To assign the newly created/updated collection

    logging.info("--- Starting Vector DB Creation/Update Process ---")
    try:
        # 1. Extract text from PDFs
        extracted = pdf_to_text(docs_folder)
        if not extracted:
            logging.error("No text extracted, cannot build vector DB.")
            return False

        # 2. Get or Create the ChromaDB Collection
        logging.info("Accessing collection '%s'...", COLLECTION_NAME)
        # get_or_create is safe: it gets if exists, creates if not
        collection = client.get_or_create_collection(name=COLLECTION_NAME)
        logging.info("Using collection '%s'. Current item count: %d", COLLECTION_NAME, collection.count())

        # 3. Chunk Text and Prepare for Indexing
        docs, metas, ids = [], [], []
        logging.info("Chunking text with target size ~%d chars, overlap %d chars...", CHUNK_SIZE, CHUNK_OVERLAP)
        total_chunks = 0
        for fname, text in extracted.items():
            if not text.strip(): continue # Skip empty documents
            file_chunks = chunk_text(text, CHUNK_SIZE, CHUNK_OVERLAP)
            logging.info("Generated %d chunks for %s.", len(file_chunks), fname)
            if not file_chunks: continue
            for idx, chunk in enumerate(file_chunks, start=1):
                docs.append(chunk) # The text chunk itself
                metas.append({"source": fname, "chunk_idx": idx}) # Metadata: filename and chunk number
                # Create a unique ID for each chunk
                safe_fname = re.sub(r'[^\w-]', '_', fname) # Make filename safe for ID
                unique_id = f"{safe_fname}_chunk_{idx}"
                ids.append(unique_id)
            total_chunks += len(file_chunks)

        if not docs:
            logging.warning("No text chunks generated for indexing.")
            pdf_collection = collection # Assign the (empty) collection handle
            return True # Succeeded, but nothing to add

        # 4. Add data to ChromaDB in Batches
        logging.info("Prepared %d total chunks for indexing.", total_chunks)
        batch_size = 100 # Process 100 chunks at a time
        num_batches = (len(docs) + batch_size - 1) // batch_size
        for i in range(0, len(docs), batch_size):
            batch_docs = docs[i:i+batch_size]
            batch_metas = metas[i:i+batch_size]
            batch_ids = ids[i:i+batch_size]
            try:
                # Use add() - Chroma handles embedding generation automatically
                collection.add(documents=batch_docs, metadatas=batch_metas, ids=batch_ids)
                logging.info("Indexed batch %d/%d (size %d chunks).",
                             (i // batch_size) + 1, num_batches, len(batch_docs))
            except Exception as e:
                logging.error("Error adding batch starting at index %d: %s", i, e)
                # Decide whether to continue or stop on batch error

        # 5. Finalize and assign global variable
        final_count = collection.count()
        logging.info("Finished indexing. Collection '%s' now contains %d items.",
                     collection.name, final_count)
        pdf_collection = collection # Make the populated collection globally available
        return True

    except Exception as e:
        logging.error("Failed during Vector DB build process: %s", e, exc_info=True)
        return False

説明：

• ワークフロー： pdf_to_text を呼び出した後、抽出したテキストをループ処理し、各ドキュメントに対して chunk_text を呼び出します。
• データの準備： 3 つのリストを準備します。docs（実際のテキストチャンク）、metas（各チャンクに関連付けられたメタデータ。ソースファイル名など）、ids（各チャンクの一意の識別子）です。一意の ID は、後で特定のチャンクを更新または削除する場合に重要です。
• ChromaDB との連携： client.get_or_create_collection を使用してコレクションへのハンドルを安全に取得します。初回の場合はコレクションを作成します。
• バッチインデックス登録： collection.add() を使用して、ドキュメント、メタデータ、ID を ChromaDB コレクションに追加します。効率を高め、一度に数千件に及ぶ可能性のあるチャンクによってデータベース接続やメモリに過度な負荷がかからないよう、バッチ（batch_size = 100）で処理します。ChromaDB は、デフォルトの埋め込みモデルを使用してテキストチャンクをベクトル埋め込みに変換する処理を自動的に行います。
• グローバル変数への代入： データの格納が完了すると、collection オブジェクトがグローバル変数 pdf_collection に代入されます。

5. スマートな起動：ベクトル DB の初期化（initialize_vector_db）

この関数は永続化に不可欠です。特にデータベースが大規模な場合、サーバーの起動ごとにベクトルデータベースを再構築する必要はありません。この関数は、指定された永続化ディレクトリにデータベースがすでに存在するかを確認し、存在する場合は読み込みます。存在しない場合は、構築プロセスを開始します。

def initialize_vector_db(persist_path: Path, docs_folder_path: Path):
    """Initializes a persistent ChromaDB client and ensures the collection exists."""
    global pdf_collection, chroma_client
    logging.info("--- Initializing Persistent Vector DB ---")
    logging.info("Storage directory: %s", persist_path)
    # Ensure the directory exists
    persist_path.mkdir(parents=True, exist_ok=True)

    try:
        # 1. Initialize Persistent Client
        # This tells ChromaDB to save data to the specified path
        chroma_client = chromadb.PersistentClient(path=str(persist_path))
        logging.info("ChromaDB Persistent Client initialized.")
    except Exception as e:
        logging.error("Fatal: Failed to initialize ChromaDB Persistent Client: %s", e, exc_info=True)
        return False # Cannot continue without a client

    try:
        # 2. Check if the Collection Already Exists
        logging.info("Checking for existing collection: '%s'", COLLECTION_NAME)
        existing_collections = chroma_client.list_collections()
        collection_exists = any(col.name == COLLECTION_NAME for col in existing_collections)

        if collection_exists:
            # 3a. Load Existing Collection
            logging.info("Collection '%s' found. Getting handle.", COLLECTION_NAME)
            collection = chroma_client.get_collection(name=COLLECTION_NAME)
            count = collection.count()
            if count > 0:
                # Collection exists and has data - load it
                logging.info("Existing collection has %d items. Using existing.", count)
                pdf_collection = collection # Assign to global variable
                return True # Success!
            else:
                # Collection exists but is empty - build it
                logging.warning("Found existing collection '%s', but it is empty. Will attempt to build.", COLLECTION_NAME)
                return build_new_vector_db(chroma_client, docs_folder_path)
        else:
            # 3b. Build New Collection
            logging.info("Collection '%s' not found. Proceeding to build a new one.", COLLECTION_NAME)
            return build_new_vector_db(chroma_client, docs_folder_path)

    except Exception as e:
        logging.error("Error during Vector DB initialization/check: %s", e, exc_info=True)
        return False # Failed to initialize or build

説明：

• 永続化： chromadb.PersistentClient(path=...) が重要です。ChromaDB に対し、指定されたディレクトリ（CHROMA_PERSIST_DIR）にデータを保存するよう指示します。
• 構築前の確認： chroma_client.list_collections() を使用して、永続ストレージに COLLECTION_NAME というコレクションがすでに存在するかを確認します。
• 読み込みまたは構築のロジック：
  • コレクションが存在し、かつアイテムが含まれている（count > 0）場合、chroma_client.get_collection() で読み込み、グローバル変数 pdf_collection に代入します。
  • コレクションが存在するものの空の場合、またはコレクションが存在しない場合は、build_new_vector_db を呼び出して作成し、データを格納します。
• グローバル変数への代入： 初期化された chroma_client もグローバル変数に保存されます。

6. 頭脳との接続：Watsonx.ai の初期化

ここでは、SDK と認証情報を使用して IBM Watsonx.ai サービスへの接続を設定します。

# --- Watsonx.ai Initialization ---
try:
    creds  = Credentials(url=URL, api_key=API_KEY)
    client = APIClient(credentials=creds, project_id=PROJECT_ID)
    model = ModelInference(
        model_id=MODEL_ID,
        credentials=creds,
        project_id=PROJECT_ID
    )
    logging.info(
        f"Initialized Watsonx.ai model '{MODEL_ID}' for project '{PROJECT_ID}'."
    )
except Exception as e:
    logging.error("Failed to initialize Watsonx.ai client/model: %s", e, exc_info=True)
    sys.exit(1) # Exit if we can't connect to the LLM service

説明：

• .env ファイルから読み込んだ API キーと URL を使用して Credentials を作成します。
• 一般的なやり取り用に APIClient をインスタンス化します（ただし、ここでは生成に直接使用しません）。
• 使用する MODEL_ID（例：「ibm/granite-13b-instruct-v2」）、認証情報、PROJECT_ID を指定して、ModelInference オブジェクトを作成します。この model オブジェクトを回答の生成に使用します。
• エラー処理により、Watsonx.ai への接続を初期化できない場合はサーバーを終了します。

7. 接続の準備：MCP サーバーのセットアップ

クライアント（コマンドラインインターフェイスや Web アプリなど）がバックエンドのロジックと通信する方法が必要です。FastMCP を使うと、Python 関数を簡単に公開できます。

# --- MCP Server Setup ---
mcp = FastMCP("Watsonx RAG Chatbot Server")

説明：

• FastMCP のインスタンスを作成します。文字列引数はサーバーの名前にすぎず、多くの場合、検出やログ記録に使用されます。デフォルトでは、FastMCP は標準入出力（STDIO）を使用して通信します。これは、サーバーをサブプロセスとして起動するシンプルなクライアント・サーバー構成に適しています。

8. コアロジック：RAG ツール（chat_with_manual）

これは RAG プロセスの中核で、MCP 経由で呼び出せるツールとして公開されています。

# --- RAG Tool Definition ---
@mcp.tool()
def chat_with_manual(query: str) -> str:
    """
    Answers questions about drone manuals using RAG with Watsonx.ai.
    Relies on the globally initialized pdf_collection.
    """
    global pdf_collection # Access the globally initialized DB collection
    logging.info("Received RAG query: %r", query)

    # Ensure the vector DB is ready
    if pdf_collection is None:
        logging.error("Vector DB collection is not available. Initialization might have failed.")
        return "Error: The document database is not ready. Please check server logs."

    # 1. Retrieve: Query the Vector DB
    try:
        logging.info("Querying Vector DB ('%s') for top %d results...", pdf_collection.name, NUM_RESULTS_RAG)
        # Find chunks most similar to the user's query
        results = pdf_collection.query(
            query_texts=[query],          # The user's question
            n_results=NUM_RESULTS_RAG,    # How many results to fetch (configured earlier)
            include=['documents']         # We need the actual text content ('documents')
        )
    except Exception as e:
        logging.error("Error querying ChromaDB: %s", e, exc_info=True)
        return f"Error: Could not retrieve information from the document database."

    # 2. Augment: Prepare the Context
    retrieved_docs = results.get('documents', [[]])[0] # Extract the list of document texts
    logging.info(f"Retrieved documents: {retrieved_docs}") # Log the actual content

    if not retrieved_docs:
        # Handle case where no relevant documents were found
        logging.warning("No relevant documents found in Vector DB for query: %r", query)
        # Return a message indicating nothing was found (alternative: let the LLM try without context)
        return "I couldn't find specific information about that in the available documents."
    else:
        # Combine the retrieved chunks into a single context string
        logging.info("Retrieved %d document chunks.", len(retrieved_docs))
        context_string = "\n\n---\n\n".join(retrieved_docs) # Separate chunks clearly

    # 3. Generate: Construct Prompt and Call LLM
    # Create a prompt that instructs the LLM to use *only* the provided context
    prompt_template = f"""
    You are a helpful assistant that answers questions based *only* on the information provided from the manual below.

    --- Manual Context ---
    {context_string}
    --- End of Context ---

    Using only the context above, please answer the following question clearly and accurately.

    Question: {query}

    Answer:
    """
    logging.info("Constructed prompt for Watsonx.ai (length: %d chars)", len(prompt_template))
    # Use DEBUG level for logging full prompts to avoid cluttering INFO logs
    logging.debug("Prompt:\n%s", prompt_template)
    # Define LLM generation parameters
    params = {
        GenParams.DECODING_METHOD: "greedy", # Simple, deterministic output
        GenParams.MAX_NEW_TOKENS:  300,     # Max length of the generated answer
        GenParams.MIN_NEW_TOKENS:  10,      # Min length of the generated answer
        # GenParams.TEMPERATURE: 0.7, # Uncomment for more varied/creative answers
        GenParams.STOP_SEQUENCES: ["\n\n", "---", "Question:", "Context:"] # Stop generation if these appear
    }
    logging.info("Sending request to Watsonx.ai model '%s'...", MODEL_ID)

    # Call the Watsonx.ai model
    try:
        resp = model.generate_text(prompt=prompt_template, params=params, raw_response=True)
        # Carefully parse the response structure
        if resp and isinstance(resp, dict) and "results" in resp and isinstance(resp["results"], list) and len(resp["results"]) > 0:
            first_result = resp["results"][0]
            if isinstance(first_result, dict) and "generated_text" in first_result:
                answer = first_result["generated_text"].strip()
                logging.info("Received Watsonx.ai response: %r", answer)
                # Simple post-processing to remove trailing incomplete sentences
                if answer.endswith("..."):
                    last_period = answer.rfind('.')
                    if last_period != -1:
                        answer = answer[:last_period+1]
                return answer
            else:
                logging.error("Watsonx.ai response structure unexpected ('generated_text' missing): %s", first_result)
                return "Error: Received an unexpected response format from the AI model (Detail 1)."
        else:
            logging.error("Watsonx.ai response structure unexpected (main structure): %s", resp)
            return "Error: Received an unexpected response format from the AI model (Detail 2)."
    except Exception as e:
        logging.error("Watsonx.ai inference error: %s", e, exc_info=True)
        return f"Error: Failed to generate an answer due to an AI model issue."

説明：

• @mcp.tool()： このデコレーターは chat_with_manual 関数を MCP サーバーに登録し、クライアントから呼び出せるようにします。
• グローバルアクセス： 先ほど初期化した pdf_collection にアクセスします。
• 取得： pdf_collection.query() を使用して、ユーザーの query のベクトル埋め込みに最も近い（最も類似した）ベクトル埋め込みを持つ NUM_RESULTS_RAG 個のドキュメントチャンクを検索します。結果に documents（テキスト）を含めるよう明示的に指定します。
• 拡張： ドキュメントが見つかったかどうかを確認します。見つかった場合は、わかりやすいように \n\n---\n\n で区切ってテキスト内容を単一の context_string に結合します。見つからなかった場合は、特定のメッセージを返します。
• 生成：
  • プロンプトエンジニアリング： 注意深く作成されたプロンプト（prompt_template）を用意します。これは RAG にとって 非常に重要 です。LLM（Watsonx.ai モデル）の役割を明示し、提供された context_string のみに基づいてユーザーの query に回答するよう指示します。これにより、LLM が一般知識を使用するのを防ぎ、ドキュメントの内容に沿った回答を強制します。
  • パラメーター： max_new_tokens（回答の長さ）、decoding_method（LLM が次の単語を選ぶ方法。「greedy」は決定論的です）、stop_sequences（LLM に生成を停止させるシーケンス）などの生成パラメーターを定義します。
  • LLM 呼び出し： model.generate_text() がプロンプトとパラメーターを Watsonx.ai に送信します。
  • 応答処理： コードは Watsonx.ai からの JSON 応答を解析し、generated_text を抽出します。基本的な後処理を行い、途中で終わっている可能性のある回答を整えます。ChromaDB へのクエリと Watsonx.ai API 呼び出しの両方にエラー処理を含めています。

9. 実演：メイン実行ブロック

最後に、if __name__ == "__main__": ブロックにより、以下のコードはスクリプトを直接実行したときにのみ（モジュールとしてインポートしたときではなく）実行されます。

# --- Main Execution ---
if __name__ == "__main__":
    # 1. Initialize/Load the Persistent Vector DB *before* starting the server
    # This ensures the RAG tool has data to query
    if not initialize_vector_db(CHROMA_PERSIST_DIR, DOCS_FOLDER):
         logging.error("CRITICAL: Failed to initialize Vector DB. Exiting.")
         sys.exit(1) # Stop server if DB isn't ready

    # 2. Start the MCP server (this call is blocking)
    # It will listen for client connections (on STDIO by default)
    # and dispatch calls to registered tools like chat_with_manual
    logging.info("Starting MCP server on STDIO transport...")
    mcp.run()

    # This line is reached only when the server stops (e.g., Ctrl+C)
    logging.info("MCP server stopped.")

説明：

• まず初期化： 最初に initialize_vector_db を呼び出します。これは不可欠です。ベクトルデータベースの準備が整ったこと（読み込み済み、または新たに構築済み）が確認されるまで、サーバーはリクエストの受付を開始してはいけません。初期化に失敗した場合、スクリプトは終了します。
• サーバーを起動： データベースの準備が整っていれば、mcp.run() が MCP サーバーを起動します。この関数は通常、処理をブロックします。つまり、スクリプトはここで待機し、クライアントからのリクエストを受信して、適切な @mcp.tool 関数（この場合は chat_with_manual）にルーティングします。
• シャットダウン： サーバーがシャットダウンされるまで（たとえば、クライアントが正常に切断するか、Ctrl+C などの割り込み信号が送られるまで）、スクリプトは mcp.run() の先に進みません。

このモジュール構造により、生の PDF から LLM が生成するコンテキストを踏まえた回答まで、データがどのように流れるかが比較的わかりやすくなります。次のステップでは、この MCP サーバーに接続し、chat_with_manual ツールを操作するクライアントアプリケーションを作成します。

ステップ 4. RAG ボットとの対話：client.py スクリプト

前のセクションでは、RAG チャットボットの強力なバックエンドである server.py を作成しました。しかし、実際にはどのように 使用 するのでしょうか。そこで登場するのが client.py です。このスクリプトはユーザーインターフェイス（現時点ではシンプルなコマンドラインインターフェイス）として機能し、サーバーを起動して質問を送信し、RAG システムから受け取った回答を表示します。

このスクリプトでは Meta Compute Protocol（MCP）クライアントライブラリを使用します。標準入出力（STDIO）経由で通信するように設定されており、server.py で構築した FastMCP サーバーとのやり取りに最適です。

以下が client.py のコード全体です。

# client.py
import asyncio
import logging
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Configure basic logging for the client
logging.basicConfig(level=logging.INFO, format='%(asctime)s [Client] %(message)s')

async def main():
    # Command to start the server.py script using the python interpreter
    server_params = StdioServerParameters(command="python", args=["server.py"])
    logging.info("Attempting to start and connect to MCP server ('server.py')...")

    try:
        # 1. Start the server and establish connection via STDIO
        async with stdio_client(server_params) as (reader, writer):
            logging.info("Connected to MCP server. Initializing session...")
            # 2. Establish MCP Session using the reader/writer streams
            async with ClientSession(reader, writer) as session:
                await session.initialize() # Perform MCP handshake
                logging.info("MCP session initialized.")

                # --- Interaction Logic ---
                # Define the question to ask the RAG system
                user_msg = "How many flight modes it has and explain them?"
                # Example alternative question:
                # user_msg = "What is the maximum flight time?"

                logging.info("Calling tool 'chat_with_manual' with query: %r", user_msg)

                try:
                    # 3. Call the remote tool on the server
                    response = await session.call_tool(
                        "chat_with_manual",          # Name must match the @mcp.tool function in server.py
                        arguments={"query": user_msg} # Arguments expected by the tool
                    )
                    logging.info("Received response from server.")

                    # 4. Process and Display the Response
                    # MCP might wrap responses; attempt to extract raw text if needed
                    final_answer = response # Default to the raw response
                    if hasattr(response, 'content') and response.content and isinstance(response.content, list):
                        # Check if response has structured content
                        first_content = response.content[0]
                        if hasattr(first_content, 'text'):
                            final_answer = first_content.text # Extract text if available

                    # Print clearly formatted question and answer
                    print("\n" + "="*20 + " Query " + "="*20)
                    print(f"Your Question: {user_msg}")
                    print("\n" + "="*20 + " Answer " + "="*20)
                    print(f"Bot Answer:\n{final_answer}")
                    print("\n" + "="*50)

                except Exception as e:
                    # Handle errors during the tool call specifically
                    logging.error("Error calling tool 'chat_with_manual': %s", e, exc_info=True)
                    print(f"\nError interacting with the chatbot: {e}")

    except Exception as e:
        # Handle errors during connection or session setup
        logging.error("Failed to connect or communicate with the MCP server: %s", e, exc_info=True)
        print(f"\nCould not connect to or run the server: {e}")

if __name__ == "__main__":
    # Run the asynchronous main function
    asyncio.run(main())

client.py の解説

このクライアントの動作を順に見ていきましょう。

1. インポートとログ記録：

   • asyncio：接続、送信、受信など、MCP クライアントの操作は非同期で行われるため、不可欠です。
   • mcp：通信プロトコルを管理する ClientSession、サーバーの起動方法を定義する StdioServerParameters、サーバーの起動と通信ストリームの提供を処理するコンテキストマネージャー stdio_client をインポートします。
   • logging：クライアント側の動作を表示するため、基本的なログ記録を設定します。

2. main 非同期関数： クライアントの主要なロジックがここにあります。

   • サーバーの定義（StdioServerParameters）： サーバーの実行方法をクライアントに伝えます。StdioServerParameters(command="python", args=["server.py"]) は、「コマンド python server.py を実行する」という意味です。このサブプロセスはクライアントが管理します。
   • 接続（stdio_client）： async with stdio_client(server_params) as (reader, writer): ブロックが、主要な処理を行います。
     • バックグラウンドで python server.py プロセスを起動します。
     • サーバーの標準入力と標準出力を取得します。
     • クライアントがサーバーの標準入力にデータを送信し、標準出力からデータを非同期で読み取れるようにする reader と writer オブジェクトを提供します。
     • 重要な点として、async with ブロックを抜ける際（正常終了時またはエラー発生時）に、サーバープロセスが適切に終了することも保証します。
   • MCP セッション（ClientSession）： stdio_client ブロック内の async with ClientSession(reader, writer) as session: では、生の reader／writer ストリームを MCP セッションハンドラーでラップします。このオブジェクトは、MCP プロトコルに従ってメッセージを整形します。
   • 初期化（session.initialize()）： 呼び出しを行う前に、MCP プロトコルを使って通信できる状態かどうかを確認するため、クライアントとサーバーが簡単なハンドシェイクを行います。
   • クエリの定義（user_msg）： RAG ボットに尋ねる質問を、シンプルな文字列変数に格納します。
   • ツールの呼び出し（session.call_tool）： ここが重要なやり取りのポイントです。
     • await session.call_tool(...)：確立されたセッションを通じて、サーバーにリクエストを送信します。
     • "chat_with_manual"：この文字列は、server.py で @mcp.tool() がデコレートされた関数の名前と完全に一致する必要があります。
     • arguments={"query": user_msg}：この辞書には、サーバー関数に渡す引数が含まれます。キー "query" は、サーバー側の定義 chat_with_manual(query: str) のパラメーター名と一致する必要があります。
     • その後、クライアントはサーバーからの応答を await で待ちます。
   • 応答の処理： response 変数には、サーバー上の chat_with_manual 関数が返した値（回答を含む文字列のはずです）が格納されます。MCP が応答をより複雑なオブジェクトでラップする場合に備え、コードにはチェック（if hasattr(response, 'content')...）が含まれており、生のテキストを抽出しようとします。最後に、元の質問と受け取った回答を整形して出力します。
   • エラー処理： try...except ブロックを使って、サーバーの起動失敗、接続の問題、サーバー側でのツール呼び出し中のエラー（クライアントに返されるもの）など、起こり得る問題を捕捉します。

3. クライアントの実行（if __name__ == "__main__":）

   • この標準的な Python の構文により、スクリプトを直接実行した場合にのみ、内部のコードが実行されます。
   • asyncio.run(main())：main は async 関数なので、イベントループを実行して非同期のクライアントロジックを実行するために asyncio が必要です。

ループを完了する

この client.py スクリプトは、ユーザーと高度な server.py バックエンドの間をうまく橋渡しします。python client.py を実行すると、RAG の全プロセスが開始されます。クライアントがサーバーを起動し、サーバーがベクトルデータベースを初期化（読み込み、または構築）し、クライアントがクエリを送信すると、サーバーがデータベースから関連するコンテキストを取得して、コンテキストとクエリを Watsonx.ai に送信します。生成された回答を受け取ってクライアントに返し、最後にクライアントがユーザーに表示します。

ステップ 5：依存関係の設定

必要なライブラリがすべてインストールされるように、requirements.txt ファイルを作成します。

# requirements.txt
ibm-watsonx-ai>=1.0.0
chromadb>=0.4.0 # Use a recent version
pypdf>=3.0.0
python-dotenv>=1.0.0
mcp-sdk>=0.3.0
# tiktoken might be needed by chromadb's default embedding model
tiktoken>=0.5.0

依存関係をインストールします（仮想環境が有効になっていることを確認してください）。

pip install -r requirements.txt

ステップ 6：RAG サーバーとクライアントの実行

必要に応じてサーバーを起動できます。

python server.py

クライアントスクリプトは、サーバーを自動的に起動するように設計されています。クライアントを実行するだけで構いません。

ターミナルを開き（watsonx_rag_mcp_server ディレクトリにいて、仮想環境が有効になっていることを確認して）、次のコマンドを実行します。

python client.py

処理の流れ：

1. client.py が起動します。
2. サブプロセスとして python server.py を実行します。
3. server.py が起動し、Watsonx.ai への接続を初期化します。
4. server.py が ChromaDB データベースを確認します。
   • 初回： データベースが見つからないため、documents フォルダー内の PDF を処理してチャンクに分割し、埋め込みを ./chroma_db_data ディレクトリに保存します。PDF のサイズと数によっては、時間がかかることがあります。
   • 2 回目以降： ./chroma_db_data ディレクトリとその中のコレクションが見つかるため、すばやく読み込み、処理をスキップします。
5. server.py が MCP サーバーを起動し、STDIO で待ち受けます。
6. client.py が STDIO 経由でサーバーに接続し、MCP セッションを初期化します。
7. client.py が定義済みのクエリ（「How many flight modes...」）を chat_with_manual ツールに送信します。
8. server.py がリクエストを受け取り、ChromaDB にクエリを実行して関連するテキストチャンクを取得し、コンテキストを含むプロンプトを作成して Watsonx.ai を呼び出し、回答を受け取ってクライアントに返します。
9. client.py が回答を受け取り、ターミナルに出力します。

ステップ 7：RAG サーバーへのクエリ

client.py を実行すると、ハードコードされた次の質問を送信します。

Your Question: How many flight modes does the drone have and explain them?

Watsonx.ai が生成した回答が表示されます。回答は、指定した PDF ドキュメント内で見つかった情報のみに基づきます。回答の品質と完全性は、情報がドキュメントに含まれているかどうか、またベクトル検索によってどれだけ適切に取得されたかに大きく左右されます。

コアコンポーネントの説明

• IBM Watsonx.ai の統合： ドキュメントから提供されたコンテキストを理解し、クエリに対して人間らしい回答を生成する LLM を提供します。ibm-watsonx-ai SDK を使用して接続します。
• ベクトルインデックス作成のための ChromaDB： RAG システムのメモリとして機能します。PDF から取得したテキストチャンクを、埋め込みモデル（指定しない場合、ChromaDB はデフォルトのモデルを使用）を使って数値表現（ベクトル埋め込み）に変換し、保存します。クエリを受け取ると、その埋め込みに最も近い（意味的に最も類似した）埋め込みを持つチャンクを検索します。実行間でデータベースを保持するために、chromadb.PersistentClient を使用します。
• Model Context Protocol（MCP）： クライアントとサーバーが通信するための標準化された方法を提供します。独自の API エンドポイントを定義する代わりに、MCP の概念を使用します。
  • ツール： chat_with_manual 関数を MCP ツールとして公開します。クライアントはこのツールを呼び出して RAG 処理を実行します。これは、専用の関数呼び出しや、Web の用語で言えば POST リクエストに相当します。
  • (リソースとプロンプト)： この例ではあまり使用していませんが、MCP ではデータソースを リソース（GET リクエストのようなもの）として直接公開したり、再利用可能な プロンプト を定義したりすることもできます。

ヒントとトラブルシューティング

• ログを確認する： server.py と client.py はどちらも詳細なログを出力します。問題が発生した場合は、ターミナルの出力にある [INFO]、[WARNING]、または [ERROR] メッセージを確認してください。必要に応じて、ログレベル（例：logging.DEBUG）を上げると、さらに詳細な情報を確認できます。
• .env の設定： .env ファイルに記載した API キー、URL、Project ID が正しいこと、またファイルが読み込まれていること（ファイル名に誤字がないこと）を再確認してください。
• PDF の読み取り可能性： documents フォルダー内の PDF ファイルが読み取り可能で、パスワードで保護されていないことを確認してください（または、pdf_to_text による簡易的な復号の試行が機能することを確認してください）。PDF が破損していると、エラーが発生する場合があります。
• 依存関係： requirements.txt に記載されているすべてのライブラリが、アクティブな仮想環境に正しくインストールされていることを確認してください。
• 強制的に再構築する： PDF を更新し、サーバーに再度インデックスを作成させるには、client.py をもう一度実行する前に chroma_db_data ディレクトリーを削除する必要があります。サーバーはデータベースが存在しないことを検知し、build_new_vector_db プロセスを開始します。
• Watsonx.ai モデル： デフォルトでは ibm/granite-13b-instruct-v2 が使用されます。Llama 3 などの別のモデルでは、結果が異なったり、別のパラメーターが必要になったりする場合があります。.env ファイルの MODEL_ID を変更できます。

フロントエンド

server.py バックエンドとやり取りする、ChatGPT 風のチャットとして機能するシンプルな Flask ウェブインターフェースを作成しましょう。

このソリューションでは、次のことを行います。

1. ウェブフレームワークとして Flask を使用します。
2. Watsonx の壁紙を使用する、提供済みの base.html を使います。
3. chat.html を使ってチャットインターフェースを再現します。
4. client.py のロジックを応用して、ユーザーメッセージごとに server.py を起動し、通信します。注： サーバー／DB 接続を繰り返し初期化するため、本番環境での効率はよくありませんが、複雑な非同期処理を管理しない標準的な Flask の構成では、実装がより簡単です。
5. Flask セッションで会話履歴を保持します。
6. トークン数の代わりの指標として文字数に基づいて会話履歴を切り詰め、扱いやすいサイズに保ちます。

プロジェクト構成：

ファイルが次のように整理されていることを確認してください。

/chatbot
├── static/
│   └── assets/
│       └── watsonx-wallpaper.jpg  <-- Place your wallpaper image here
├── templates/
│   ├── base.html
│   └── chat.html
├── frontend.py            <-- The Flask app we will create
├── server.py              <-- Your existing server script
├── documents/             <-- Folder needed by server.py
│   └── ... (your PDFs)
├── .env                   <-- File needed by server.py
└── chroma_db_data/        <-- Directory created/used by server.py

1. ベーステンプレート（templates/base.html）

これは、背景と Tailwind CSS を設定する、提供済みのテンプレートです。

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>{% block title %}Watsonx RAG Chat{% endblock %}</title>
  <link href="https://cdn.jsdelivr.net/npm/[email protected]/dist/tailwind.min.css" rel="stylesheet">
  <style>
    html, body {
      height: 100%;
      margin: 0;
    }
    body {
      background-image: url('{{ url_for("static", filename="assets/watsonx-wallpaper.jpg") }}');
      background-size: cover;
      background-position: center center;
      background-repeat: no-repeat;
      background-attachment: fixed; /* Keep wallpaper fixed during scroll */
      display: flex;
      flex-direction: column; /* Stack content vertically */
    }
    /* Ensure content block takes available space */
    .content-wrapper {
       flex-grow: 1;
       display: flex;
       align-items: center; /* Center content vertically */
       justify-content: center; /* Center content horizontally */
       width: 100%;
       padding: 1rem; /* Add some padding */
    }
  </style>
</head>
<body class="bg-gray-900">  {# Default background if image fails #}
  <div class="content-wrapper">
    {% block content %}{% endblock %}
  </div>
  {# Optional: Add scripts here if needed globally #}
  <script>
      // Auto-scroll to bottom of chat window
      function scrollToBottom() {
          const chatbox = document.getElementById('chatbox');
          if (chatbox) {
              chatbox.scrollTop = chatbox.scrollHeight;
          }
      }
      // Scroll on initial load
      window.onload = scrollToBottom;
      // If using HTMX or similar for partial updates, you might need
      // to call scrollToBottom() after content is swapped.
  </script>
</body>
</html>

(注：Tailwind CDN リンクを更新し、固定された背景でコンテンツの中央揃えやスクロールを適切に処理できるように、background-attachment: fixed と Flexbox レイアウトを追加しました。)

2. チャットインターフェースのテンプレート（templates/chat.html）

このテンプレートは base.html を継承し、チャット UI を作成します。

{% extends 'base.html' %}

{% block title %}Watsonx RAG Chat{% endblock %}

{% block content %}
<div class="bg-white bg-opacity-80 backdrop-filter backdrop-blur-sm rounded-lg shadow-xl w-full max-w-2xl h-full max-h-[85vh] flex flex-col">

  <div class="p-4 border-b border-gray-300 flex justify-between items-center">
    <h1 class="text-xl font-semibold text-gray-800">Chat with PDF Manuals</h1>
    <a href="{{ url_for('clear_chat') }}" class="text-sm text-blue-600 hover:text-blue-800">Clear Chat</a>
  </div>

  <div id="chatbox" class="flex-grow p-4 space-y-4 overflow-y-auto">
    {% if not history %}
      <div class="text-center text-gray-500">Ask a question about the PDF documents!</div>
    {% endif %}
    {% for message in history %}
      {% if message.role == 'user' %}
        <div class="flex justify-end">
          <div class="bg-blue-500 text-white p-3 rounded-lg max-w-xs lg:max-w-md shadow">
            {{ message.content }}
          </div>
        </div>
      {% elif message.role == 'assistant' %}
        <div class="flex justify-start">
          <div class="bg-gray-200 text-gray-800 p-3 rounded-lg max-w-xs lg:max-w-md shadow">
            {# Use safe filter or pre tags if response might contain HTML/code #}
            <pre class="whitespace-pre-wrap font-sans">{{ message.content }}</pre>
          </div>
        </div>
      {% elif message.role == 'error' %}
        <div class="flex justify-start">
          <div class="bg-red-100 text-red-700 p-3 rounded-lg max-w-xs lg:max-w-md shadow">
            <strong>Error:</strong> {{ message.content }}
          </div>
        </div>
      {% endif %}
    {% endfor %}
  </div>

  <div class="p-4 border-t border-gray-300">
    <form method="POST" action="{{ url_for('chat') }}" class="flex space-x-2">
      <input type="text" name="message" placeholder="Type your question..." autocomplete="off" required
             class="flex-grow px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-blue-500">
      <button type="submit"
              class="bg-blue-600 text-white px-4 py-2 rounded-md hover:bg-blue-700 focus:outline-none focus:ring-2 focus:ring-blue-500 focus:ring-offset-2">
        Send
      </button>
    </form>
  </div>

</div>
{% endblock %}

(注：アシスタントのメッセージで改行などの書式を保持するために pre タグを追加し、自動スクロール用の id="chatbox" と基本的な JavaScript を base.html に追加しました。また、チャットをクリアするリンクと基本的なエラー表示も追加しました。)

3. Flask アプリケーション（frontend.py）

これは、ウェブフロントエンドのメイン Python スクリプトです。

import os
import sys
import asyncio
import logging
import uuid # To generate unique session IDs if needed, but Flask session handles this
from flask import Flask, render_template, request, redirect, url_for, session

# --- MCP Client Imports ---
# Assuming mcp library is installed: pip install meta-compute-protocol
# Make sure server.py is in the same directory or Python path
try:
    from mcp import ClientSession, StdioServerParameters
    from mcp.client.stdio import stdio_client
    # from mcp.common.content import TextContent # Might be needed depending on exact response wrapping
except ImportError:
    print("ERROR: 'meta-compute-protocol' library not found.")
    print("Please install it: pip install meta-compute-protocol")
    sys.exit(1)

# --- Configuration ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s [Frontend] %(levelname)s: %(message)s')

# --- Flask App Setup ---
app = Flask(__name__)
# IMPORTANT: Change this to a random secret key for production!
app.secret_key = os.environ.get("FLASK_SECRET_KEY", "dev-secret-key-replace-me")

# --- Conversation Memory Configuration ---
MAX_HISTORY_CHARS = 4000 # Approximate token limit (adjust as needed) ~1000 tokens

# --- Helper Function for MCP Interaction ---

async def async_mcp_call(user_query: str):
    """
    Starts server.py, connects via MCP, calls the tool, and returns the response.
    This runs the full client logic for each call - inefficient but simpler for Flask.
    """
    server_params = StdioServerParameters(command=sys.executable, args=["server.py"]) # Use sys.executable for portability
    logging.info("Attempting to start and connect to MCP server ('server.py')...")
    response_text = None
    error_message = None

    try:
        async with stdio_client(server_params) as (reader, writer):
            logging.info("Connected to MCP server. Initializing session...")
            async with ClientSession(reader, writer) as mcp_session:
                await mcp_session.initialize()
                logging.info("MCP session initialized.")
                logging.info("Calling tool 'chat_with_manual' with query: %r", user_query)

                try:
                    response = await mcp_session.call_tool(
                        "chat_with_manual",
                        arguments={"query": user_query}
                    )
                    logging.info("Received response from server.")

                    # Process response (handle potential wrapping)
                    final_answer = response
                    # Example check if response is wrapped like TextContent (adjust based on actual MCP response)
                    if hasattr(response, 'content') and response.content and isinstance(response.content, list):
                         first_content = response.content[0]
                         if hasattr(first_content, 'text'):
                             final_answer = first_content.text
                    elif isinstance(response, str): # If it's already a string
                         final_answer = response

                    # Ensure we have a string before returning
                    response_text = str(final_answer) if final_answer is not None else "Received an empty response."


                except Exception as tool_call_e:
                    logging.error("Error calling tool 'chat_with_manual': %s", tool_call_e, exc_info=True)
                    error_message = f"Error calling backend tool: {tool_call_e}"

    except Exception as connection_e:
        logging.error("Failed to connect or communicate with the MCP server: %s", connection_e, exc_info=True)
        error_message = f"Could not connect to or run the server: {connection_e}"

    # Ensure server process is terminated (stdio_client context manager handles this)
    logging.info("MCP client connection closed.")
    return response_text, error_message


def get_rag_response(user_query: str) -> tuple[str | None, str | None]:
    """
    Synchronous wrapper to run the asynchronous MCP call.
    Returns (response_text, error_message)
    """
    # --- Running asyncio logic within Flask ---
    # Note: Running asyncio.run() inside a synchronous Flask route handler
    # is generally discouraged for production scalability, but works for simple demos.
    # More robust solutions involve running the server independently (e.g., via TCP)
    # or using async Flask extensions.
    try:
        # For Python 3.7+
        response_text, error_message = asyncio.run(async_mcp_call(user_query))
        return response_text, error_message
    except RuntimeError as e:
        # Handle cases like "asyncio.run() cannot be called from a running event loop"
        # This might happen in certain deployment scenarios or with specific Flask extensions.
        logging.error("Asyncio runtime error: %s. Trying get_event_loop().run_until_complete()", e)
        try:
            loop = asyncio.get_event_loop()
            response_text, error_message = loop.run_until_complete(async_mcp_call(user_query))
            return response_text, error_message
        except Exception as fallback_e:
             logging.error("Fallback asyncio execution failed: %s", fallback_e)
             return None, f"Internal server error during async execution: {fallback_e}"
    except Exception as e:
        logging.error("Unexpected error during RAG response retrieval: %s", e)
        return None, f"Unexpected error: {e}"


# --- Conversation History Management ---

def trim_history(history: list, max_chars: int) -> list:
    """Removes oldest messages if total character count exceeds max_chars."""
    current_chars = sum(len(msg.get('content', '')) for msg in history)
    # Keep removing the oldest messages (index 1 and 2, skipping potential system prompt at 0)
    # until the total character count is below the limit.
    while current_chars > max_chars and len(history) > 2: # Always keep at least one pair if possible
        # Remove the first user message and the first assistant response
        removed_user = history.pop(0) # Assuming user message is older
        current_chars -= len(removed_user.get('content', ''))
        if history: # Ensure there's another message to remove (the assistant's reply)
            removed_assistant = history.pop(0)
            current_chars -= len(removed_assistant.get('content',''))
        logging.info(f"History trimmed. Current chars: {current_chars}")

    # Simplified trimming (removes oldest message regardless of role):
    # while current_chars > max_chars and len(history) > 1:
    #     removed_message = history.pop(0) # Remove the very first message
    #     current_chars -= len(removed_message.get('content', ''))
    #     logging.info(f"History trimmed. Current chars: {current_chars}")
    return history


# --- Flask Routes ---

@app.route('/')
def index():
    """Displays the chat interface."""
    if 'history' not in session:
        session['history'] = [] # Initialize history for new session
        # Optional: Add an initial system message
        # session['history'].append({"role": "system", "content": "You are chatting with PDF manuals."})
    return render_template('chat.html', history=session['history'])

@app.route('/chat', methods=['POST'])
def chat():
    """Handles user messages and gets bot responses."""
    user_message = request.form.get('message')

    if not user_message:
        # Handle empty submission if required='required' is removed from input
        return redirect(url_for('index'))

    # Ensure history exists in session
    if 'history' not in session:
        session['history'] = []

    # Add user message to history
    session['history'].append({"role": "user", "content": user_message})

    # --- Get Response from RAG Backend ---
    bot_response, error = get_rag_response(user_message)
    # ------------------------------------

    if error:
        session['history'].append({"role": "error", "content": error})
    elif bot_response:
        session['history'].append({"role": "assistant", "content": bot_response})
    else:
        # Handle case where bot gives no response and no error
        session['history'].append({"role": "error", "content": "No response received from the backend."})

    # Trim history after adding new messages
    session['history'] = trim_history(session['history'], MAX_HISTORY_CHARS)

    # Mark session as modified since we changed a mutable list inside it
    session.modified = True

    return redirect(url_for('index')) # Redirect back to display the updated chat

@app.route('/clear')
def clear_chat():
    """Clears the chat history from the session."""
    session.pop('history', None) # Remove history safely
    logging.info("Chat history cleared.")
    return redirect(url_for('index'))

# --- Run the App ---
if __name__ == '__main__':
    # Make sure the static folder is correctly configured relative to this script
    # app.static_folder = 'static'
    # Consider using waitress or gunicorn for production instead of Flask's dev server
    app.run(debug=True, host='0.0.0.0', port=5001) # Run on port 5001 to avoid conflict if server.py uses 5000

実行方法：

1. コードを保存する： Flask のコードを frontend.py として保存し、HTML テンプレートを templates フォルダーに保存してください。また、watsonx-wallpaper.jpg が static/assets にあることを確認してください。
2. 依存ファイルを配置する： server.py、.env ファイル、documents フォルダーが frontend.py と同じディレクトリーにあるか、Python のパスからアクセスできることを確認してください。
3. ライブラリをインストールする：

   pip install Flask meta-compute-protocol python-dotenv ibm-watsonx-ai chromadb pypdf
   

4. Flask アプリを実行する：

   python frontend.py
   

5. ブラウザーでアクセスする： ウェブブラウザーを開き、http://127.0.0.1:5001 にアクセスしてください（別のマシン／ネットワークで実行している場合は、http://<your-ip-address>:5001 にアクセスしてください）。

Docker イメージ：イメージを取得して別の場所で実行する（任意）

別のマシンで：

1. イメージを取得する

docker pull ruslanmv/watsonx-rag-chatbot:latest

2. コンテナーを実行する

.env、documents、chroma_db_data のローカルコピーが存在することを確認してください。

docker run --rm -p 5001:5001 \
  -v "$(pwd)/.env:/app/.env" \
  -v "$(pwd)/documents:/app/documents" \
  -v "$(pwd)/chroma_db_data:/app/chroma_db_data" \
  --name rag-chat-container \
  ruslanmv/watsonx-rag-chatbot:latest

前述のとおり、Windows 用にボリュームのパスを調整してください。

メモと考慮事項

• .env のセキュリティ：
  開発ではボリュームのマウントが一般的です。本番環境では、-e VARIABLE=value または Docker secrets の使用を検討してください。

• 永続化：
  chroma_db_data ボリュームを使用すると、実行のたびにベクトルインデックスを再利用できます。

• 停止／削除：

  • フォアグラウンドコンテナを停止するには、Ctrl+C を使用します。
  • デタッチモードで実行した場合は、docker stop rag-chat-container と docker rm rag-chat-container を使用します。

カスタム Python RAG サーバーと Claude Desktop の統合

PDF ドキュメントに基づいて質問に回答できる、強力な Retrieval-Augmented Generation（RAG）サーバーを Python（server.py）で構築しました。これを Claude Desktop 内で直接操作できたら便利だと思いませんか？
Meta Compute Protocol（MCP） があれば実現できます！

このガイドでは、カスタムの server.py を起動して通信できるように Claude Desktop を設定する方法を説明します。

前提条件

1. Claude Desktop
   macOS または Windows 用の最新バージョン。 こちらからダウンロードしてください。

2. Python 3.x
   システムの PATH に含まれていることを確認してください。次のコマンドで確認できます。

   python --version   # or
   python3 --version
   

3. server.py プロジェクト
   次のファイルとフォルダーを、場所がわかるフォルダーに用意してください。

   • server.py
   • requirements.txt（同じ Python 環境に依存関係をインストールしてください）
   • .env ファイル（API キーなど）
   • PDF を含む documents/ フォルダー
   • 永続ベクトルデータベース用の chroma_db_data/ フォルダー

Step 1 – Claude Desktop の設定ファイルを探す

1. Claude Desktop の設定を開く
   • macOS：メニューバーの Claude ▸ Settings…
   • Windows：システムトレイメニューの Claude アイコン ▸ Settings…

2. Developer ▸ Edit Config
   サイドバーの Developer をクリックし、次に Edit Config をクリックします。
   Claude は既存のファイルを開くか、新しいファイルを作成してファイルシステム上に表示します。

一般的な場所：

OS	パス
macOS	~/Library/Application Support/Claude/claude_desktop_config.json
Windows	%APPDATA%\Claude\claude_desktop_config.json (例：C:\Users\<You>\AppData\Roaming\Claude\claude_desktop_config.json)

Step 2 – カスタム server.py を設定する

テキストエディターで claude_desktop_config.json を開き、トップレベルの mcpServers オブジェクトにエントリーを追加します。

{
  "mcpServers": {
    "rag_chatbot": {
      // Name shown inside Claude (change if you like)
      "command": "python3",          // Or "python" or full path to python.exe
      "args": [
        "/absolute/path/to/your_project/server.py"
      ],
      "workingDirectory": "/absolute/path/to/your_project/"   // ← highly recommended
    }

    /* You can add additional servers here, e.g.
    , "filesystem": { ... }
    */
  }
}

{
    "mcpServers": {
      "rag_chatbot": {
   
        "command": "wsl",
  
        "args": [
          "bash",
          "-c",
          "cd /mnt/c/blog/watsonx-rag-mcp-server && source .venv/bin/activate && python3 server.py"
        ],
  
        "workingDirectory": "/mnt/c/blog/watsonx-rag-mcp-server"
      }
    }
  }

ポイント

依存関係： Claude Desktop はコマンドを直接実行します。選択した Python インタープリターで、requirements.txt に記載されたすべてのパッケージが（グローバルまたは指定した環境で）利用できることを確認してください。

Step 3 – Claude Desktop を再起動する

Claude Desktop を完全に終了してから再度開き、新しい設定を読み込みます。 設定を開き、開発者モードを有効にします。

ステップ 4 – 統合を確認する

1. チャットウィンドウで、メッセージボックスの右下にある**ハンマーアイコン（🔨）**を探します。
2. クリックします。
3. カスタムツールの下に rag_chatbot（または付けた名前）が表示されます。
4. そのツール（例：chat_with_manual）がその下に表示されます。

何かが見つからない場合：

• 絶対パスと JSON の構文をもう一度確認します。
• Python のパスが正しいことを確認します。
• 同じコマンドをターミナルで手動実行し、実行時エラーがないか確認します。
• Claude Desktop のトラブルシューティングドキュメントを参照します。

ステップ 5 – RAG サーバーとチャットする

次のようなプロンプトを試してみてください。

• 「RAG チャットボットを使って、マニュアルに記載されている最大飛行時間を教えてください」
• 「PDF サーバーに、さまざまな飛行モードについて尋ねてください」
• 「rag_chatbot ツールで、ドローンのキャリブレーション方法を教えてもらえますか？」

Claude は意図を検出し、server.py 経由で chat_with_manual ツールを呼び出して、その結果を返信に取り込みます。

次のように尋ねることができます。

How many flight modes it has and explain them?

すると、次のような結果が得られます。

JSON 設定ファイルを 1 つ編集するだけで、カスタム Python MCP サーバーを Claude Desktop と統合できました。これで、次のことができます。

• Python で強力な独自ツールを開発する
• お気に入りのチャットインターフェース内でシームレスにツールを利用する
• PDF のドメイン固有の知識を Claude に活用させる

結論

おめでとうございます！永続的なベクトルストレージに ChromaDB を、標準化された通信に Model Context Protocol（MCP）を使用する、機能的な Watsonx.ai RAG サーバーを構築できました。このサーバーは、提供された PDF ドキュメントの内容に基づく質問に効果的に回答でき、ドメイン固有のチャットボットや情報検索システムを構築するための強力なパターンを示しています。ユーザー入力を受け取るようにクライアントを変更したり、このサーバーをより大規模なアプリケーションに統合したりできるようになりました。