完整的分步指南:从头开始构建MCP服务器
概述
本指南将引导您构建一个完整的MCP(模型上下文协议)服务器,该服务器具有用于库存管理的REST API功能。我们将使用 uv 作为Python包管理器,逐步构建所有内容。
先决条件
- 已安装Python 3.13或更高版本
- Python语法的基本理解
- 终端/命令提示符访问
______________________________________________________________________
第一阶段:项目设置和环境
步骤1.1:安装uv包管理器
为什么? uv 是一个快速、现代的Python包管理器,可以有效地处理依赖关系。
行动:
# Windows (PowerShell)
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
# Or using pip (if you have it)
pip install uv验证安装:
uv --version步骤1.2:创建项目目录
为什么? 在专用文件夹中组织您的项目。
行动:
mkdir inventory-mcp
cd inventory-mcp步骤1.3:使用uv初始化Python项目
为什么? 创建项目结构和依赖关系管理文件。
行动:
# Windows (PowerShell)
# Create a new directory for our project
uv init ${PROJECT_NAME}
cd ${PROJECT_NAME}
# Create virtual environment and activate it
uv venv
source .venv/bin/activate
# Install dependencies
uv add "mcp[cli]" httpxThis creates:
pyproject.toml- Project configuration and dependencies.python-version- Python version specification
Step 1.4: Create Main Python File
Why: This will be our main server file.
Action:
# Create main.py (empty file to start)
touch main.py # Linux/Mac
# Or on Windows PowerShell:
New-Item main.py -ItemType File______________________________________________________________________
第二阶段:理解和添加依赖关系
步骤2.1:在pyproject.toml中配置依赖关系
为什么? 在一个地方定义所有必需的包,以便于管理。
行动: 编辑 pyproject.toml:
[project]
name = "inventory-mcp"
version = "0.1.0"
description = "Inventory Management MCP Server with REST API"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"mcp[cli]>=1.22.0", # MCP framework for Claude Desktop
"fastapi>=0.104.0", # REST API framework
"uvicorn[standard]>=0.24.0", # ASGI server for FastAPI
"pydantic>=2.0.0", # Data validation (usually included with FastAPI)
]每种依赖关系的解释:
mcp[cli]:提供FastMCP类,用于创建Claude Desktop可以连接到的MCP服务器fastapi:用于构建带有自动文档的REST API的现代web框架uvicorn:ASGI服务器运行FastAPI应用程序pydantic:数据验证库(随FastAPI自动安装,但显式更好)
步骤2.2:安装依赖项
为什么? 下载并安装所有必需的软件包。
行动:
uv sync这将创建:
uv.lock-锁定的依赖关系版本- 虚拟环境(由uv管理)
______________________________________________________________________
第三阶段:构建代码——理解结构
步骤3.1:标准库导入(第16-22行)
为什么? 这些是内置的Python模块,无需安装。
添加到main.py:
import json # For reading/writing JSON files (data persistence)
import os # For file path operations
import sys # For command-line arguments (http vs mcp mode)
import uuid # For generating unique product IDs
from typing import Dict, List, Optional # Type hints for better code clarity
from datetime import datetime # For timestamps (if needed later)为什么每次导入:
json:以JSON格式存储库存数据os:获取文件路径的脚本目录sys:检查命令行参数以在MCP/HTTP模式之间切换uuid:生成唯一的产品IDtyping:类型提示有助于捕获错误和文档代码datetime:用于审计日志等未来功能
步骤3.2:第三方进口(第21-29行)
为什么? 这些需要通过紫外线进行安装。
添加到main.py:
from pydantic import BaseModel, Field # Data models with validation
from fastapi import FastAPI, HTTPException, Security, status, Depends, Query, Path
from fastapi.security.api_key import APIKeyHeader
from fastapi.responses import JSONResponse
from mcp.server.fastmcp import FastMCP # MCP server framework
import uvicorn # Web server for FastAPI为什么每次导入:
pydantic.BaseModel:创建具有自动验证功能的数据模型pydantic.Field:在模型字段中添加描述和示例fastapi.FastAPI:REST API的主要应用程序类fastapi.HTTPException:引发HTTP错误(404、400等)fastapi.Query/Path:从URL提取参数FastMCP:创建Claude Desktop连接到的MCP服务器uvicorn:运行FastAPI服务器
______________________________________________________________________
阶段4:数据模型(Pydantic Schema)
步骤4.1:产品型号
为什么? 通过验证定义库存项目的结构。
添加到main.py:
class Product(BaseModel):
"""Represents a product in the inventory system."""
product_id: str = Field(..., json_schema_extra={"example": "P-001"})
name: str = Field(..., json_schema_extra={"example": "Cans of Beer"})
quantity: int = Field(..., json_schema_extra={"example": 100})
unit_price: float = Field(..., json_schema_extra={"example": 12.50})说明:
BaseModel:添加验证的Pydantic基类Field(...):必填字段(省略号表示必填)json_schema_extra:API文档示例- 键入提示(
str,int,float):自动类型验证
步骤4.2:REST API的请求模型
为什么? API请求的单独模型(不同于内部产品模型)。
添加到main.py:
class NewProductRequest(BaseModel):
"""Request model for creating a new product (used by REST API)."""
name: str = Field(..., json_schema_extra={"example": "Coffee Mugs (Black)"})
initial_quantity: int = Field(..., json_schema_extra={"example": 25})
unit_price: float = Field(..., json_schema_extra={"example": 8.00})
class AdjustmentRequest(BaseModel):
"""Request model for stock adjustments (used by REST API)."""
product_name: str = Field(..., description="The product name to adjust.")
quantity_change: int = Field(..., description="Positive to add stock, negative to remove stock.")为什么要分开模型:
- API用户不提供
product_id(我们生成它) - 更清洁的API设计-只要求需要什么
______________________________________________________________________
阶段5:数据持久层
步骤5.1:文件路径配置
为什么? 将库存数据存储在与脚本位于同一目录的JSON文件中。
添加到main.py:
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
INVENTORY_FILE = os.path.join(SCRIPT_DIR, "inventory.json")
INVENTORY_DB: Dict[str, Product] = {} # In-memory database: product_id -> Product说明:
os.path.abspath(__file__):获取main.py的绝对路径os.path.dirname():获取包含main.py的目录os.path.join():创建investory.json的路径(适用于所有操作系统)INVENTORY_DB:将产品存储在内存中的词典(快速访问)
步骤5.2:加载库存功能
为什么? 服务器启动时从JSON文件读取保存的数据。
添加到main.py:
def load_inventory():
"""Loads inventory data from JSON file into memory."""
global INVENTORY_DB
if os.path.exists(INVENTORY_FILE):
try:
with open(INVENTORY_FILE, 'r') as f:
data = json.load(f)
# Convert JSON dict to Product objects
INVENTORY_DB = {k: Product(**v) for k, v in data.items()}
except json.JSONDecodeError:
# If file is corrupted, start fresh
INVENTORY_DB = {}
else:
# File doesn't exist yet - will be created on first save
pass说明:
global INVENTORY_DB:修改全局变量os.path.exists():检查文件是否存在json.load():解析JSON文件Product(**v):将dict转换为Product对象(Pydantic验证)try/except:优雅地处理损坏的文件
步骤5.3:保存库存功能
为什么? 每次修改后将更改保存到磁盘。
添加到main.py:
def save_inventory():
"""Persists current inventory state to JSON file."""
data_to_save = {k: v.model_dump() for k, v in INVENTORY_DB.items()}
with open(INVENTORY_FILE, 'w') as f:
json.dump(data_to_save, f, indent=2)说明:
model_dump():将Pydantic模型转换为字典json.dump():将JSON写入文件indent=2:格式美观,便于阅读
步骤5.4:启动时加载
为什么? 服务器启动时加载现有数据。
添加到main.py:
load_inventory() # Load inventory data when the script starts步骤5.5:模糊搜索功能
为什么? 允许部分产品名称匹配(用户友好)。
添加到main.py:
def fuzzy_match_product(query: str) -> List[Product]:
"""Performs case-insensitive partial name matching to find products."""
if not query:
return list(INVENTORY_DB.values())
query_lower = query.lower()
matches = [
product for product in INVENTORY_DB.values()
if query_lower in product.name.lower()
]
return matches说明:
-> List[Product]:返回类型提示- 列表理解:筛选查询与名称匹配的产品
- 不区分大小写:将两者都转换为小写进行比较
______________________________________________________________________
第6阶段:MCP服务器设置
步骤6.1:初始化FastMCP服务器
为什么? 创建Claude Desktop连接到的MCP服务器。
添加到main.py:
mcp = FastMCP(
"Inventory Manager (Explicit Tools)", # Server name shown in Claude Desktop
json_response=True # Use JSON format for responses
)说明:
FastMCP:处理MCP协议通信的框架- 服务器名称:出现在Claude Desktop的MCP服务器列表中
json_response=True:使用JSON格式(MCP标准)
______________________________________________________________________
第7阶段:MCP工具(CRUD操作)
步骤7.1:读取工具-获取库存状态
为什么? 允许克劳德查询库存。
添加到main.py:
@mcp.tool()
async def get_inventory_status(
product_name: Optional[str] = Field(None, description="The name or partial name of the product to search for."),
) -> List[Product]:
"""READ operation: Retrieves inventory status for all products or a specific product."""
matches = fuzzy_match_product(product_name)
if not matches and product_name:
raise ValueError(f"No products found matching '{product_name}'.")
return matches说明:
@mcp.tool():注册的装饰器用作MCP工具async def:异步功能(MCP要求)Optional[str]:参数可以为None(获取所有产品)Field():为Claude添加说明,以便其理解该工具raise ValueError:错误处理(MCP转换为正确的错误响应)
步骤7.2:创建工具-添加新产品
为什么? 允许克劳德将产品添加到库存中。
添加到main.py:
@mcp.tool()
async def add_new_product(
name: str = Field(..., description="The name of the product to add."),
initial_quantity: int = Field(..., description="The initial stock quantity."),
unit_price: float = Field(..., description="The price per unit."),
) -> Product:
"""CREATE operation: Adds a new product to the inventory."""
# Generate unique product ID using UUID
product_id = "P-" + str(uuid.uuid4()).split('-')[0].upper()
product = Product(
product_id=product_id,
name=name,
quantity=initial_quantity,
unit_price=unit_price
)
INVENTORY_DB[product_id] = product
save_inventory() # Persist to disk immediately
return product说明:
uuid.uuid4():生成唯一ID.split('-')[0]:获取UUID的第一段.upper():转换为大写(例如,“P-80562C3C”)save_inventory():立即写入磁盘
步骤7.3:更新工具-调整库存
为什么? 允许克劳德修改库存数量。
添加到main.py:
@mcp.tool()
async def adjust_stock_quantity(
product_name: str = Field(..., description="The name of the product to adjust."),
quantity_change: int = Field(..., description="Positive number to increase stock, negative number to decrease stock."),
) -> Product:
"""UPDATE operation: Adjusts the stock quantity of an existing product."""
matches = fuzzy_match_product(product_name)
if not matches:
raise ValueError(f"Product not found: '{product_name}'. Cannot adjust stock.")
# Prevent ambiguity - require unique match
if len(matches) > 1:
names = [m.name for m in matches]
raise ValueError(f"Ambiguous product name: '{product_name}' matched multiple items: {names}. Please clarify.")
product_to_adjust = matches[0]
original_id = product_to_adjust.product_id
original_name = product_to_adjust.name
new_quantity = product_to_adjust.quantity + quantity_change
# Business rule: prevent negative stock
if new_quantity 1:
names = [m.name for m in matches]
raise ValueError(f"Ambiguous product name: '{product_name}' matched multiple items: {names}. Please clarify.")
product_to_remove = matches[0]
original_id = product_to_remove.product_id
del INVENTORY_DB[original_id]
save_inventory()
return {"status": "success", "message": f"Product '{product_name}' (ID: {original_id}) has been removed from inventory."}说明:
- 与UPDATE工具相同的歧义检查
del INVENTORY_DB[original_id]:从词典中删除- 返回成功消息进行确认
______________________________________________________________________
阶段8:REST API服务器设置
步骤8.1:初始化FastAPI应用程序
为什么? 为HTTP访问创建REST API端点(替代MCP)。
添加到main.py:
app = FastAPI(
title="Inventory Manager API",
description="REST API for managing inventory with full CRUD operations",
version="1.0.0",
docs_url="/docs", # Swagger UI documentation
redoc_url="/redoc" # ReDoc documentation
)说明:
FastAPI():主应用程序实例docs_url:上的自动生成API文档http://localhost:8000/docsredoc_url:替代文件http://localhost:8000/redoc
步骤8.2:可选安全配置
为什么? 添加API密钥身份验证(可选,如果不需要,可以删除)。
添加到main.py:
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True)
SECRET_API_KEY = os.environ.get("MCP_API_KEY", "super-secret-mcp-key")
def get_api_key(api_key: str = Security(api_key_header)):
"""Validates the API key from request headers."""
if api_key == SECRET_API_KEY:
return api_key
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing API Key",
)注: 这目前未在端点中使用,但可供将来使用。
______________________________________________________________________
第9阶段:REST API端点
步骤9.1:获取-列出所有产品或搜索
为什么? 通过HTTP GET请求检索产品。
添加到main.py:
@app.get("/api/products",
response_model=List[Product],
summary="Get all products or search by name",
tags=["Products"])
async def get_products(
name: Optional[str] = Query(None, description="Filter products by name (fuzzy match)")
):
"""Retrieve all products or search for products by name."""
load_inventory() # Reload from disk to sync with MCP changes
matches = fuzzy_match_product(name)
if not matches and name:
raise HTTPException(
status_code=404,
detail=f"No products found matching '{name}'."
)
return matches说明:
@app.get():HTTP GET端点装饰器Query():从URL提取查询参数(?name=beer)load_inventory():与任何MCP更改同步HTTPException:返回正确的HTTP错误代码
步骤9.2:获取-按ID获取产品
为什么? 使用精确的ID检索特定产品。
添加到main.py:
@app.get("/api/products/{product_id}",
response_model=Product,
summary="Get product by ID",
tags=["Products"])
async def get_product_by_id(product_id: str = Path(..., description="Product ID")):
"""Retrieve a specific product by its unique product ID."""
load_inventory()
if product_id not in INVENTORY_DB:
raise HTTPException(
status_code=404,
detail=f"Product with ID '{product_id}' not found."
)
return INVENTORY_DB[product_id]说明:
{product_id}:URL中的路径参数(/api/products/P-001)Path(...):必需的路径参数
步骤9.3:POST-创建新产品
为什么? 通过HTTP POST请求添加产品。
添加到main.py:
@app.post("/api/products",
response_model=Product,
status_code=status.HTTP_201_CREATED,
summary="Add a new product",
tags=["Products"])
async def create_product(product: NewProductRequest):
"""CREATE operation: Add a new product to the inventory."""
product_id = "P-" + str(uuid.uuid4()).split('-')[0].upper()
new_product = Product(
product_id=product_id,
name=product.name,
quantity=product.initial_quantity,
unit_price=product.unit_price
)
INVENTORY_DB[product_id] = new_product
save_inventory()
return new_product说明:
@app.post():HTTP POST端点status_code=201:已创建状态代码product: NewProductRequest:请求正文由Pydantic自动验证
步骤9.4:PATCH-调整库存数量
为什么? 通过HTTP PATCH请求更新库存。
添加到main.py:
@app.patch("/api/products/{product_name}/stock",
response_model=Product,
summary="Adjust product stock quantity",
tags=["Products"])
async def adjust_stock(
product_name: str = Path(..., description="Product name to adjust"),
quantity_change: int = Query(..., description="Positive to increase, negative to decrease")
):
"""UPDATE operation: Adjust the stock quantity of a product."""
load_inventory()
matches = fuzzy_match_product(product_name)
if not matches:
raise HTTPException(
status_code=404,
detail=f"Product not found: '{product_name}'. Cannot adjust stock."
)
if len(matches) > 1:
names = [m.name for m in matches]
raise HTTPException(
status_code=400,
detail=f"Ambiguous product name: '{product_name}' matched multiple items: {names}. Please clarify."
)
product_to_adjust = matches[0]
original_id = product_to_adjust.product_id
original_name = product_to_adjust.name
new_quantity = product_to_adjust.quantity + quantity_change
if new_quantity 1:
names = [m.name for m in matches]
raise HTTPException(
status_code=400,
detail=f"Ambiguous product name: '{product_name}' matched multiple items: {names}. Please clarify."
)
product_to_remove = matches[0]
original_id = product_to_remove.product_id
del INVENTORY_DB[original_id]
save_inventory()
return None # 204 No Content - no response body说明:
status_code=204:无内容(DELETE标准)- 退货
None(无响应体)
步骤9.6:GET-健康检查端点
为什么? 验证API是否正在运行(有助于监控)。
添加到main.py:
@app.get("/api/health",
summary="Health check endpoint",
tags=["Health"])
async def health_check():
"""Health check endpoint to verify the API is running."""
return {
"status": "healthy",
"total_products": len(INVENTORY_DB)
}______________________________________________________________________
第10阶段:服务器执行
步骤10.1:主执行块
为什么? 允许脚本在两种模式下运行(MCP或HTTP)。
添加到main.py:
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "http":
# HTTP REST API mode
print("Starting Inventory Manager REST API server...")
print("Swagger docs available at: http://localhost:8000/docs")
print("API available at: http://localhost:8000/api")
uvicorn.run(app, host="0.0.0.0", port=8000)
else:
# MCP mode (default): Run as stdio server for Claude Desktop
mcp.run(transport="stdio")说明:
sys.argv:命令行参数uvicorn.run():启动FastAPI服务器mcp.run(transport="stdio"):启动MCP服务器(通过stdin/stdout通信)
______________________________________________________________________
第11阶段:测试和运行
步骤11.1:为MCP配置Claude桌面
为什么? 将您的MCP服务器连接到Claude Desktop,以便Claude可以使用您的库存工具。
行动:
- 找到Claude桌面配置文件:
配置文件的位置取决于您的操作系统:
- 窗户: %APPDATA%\Claude\claude_desktop_config.json - 完整路径: C:\Users\YOUR_USERNAME\AppData\Roaming\Claude\claude_desktop_config.json - macOS: ~/Library/Application Support/Claude/claude_desktop_config.json - Linux: ~/.config/Claude/claude_desktop_config.json
- 打开配置文件:
- 如果文件不存在,请将其创建为新的JSON文件 - 使用文本编辑器(VS Code、Notepad++等)
- 添加您的MCP服务器配置:
将以下JSON结构添加到配置文件中。用实际路径替换路径 main.py 位置:
对于Windows(使用Python系统):
{
"mcpServers": {
"inventory-manager": {
"command": "python",
"args": [
"C:\\Users\\YOUR_USERNAME\\path\\to\\inventory-mcp\\main.py"
],
"env": {
"MCP_API_KEY": "super-secret-mcp-key"
}
}
}
}对于Windows(使用uv的Python):
{
"mcpServers": {
"inventory-manager": {
"command": ".venv\\Scripts\\python.exe",
"args": [
"main.py"
],
"cwd": "C:\\Users\\YOUR_USERNAME\\path\\to\\inventory-mcp",
"env": {
"MCP_API_KEY": "super-secret-mcp-key"
}
}
}
}对于macOS/Linux(使用Python系统):
{
"mcpServers": {
"inventory-manager": {
"command": "python",
"args": [
"/full/path/to/inventory-mcp/main.py"
],
"env": {
"MCP_API_KEY": "super-secret-mcp-key"
}
}
}
}对于macOS/Linux(使用uv的Python):
{
"mcpServers": {
"inventory-manager": {
"command": ".venv/bin/python",
"args": [
"main.py"
],
"cwd": "/full/path/to/inventory-mcp",
"env": {
"MCP_API_KEY": "super-secret-mcp-key"
}
}
}
}重要提示:
- 替换 YOUR_USERNAME 以及符合您实际值的路径 - 使用双反睫毛(\\)在Windows路径中 - 如果您已经配置了其他MCP服务器,请添加 "inventory-manager" 到现有 mcpServers 对象(不要替换它) - 这 env 部分是可选的,但如果您想使用API关键功能,则它很有用
- 保存并重新启动Claude Desktop:
- 保存配置文件 - 完全关闭并重新启动Claude Desktop - MCP服务器现在应该可用
步骤11.2:测试MCP模式
为什么? 验证MCP服务器是否与Claude Desktop兼容。
行动:
- 验证连接:
- 打开克劳德桌面 - 库存管理工具应出现在Claude的可用工具中 - 试着问克劳德:“库存中有什么产品?”或“添加一个名为“测试项目”的新产品,数量为10,价格为5.99”
- 手动运行MCP服务器(用于测试):
python main.py- 服务器在stdio模式下运行(等待输入) - 配置后,Claude Desktop会自动连接
步骤11.2:测试REST API模式
为什么? 验证REST API端点是否工作。
行动:
- 启动HTTP服务器:
python main.py http- 打开浏览器:
- 访问:http://localhost:8000/docs - 交互式API文档(Swagger UI)
- 测试终点:
- 在任何端点上单击“试用” - 输入参数 - 点击“执行”
步骤11.3:使用curl(命令行)进行测试
为什么? 从终端验证API是否有效。
示例:
# Get all products
curl http://localhost:8000/api/products
# Search for product
curl "http://localhost:8000/api/products?name=beer"
# Create product
curl -X POST http://localhost:8000/api/products \
-H "Content-Type: application/json" \
-d '{"name": "Test Product", "initial_quantity": 10, "unit_price": 5.99}'
# Adjust stock
curl -X PATCH "http://localhost:8000/api/products/Test%20Product/stock?quantity_change=-2"
# Delete product
curl -X DELETE "http://localhost:8000/api/products/Test%20Product"______________________________________________________________________
第12阶段:项目结构总结
最终文件结构
inventory-mcp/
├── main.py # Main server code (all code above)
├── pyproject.toml # Dependencies and project config
├── uv.lock # Locked dependency versions
├── inventory.json # Data file (created automatically)
├── README.md # Project documentation
└── .venv/ # Virtual environment (created by uv)关键文件说明
- main.py:完成服务器实施(~538行)
- pyproject.toml:依赖关系管理
- investory.json:持久数据存储
- uv.lock:确保可复制的构建
______________________________________________________________________
第13阶段:下一步和改进
潜在改进
- 数据库集成:
- 用SQLite/PPostgreSQL替换JSON文件 - 使用SQLAlchemy ORM
- 身份验证:
- 实现JWT令牌 - 添加用户管理
- 高级功能:
- 产品类别 - 库存历史/审计日志 - 低库存警报 - 批量操作
- 测试:
- 使用pytest添加单元测试 - 添加集成测试 - 测试MCP工具
- 部署:
- Docker容器化 - 部署到云(AWS、GCP、Azure) - 添加CI/CD管道
______________________________________________________________________
故障排除
常见问题
- 导入错误:
- 确保虚拟环境已激活 - 跑 uv sync 安装依赖项
- 端口已在使用中:
- 更改端口 uvicorn.run(app, port=8001) - 或使用端口8000终止进程
- 文件权限错误:
- 检查investory.json的写入权限 - 确保脚本目录可写
- MCP连接问题:
- 验证克劳德桌面配置 - 检查stdio传输是否正确
______________________________________________________________________
结论
您现在拥有了一个完整的、可用于生产的、具有REST API功能的MCP服务器!服务器支持:
- ✅ 完整的CRUD操作(创建、读取、更新、删除)
- ✅ Claude Desktop的MCP集成
- ✅ 带有自动文档的REST API
- ✅ 持久数据存储
- ✅ 模糊产品搜索
- ✅ 输入验证和错误处理
下一页: 开始构建自己的增强功能和定制!
