使用MCP构建Watsonx.ai聊天机器人RAG服务器
 在本教程中，您将学习如何使用IBM Watsonx.ai、ChromaDB构建功能齐全的检索增强生成（RAG）服务器进行矢量索引，并通过模型上下文协议（MCP）Python SDK公开它。到最后，您将拥有一个处理PDF文档的独立服务器和一个对其进行查询的客户端，利用基于特定数据的大型语言模型的强大功能。最后，我们展示了如何直接集成到Claude Desktop中。
这 模型上下文协议（MCP） 标准化应用程序和LLM之间的接口。使用MCP，您可以将提供上下文、执行代码和管理用户交互的关注点分开。MCP Python SDK实现了完整的MCP规范，允许您：
- 公开资源： 将数据传递给LLM（类似于GET端点）。
- 定义工具： 提供执行操作或计算的功能（如POST端点）。
- 创建提示： 提供可重用、模板化的交互。
我们的服务器将实现一个客户端可以调用的RAG“工具”。
先决条件
- 已安装Python 3.8+（代码使用
typing.Union，与旧版本兼容，但通常建议使用较新的Python）。 - 访问已配置的IBM Cloud帐户 Watsonx.cn 服务。您需要一个API密钥和项目ID。
pip用于安装Python包。
项目结构
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（元计算协议）服务器公开聊天功能。
目标是创建一个能够基于以下内容回答问题的系统 *具体地* 在提供的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 # 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) = 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:
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.构建知识库：创建矢量数据库(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对于每个文档。 - 数据准备： 它准备了三份清单：
docs（实际文本块），metas（与每个块关联的元数据，如源文件名），以及ids（每个块的唯一标识符）。唯一ID对于以后可能更新或删除特定块非常重要。 - ChromaDB交互： 使用
client.get_or_create_collection为了安全地获取集合的句柄，如果是第一次创建，请创建它。 - 批量索引： 它使用以下命令将文档、元数据和ID添加到ChromaDB集合中
collection.add()。这是分批完成的(batch_size = 100)为了提高效率，并避免一次可能有数千个块淹没数据库连接或内存。ChromaDB使用默认嵌入模型自动处理将文本块转换为向量嵌入的过程。 - 全球派遣： 一旦被填充
collection对象被分配给全局pdf_collection变量。
5.智能启动：初始化矢量数据库(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说明：
- 它创造了
Credentials使用从.env文件。 - 一
APIClient被实例化用于一般交互（尽管这里不直接用于生成）。 - A.
ModelInference创建对象，指定所需的MODEL_ID（例如，“ibm/granite-13b-instruct-v2”）、凭据和PROJECT_ID.这个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()找到NUM_RESULTS_RAG向量嵌入与用户的向量嵌入最接近（最相似）的文档块query。它明确要求documents（文本）将包含在结果中。 - 加强： 它检查是否找到了任何文件。如果是，它将它们的文本内容合并为一个
context_string，分开\n\n---\n\n为了清楚起见。如果没有找到文档，它将返回一条特定的消息。 - 生成：
- 快速工程： 精心制作的提示(prompt_template)创建。这是 *关键的* 对于RAG，它明确地“告诉”LLM（Watsonx.ai模型）它的角色，并指示它回答用户的 query *仅基于提供的 context_string*这会阻止LLM使用其一般知识，并迫使其坚持文档内容。 - 参数： 生成参数，如 max_new_tokens （答案长度）， decoding_method （法学硕士如何选择下一个单词——“贪婪”是决定性的），以及 stop_sequences （用信号通知LLM停止生成的序列）。 - LLM电话： model.generate_text() 将提示和参数发送到Watsonx.ai。 - 响应处理： 该代码解析来自Watsonx.ai的JSON响应，以提取 generated_text基本的后处理尝试清理可能突然结束的答案。ChromaDB查询和Watsonx.ai API调用都包含错误处理。
9.放映时间：主执行块
最后，the 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这是至关重要的——在确认向量数据库准备就绪（无论是加载的还是新构建的）之前，服务器都不应该开始接受请求。如果初始化失败，则脚本退出。 - 启动服务器： 如果DB准备就绪，
mcp.run()启动MCP服务器。此函数通常会阻塞，这意味着脚本将在此处等待，监听传入的客户端请求并将其路由到适当的位置@mcp.tool功能(chat_with_manual在这种情况下）。 - 关机： 脚本只会继续执行
mcp.run()当服务器关闭时（例如通过客户端优雅地断开连接或通过像Ctrl+C这样的中断信号）。
这种模块化结构使数据如何从原始PDF流到LLM生成的上下文感知答案变得相对清晰。下一步是创建一个客户端应用程序，该应用程序连接到此MCP服务器并与 chat_with_manual 工具。
______________________________________________________________________
步骤4。与RAG机器人对话： client.py 脚本
在上一节中，我们构建了强大的 server.py RAG聊天机器人的后端。但我们实际上是如何做到的 *使用* 它？那就是 client.py 这个脚本充当用户界面（尽管现在是一个简单的命令行界面），负责启动服务器、发送问题和显示从我们的RAG系统收到的答案。
它使用元计算协议（MCP）客户端库，专门配置为通过标准输入/输出（STDIO）进行通信，这非常适合与 FastMCP 我们设置的服务器 server.py.
这是完整的 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
让我们来分解一下这个客户端是如何工作的：
- 进口和记录：
- asyncio：至关重要，因为MCP的客户端操作（连接、发送、接收）是异步的。 - mcp：我们进口 ClientSession 为了管理通信协议， StdioServerParameters 定义如何启动我们的服务器，以及 stdio_client 其是处理启动服务器和提供通信流的上下文管理器。 - logging：设置基本日志记录以显示客户端活动。
- 这
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 允许客户端发送数据的对象 *向* 服务器的stdin和读取数据 *从* 服务器的stdout是异步的。 - 至关重要的是，它还确保服务器进程在以下情况下正确终止： async with 块退出（正常或由于错误）。 - MCP会话(ClientSession): 里面 stdio_client 块， async with ClientSession(reader, writer) as session: 获取原始读写器流，并将其封装在MCP会话处理程序中。此对象知道如何根据MCP协议格式化消息。 - 初始化(session.initialize()): 在拨打电话之前，客户端和服务器会快速握手，以确保它们已准备好使用MCP协议进行通信。 - 定义查询(user_msg): 一个简单的字符串变量包含我们想问RAG机器人的问题。 - 调用工具(session.call_tool): 这是关键的互动点。 - await session.call_tool(...)：通过已建立的会话向服务器发送请求。 - "chat_with_manual"：此字符串 *必须完全匹配* 装饰有的功能名称 @mcp.tool() 在……里面 server.py. - arguments={"query": user_msg}：此字典包含要传递给服务器函数的参数。钥匙 "query" 必须与中的参数名称匹配 chat_with_manual(query: str) 服务器上的定义。 - 那么，客户 await这是服务器的响应。 - 处理响应： 这 response 变量包含任何 chat_with_manual 服务器上返回的函数（应该是一个包含答案的字符串）。该代码包含一个检查(if hasattr(response, 'content')...)对于MCP可能将响应包装在更复杂的对象中，试图提取原始文本的情况。最后，它以格式化的方式打印原始问题和收到的答案。 - 错误处理： try...except 块用于捕获潜在的问题，例如无法启动服务器、连接问题或服务器端工具调用过程中发生的错误（这些问题将被中继回来）。
- 运行客户端(
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 发生了什么：
client.py开始。- 它执行
python server.py作为一个子流程。 server.py启动，初始化Watsonx.ai连接。server.py检查ChromaDB数据库。
- 第一次： 它找不到它，因此它将在 documents 文件夹，将它们分块，并将嵌入保存到 ./chroma_db_data 目录。这可能需要一些时间，具体取决于PDF的大小和数量。 - 后续时间： 它会发现 ./chroma_db_data 目录和里面的集合，快速加载，跳过处理步骤。
server.py启动MCP服务器，监听STDIO。client.py通过STDIO连接到服务器并初始化MCP会话。client.py将预定义的查询（“有多少种飞行模式…”）发送到chat_with_manual工具。server.py接收请求，查询ChromaDB，获取相关文本块，使用上下文构建提示，调用Watsonx.ai，获取答案，并将其发送回客户端。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-aiSDK与之交互。 - ChromaDB用于矢量索引： 充当RAG系统的内存。它从PDF中提取文本块，使用嵌入模型将其转换为数值表示（向量嵌入）（如果未指定，ChromaDB将使用默认模型），并存储它们。当你查询时，它会找到嵌入最接近（语义上最相似）你的查询嵌入的块。我们使用
chromadb.PersistentClient使此数据库在运行之间存活。 - 模型上下文协议（MCP）： 为客户端和服务器提供了一种标准化的通信方式。我们没有定义自定义API端点，而是使用MCP的概念：
- 工具： 我们的 chat_with_manual 该功能作为MCP工具公开。客户端调用此工具执行RAG操作。这就像一个专门的函数调用或web术语中的POST请求。 - *（资源和提示）：* 虽然在此特定示例中没有大量使用，但MCP也允许直接暴露数据源，如 *资源* （如GET请求）和定义可重用 *鼓励*.
提示和故障排除
- 检查日志： 两者
server.py和client.py生成详细的日志。如果出现问题，请检查终端输出[INFO],[WARNING]，或[ERROR]信息。增加日志记录级别（例如。，logging.DEBUG)如有需要，请提供更多详细信息。 .env配置： 双击检查API密钥、URL和项目ID.env文件是正确的，并且文件正在加载（文件名中没有拼写错误）。- PDF可访问性： 确保PDF文件在
documents文件夹是可读的，不受密码保护（或者其中的简单解密尝试pdf_to_text作品）。损坏的PDF可能会导致错误。 - 依赖关系： 确保中列出的所有库
requirements.txt已正确安装在活动虚拟环境中。 - 强制重建： 如果您更新PDF并希望服务器重新索引它们，则需要 删除
chroma_db_data目录 跑步前client.py再一次。然后，服务器将检测到数据库的缺失并触发build_new_vector_db过程。 - Watsonx.ai型号： 默认值
ibm/granite-13b-instruct-v2使用。对于其他模型（如Llama 3），您可能会得到不同的结果或需要不同的参数。你可以改变MODEL_ID在你的.env文件。
前端
让我们创建一个简单的Flask web界面，就像ChatGPT风格的聊天一样，与您的 server.py 后端。
该解决方案将：
- 使用Flask作为web框架。
- 使用提供的
base.html与Watsonx壁纸。 - 使用模拟聊天界面
chat.html. - 启动并与
server.py对于每条用户消息，使用从以下内容改编的逻辑client.py. 注： 这对于生产来说效率低下（反复重新初始化服务器/DB连接），但在没有复杂异步管理的标准Flask设置中实现起来更简单。 - 在Flask会话中维护对话历史记录。
- 根据字符数（作为令牌的代理）修剪对话历史记录，使其易于管理。
项目结构：
确保你的文件是这样组织的：
/chatbot
├── static/
│ └── assets/
│ └── watsonx-wallpaper.jpg
{% block title %}Watsonx RAG Chat{% endblock %}
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 */
}
{# Default background if image fails #}
{% block content %}{% endblock %}
{# Optional: Add scripts here if needed globally #}
// 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.
*（注：更新顺风CDN链接并添加 background-attachment: fixed 以及flexbox布局，以更好地处理固定背景的内容居中和滚动）。*
______________________________________________________________________
2.聊天界面模板(templates/chat.html)
此模板扩展 base.html 并创建聊天UI。
{% extends 'base.html' %}
{% block title %}Watsonx RAG Chat{% endblock %}
{% block content %}
Chat with PDF Manuals
Clear Chat
{% if not history %}
Ask a question about the PDF documents!
{% endif %}
{% for message in history %}
{% if message.role == 'user' %}
{{ message.content }}
{% elif message.role == 'assistant' %}
{# Use safe filter or pre tags if response might contain HTML/code #}
{{ message.content }}
{% elif message.role == 'error' %}
Error: {{ message.content }}
{% endif %}
{% endfor %}
Send
{% endblock %}*（注：已添加 pre 添加了助理消息标签，以保留换行符等格式 id="chatbox" 和基本的JavaScript base.html 对于自动滚动，添加了一个清晰的聊天链接和基本错误显示）*
______________________________________________________________________
3.瓶子应用(frontend.py)
这是web前端的主要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______________________________________________________________________
如何跑步：
- 保存代码： 将Flask代码另存为
frontend.py，中的HTML模板templates文件夹，并确保watsonx-wallpaper.jpg在...里static/assets. - 地点依赖关系： 确保
server.py，你的.env文件，以及documents文件夹与位于同一目录中frontend.py或者可以在Python路径中访问。 - 安装库：
pip install Flask meta-compute-protocol python-dotenv ibm-watsonx-ai chromadb pypdf- 运行Flask应用程序：
python frontend.py
- 在浏览器中访问： 打开您的网络浏览器并转到
http://127.0.0.1:5001（或http://:5001如果在不同的机器/网络上运行）。
Docker镜像：在其他地方拉取并运行镜像（可选）
在另一台机器上：
1.拉取图像
docker pull ruslanmv/watsonx-rag-chatbot:latest2.运行容器
确保本地副本 .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的秘密。
- 坚持不懈：\
这 chroma_db_data volume允许跨运行重用向量索引。
- 停止/删除：
- 使用 Ctrl+C 停止前台容器。 - 使用 docker stop rag-chat-container 和 docker rm rag-chat-container 如果运行分离。
将自定义Python RAG服务器与Claude Desktop集成
我们用Python构建了一个强大的检索增强生成（RAG）服务器(server.py)它可以根据您的PDF文档回答问题。现在，直接在里面与它互动不是很棒吗 克劳德桌面?\ 感谢the 元计算协议（MCP）你可以！
本指南将引导您完成配置Claude Desktop以启动和与您的自定义通信 server.py.
##先决条件
- 克劳德桌面\
*适用于macOS或Windows的最新版本。* 下载 这里.
- Python 3.x\
确保它在你的系统上PATH.通过以下方式进行验证：
python --version # or
python3 --version- 你的
server.py项目\
在已知文件夹中准备好这些：
- server.py - requirements.txt（在同一Python环境中安装的依赖项） - .env文件（API密钥等） - documents/包含PDF的文件夹 - chroma_db_data/文件夹（持久矢量数据库）
##步骤1-找到Claude Desktop配置文件
- 打开克劳德桌面设置
- macOS： 克劳德▸设置… 在菜单栏中 - 窗户： 克劳德图标▸设置… 在系统托盘菜单中
- 开发者▸编辑配置\
点击 开发者 在侧边栏中，然后 编辑配置.\ Claude要么打开现有文件，要么创建一个新文件并将其显示在您的文件系统中。
典型位置：
|操作系统|路径| |---------|------| |macOS| ~/Library/Application Support/Claude/claude_desktop_config.json | |窗户| %APPDATA%\Claude\claude_desktop_config.json
（例如C:\Users\\AppData\Roaming\Claude\claude_desktop_config.json) | 
##步骤2-配置您的自定义 server.py
打开 claude_desktop_config.json 在文本编辑器中，在顶层下添加一个条目 mcpServers 对象。
Example configuration
{
"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"
}
}
}###关键点
| 领域 | 目的 | 提示 |
|---|---|---|
"rag_chatbot" | Claude显示的内部名称 | 选择任何令人难忘的名称 |
"command" | 可执行文件启动 | 使用 python/python3 如果在PATH上，否则 完整路径 （例如/usr/local/bin/python3 在macOS或 C:\\Python310\\python.exe 在Windows上--注意双反斜杠） |
"args" | 传递给命令的参数 | 通常只是绝对路径 server.py |
"workingDirectory" （可选但推荐） | 运行命令前Claude切换到的目录 | 确保 .env, documents/等被正确找到 |
依赖关系： Claude Desktop直接运行该命令。确保所选Python解释器包含来自的所有包 requirements.txt 可用（全局或在您指定的环境中）。##步骤3–重新启动克劳德桌面
完全关闭Claude Desktop，然后重新打开它以加载新配置。 打开设置并启用开发人员模式。
______________________________________________________________________
##步骤4–验证集成
- 在聊天窗口中，查找 锤子图标(🔨) 在消息框的右下角。
- 点击它。
- 你应该看看
rag_chatbot（或你给的任何名字） 自定义工具. - 其工具（例如
chat_with_manual)应该出现在它下面。
如果缺少了什么：
- 重新检查绝对路径和JSON语法。
- 确认Python路径正确。
- 在终端中手动运行相同的命令以捕获运行时错误。
- 请参阅Claude Desktop故障排除文档。
______________________________________________________________________
##步骤5–与您的RAG服务器聊天
尝试以下提示：
- “使用 RAG聊天机器人，手册中提到的最长飞行时间是多少？”
- “向PDF服务器询问不同的飞行模式。”
- “可以 rag_chatbot 工具告诉我如何校准无人机？”
克劳德会发现意图，打电话给你 chat_with_manual 工具via server.py，并将结果纳入其答复中。
你可以问
How many flight modes it has and explain them?
你得到 
通过编辑单个JSON配置文件，您已经将自定义Python MCP服务器与Claude Desktop集成在一起。现在您可以：
- 在Python中开发强大的定制工具
- 在您最喜欢的聊天界面中无缝访问它们
- 从您的PDF中为Claude提供特定领域的知识
结论
祝贺您已成功构建了一个功能齐全的Watsonx.ai RAG服务器，该服务器与ChromaDB集成用于持久矢量存储，并与模型上下文协议（MCP）集成用于标准化通信。该服务器可以根据您提供的PDF文档的内容有效地回答查询，展示了构建特定领域聊天机器人和信息检索系统的强大模式。现在，您可以调整客户端以接受用户输入或将此服务器集成到更大的应用程序中。