inventory-mcp

用于库存查询的演示 MCP 服务器，使用 Python 和 FastMCP 开发。该项目帮助学习 Model Context Protocol (MCP) 的核心概念，并在传输、MCP 接口、业务规则、验证和数据之间进行了分离。

当前范围有意为只读：服务器允许查询产品和库存数量，不提供新增、修改或删除操作。

技术

• Python 3.11+

• FastMCP

• Pydantic

• pytest

• Ruff

Related MCP server: vanam-erp-mcp

架构

• app/server.py: 创建 FastMCP 服务器，注册工具，并启动 stdio 或 SSE 传输。

• app/client.py: 演示客户端，通过 stdio 或 SSE 列出并调用工具。

• app/tools/: MCP 接口；验证输入，委托给服务层，并将预期错误转换为稳定的响应。

• app/services/: 查询规则和库存加载。

• app/schemas/: Pydantic 模型，定义并验证产品和库存契约。

• app/data/: 本地数据源，当前为 inventory.json 文件。

• tests/: 针对服务、工具和服务器配置的自动化测试。

Client → MCP Server → Tool → InventoryService → inventory.json

工具不直接访问文件。它们将业务规则委托给 InventoryService。

MCP 工具

get_product

• 目的： 按名称查询产品的完整数据。

• 输入： name（非空 string）。

• 成功时的输出： 包含 name、quantity 和 price 的对象。

• 产品不存在时的输出： 包含 error: "product_not_found" 和描述性 message 的对象。

• MCP 描述： Use this tool to retrieve the complete data of a product by name, including its price and stock quantity.

• 分类： 只读。

{
  "name": "Mouse",
  "quantity": 25,
  "price": 89.9
}

get_stock

• 目的： 按名称仅查询产品的当前数量。

• 输入： name（非空 string）。

• 成功时的输出： 包含 quantity 的对象。

• 产品不存在时的输出： 包含 error: "product_not_found" 和描述性 message 的对象。

• MCP 描述： Use this tool to retrieve only the current stock quantity of a product by name.

• 分类： 只读。

{
  "quantity": 25
}

输入验证

工具要求 name 为包含内容的字符串。空名称或仅由空格组成的名称会在查询前被拒绝。服务使用 strip() 去除两端空格，并使用 casefold() 在不区分大小写的情况下比较名称。

Pydantic 会验证从 JSON 加载的记录和输出模型。产品必须具有非空名称、非负整数量和非负数值价格。空查询名称的拒绝由 _validate_product_name() 完成。无效记录会以显式错误中断加载。

错误处理

InventoryService 在找不到所请求的产品时抛出 ProductNotFoundError。工具捕获这一预期错误，并返回可预测的负载：

{
  "error": "product_not_found",
  "message": "Product not found: Monitor"
}

输入错误（如空名称或非字符串值）不会被隐藏：它们会作为工具调用的错误被报告。

MCP 传输

• stdio： 通过标准输入和输出通信。在本项目中，客户端将 FastMCP 服务器作为子进程启动，执行调用，并在结束时终止进程。

• SSE： 通过带有 Server-Sent Events 的 HTTP 端点通信。服务器和客户端运行在独立进程中；默认情况下，服务器监听 http://127.0.0.1:8000/sse。

如何运行

以下命令使用 PowerShell，并应在项目根目录中执行。

创建并激活虚拟环境

python -m venv .venv
.\.venv\Scripts\Activate.ps1

安装依赖

python -m pip install --upgrade pip
python -m pip install -e ".[dev]"

通过 stdio 运行

客户端默认使用 stdio，并将服务器作为子进程启动：

.\.venv\Scripts\python.exe -m app.client

要仅直接启动服务器：

.\.venv\Scripts\python.exe -m app.server --transport stdio

通过 SSE 运行

在一个终端中启动服务器（sse 是服务器的默认传输方式）：

.\.venv\Scripts\python.exe -m app.server

等效的显式命令是 python -m app.server --transport sse。在另一个终端中，连接客户端：

.\.venv\Scripts\python.exe -m app.client --transport sse

客户端可以通过 --url 接受其他端点。

运行测试

.\.venv\Scripts\pytest.exe

运行 Ruff

.\.venv\Scripts\ruff.exe check .
.\.venv\Scripts\ruff.exe format --check .

工具风险评估

当前的工具是只读的，不能创建、修改或删除数据。这一决定降低了风险面，但并不能消除对机密性和可用性的潜在影响。

大量调用仍可能消耗服务器资源。未来对工具或返回数据的更改应伴随新的风险评估。

信任边界

从 MCP 客户端接收的参数被视为不可信输入。

MCP Client
    ↓
MCP Server
    ↓
Tool
    ↓
InventoryService
    ↓
inventory.json

验证发生在参数被服务层使用之前。服务器不会因为数据是通过 MCP 协议到达的，就假定客户端发送的数据是有效的。inventory.json 中的记录同样被视为外部输入，并在加载时由 Pydantic 进行验证。

MCP 工具注解

工具根据其行为进行语义分类。当前的两个操作声明：

readOnlyHint=true
openWorldHint=false

readOnlyHint=true 向 MCP 客户端表明该操作不打算修改状态。

openWorldHint=false 表示该工具在一个封闭且已知的领域内工作——此处为本地库存——而不是查询外部系统或开放来源。

这些注解是面向 MCP 客户端的元数据和提示，而非安全机制。客户端不应将其视为验证、授权或其他实际控制的替代品。

写入工具的风险

未来的类似操作：

update_stock(name, quantity)

将具有显著更高的风险，因为它会修改系统的持久状态。

错误或恶意的调用可能会修改错误的产品、记录无效值或允许未经授权的更改。未来的 update_stock 之类的工具将需要严格的验证、身份验证、授权、审计和追踪。破坏性操作在适用时还需要确认或批准。

传输方式的风险

在 stdio 中，服务器作为客户端的子进程在本地启动，减少了网络暴露。在 SSE 中，服务器和客户端是独立进程，通信使用 HTTP 端点。如果将该端点发布到本地主机之外，则需要额外的访问和可用性控制。

测试

当前的测试套件验证：

• InventoryService 的加载、查找、规范化和错误；

• 工具的返回，以及将不存在的产品转换为可预测的错误；

• 拒绝空名称和非字符串值；

• Pydantic 拒绝无效的库存记录；

• 工具在服务器上的注册；

• SSE 和 stdio 传输的选择与配置；

• 通过 stdio 的真实集成，包括 list_tools()、调用 get_stock 和读取 MCP 注解。

场景包括存在和不存在的产品、两端空格、大小写差异和无效输入。在端到端测试中，真实的 FastMCP 客户端将服务器作为子进程启动，验证 readOnlyHint 和 openWorldHint，查询从本地 JSON 加载的库存，并通过上下文管理器关闭连接。

代码质量

项目使用类型提示，将 MCP、服务、模式和数据之间的职责分离，并保持最小依赖。pytest 覆盖已实现的行为，而 Ruff 检查 lint、导入、Python 3.11 兼容性和格式。

当前限制

• 数据从本地 JSON 文件加载。

• 没有数据库。

• 没有 AI 或 LLM 集成。

• 没有写入工具。

• 没有身份验证或授权。

可能的演进

• 追踪和结构化日志，目前保留在范围之外以保持项目的教学重点；

• 支持 Streamable HTTP；

• 数据库持久化；

• 身份验证和授权；

• 带有安全防护的写入工具；

• 未来与 LLM 集成。