[FEATURE] mcp tool definition tutorial

Preflight Checklist

• [x] I have searched existing requests and this feature hasn't been requested yet
• [x] This is a single feature request (not multiple features)

Problem Statement

I have written an mcp tool code, but when I open Claude Code, it cannot recognize this tool. I don't know how to define the tool correctly. The tool code is:
import asyncio
import os
import sys
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import chromadb
from chromadb.config import Settings
from docx import Document

初始化MCP服务器

app = Server("rag-server")

获取当前脚本所在目录的绝对路径

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_DIR = os.path.dirname(BASE_DIR) # 项目根目录

修正的路径：使用绝对路径

DOC_PATH = os.path.join(PROJECT_DIR, ".claude", "skills", "product_rag", "白皮书.docx")
CHROMA_DB_DIR = os.path.join(PROJECT_DIR, "chroma_db")

print("=" * 50)
print("MCP服务器启动信息")
print(f"脚本目录: {BASE_DIR}")
print(f"项目目录: {PROJECT_DIR}")
print(f"文档路径: {DOC_PATH}")
print(f"向量数据库: {CHROMA_DB_DIR}")
print("=" * 50)

初始化向量数据库

chroma_client = chromadb.PersistentClient(path=CHROMA_DB_DIR)
collection = chroma_client.get_or_create_collection(name="product_whitepaper")

def extract_text_from_docx(doc_path: str) -> str:
"""从Word文档提取文本"""
print(f"正在读取Word文档: {doc_path}")
doc = Document(doc_path)
text = "\n".join([para.text for para in doc.paragraphs if para.text.strip()])
print(f"文档读取完成，字符数: {len(text)}")
return text

def chunk_text(text: str, chunk_size: int = 1000) -> list[str]:
"""将文本分块"""
chunks = []
for i in range(0, len(text), chunk_size):
chunks.append(text[i:i + chunk_size])
print(f"文本分块完成，共 {len(chunks)} 块")
return chunks

def index_document():
"""索引Word文档到向量数据库"""
print("开始索引文档...")

if not os.path.exists(DOC_PATH):
print(f"错误：文档不存在！路径: {DOC_PATH}")
return False

try:
text = extract_text_from_docx(DOC_PATH)
chunks = chunk_text(text)

# 清空旧数据
collection.delete(where={"source": "白皮书"})
print("已清空旧索引")

# 添加新数据
for i, chunk in enumerate(chunks):
collection.add(
documents=[chunk],
ids=[f"chunk_{i}"],
metadatas=[{"source": "白皮书", "chunk_index": i}]
)

print(f"索引完成，共添加 {len(chunks)} 个文档块")
return True
except Exception as e:
print(f"索引过程中出错: {e}")
import traceback
traceback.print_exc()
return False

@app.list_tools()
async def list_tools() -> list[Tool]:
"""列出可用工具"""
return [
Tool(
name="query_whitepaper",
description="查询产品白皮书知识库",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "用户查询问题"
}
},
"required": ["query"]
}
),
Tool(
name="reindex_whitepaper",
description="重新索引Word文档",
inputSchema={
"type": "object",
"properties": {}
}
)
]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""调用工具"""
print(f"调用工具: {name}, 参数: {arguments}")

if name == "query_whitepaper":
query = arguments.get("query", "")
print(f"执行查询: {query}")

try:
results = collection.query(
query_texts=[query],
n_results=3
)

if results["documents"] and results["documents"][0]:
context = "\n\n---\n\n".join(results["documents"][0])
print(f"查询成功，返回 {len(results['documents'][0])} 个结果")

return [TextContent(
type="text",
text=f"知识库检索结果：\n\n{context}"
)]
else:
print("查询结果为空")
return [TextContent(
type="text",
text="未在知识库中找到相关信息。"
)]
except Exception as e:
print(f"查询出错: {e}")
return [TextContent(
type="text",
text=f"查询过程中出错: {str(e)}"
)]

elif name == "reindex_whitepaper":
success = index_document()
status = "成功" if success else "失败"
print(f"重新索引结果: {status}")

return [TextContent(
type="text",
text=f"文档重新索引{status}"
)]

return [TextContent(type="text", text="未知工具")]

async def main():
print("MCP服务器初始化...")

# 首次启动时索引文档
print("检查是否需要索引文档...")
try:
count = collection.count()
print(f"当前向量数据库已有 {count} 个文档块")
if count == 0:
print("数据库为空，开始索引文档")
index_document()
else:
print("使用现有索引")
except Exception as e:
print(f"检查数据库时出错，开始索引: {e}")
index_document()

print("MCP服务器启动，等待连接...")

try:
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
except Exception as e:
print(f"服务器运行出错: {e}")
import traceback
traceback.print_exc()

if __name__ == "__main__":
print("开始运行MCP服务器...")
asyncio.run(main())

Proposed Solution

Can you provide an official tutorial? Thank you.

Alternative Solutions

_No response_

Priority

Medium - Would be very helpful

Feature Category

Documentation

Use Case Example

_No response_

Additional Context

_No response_