┌─────────────┐
│   Claude    │  <-- AI 助手
└──────┬──────┘
       │ 问题："我们有哪些产品？"
       │
       ▼
┌─────────────────────────┐
│   MCP 协议              │
│   (标准通信方式)         │
└──────┬──────────────────┘
       │ 调用工具：list_products()
       │
       ▼
┌─────────────────────────┐
│   MCP Server            │ <-- 你的服务器
│  (提供 Tool 列表)        │
└──────┬──────────────────┘
       │ 返回：全部产品列表
       │
       ▼
┌─────────────────────────┐
│   本地数据源            │
│  (CSV / DB / API)       │
└─────────────────────────┘

Sales Manager → 打开 Excel → CTRL+F 查找 → 手动汇总 → 花 30 分钟

Sales Manager: "过去一周销售最好的产品是什么？"
Claude 直接查询数据库 → 一秒得到答案

# 检查 Python 版本（需要 3.11+）
python --version

# 安装依赖
pip install mcp anthropic pydantic python-dotenv

product_id,product_name,category,price,stock,created_date
P001,Claude Opus 4.6,AI Model,2.99,5000,2026-03-15
P002,Claude Sonnet 4.6,AI Model,1.99,8000,2026-03-15
P003,Claude Haiku 4.5,AI Model,0.80,10000,2026-02-20
P004,API Gateway Pro,Infrastructure,29.99,100,2026-04-01
P005,ClaudeAPI Console,Tools,0.00,999999,2026-01-01
P006,MCP Integration Kit,Tools,99.99,50,2026-04-10
P007,Premium Support,Service,199.99,200,2026-03-01
P008,Batch API Access,Feature,49.99,300,2026-04-01

# 工具1：列出所有产品
{
    "name": "list_products",
    "description": "List all products in the database",
    "inputSchema": {"type": "object", "properties": {}, "required": []}
}

# 工具2：按 ID 查询
{
    "name": "query_product",
    "description": "Get details of a specific product by ID",
    "inputSchema": {
        "type": "object",
        "properties": {
            "product_id": {"type": "string", "description": "Product ID (e.g., P001)"}
        },
        "required": ["product_id"]
    }
}

# 工具3：按价格范围搜索
{
    "name": "search_price_range",
    "inputSchema": {
        "type": "object",
        "properties": {
            "min_price": {"type": "number"},
            "max_price": {"type": "number"}
        },
        "required": ["min_price", "max_price"]
    }
}

# 工具4：按类别筛选 / 工具5：获取统计信息（同理）

import csv, json
from pathlib import Path

products_data = []

def load_csv():
    global products_data
    with open("products.csv", "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        products_data = [row for row in reader]

def list_all_products():
    result = "All Products:\n"
    for idx, p in enumerate(products_data, 1):
        result += f"{idx}. {p['product_name']} (ID: {p['product_id']})\n"
    return result

def query_product_by_id(product_id):
    for p in products_data:
        if p['product_id'].lower() == product_id.lower():
            return json.dumps(p, indent=2)
    return f"Product {product_id} not found"

def search_by_price_range(min_price, max_price):
    results = [p for p in products_data
               if min_price <= float(p['price']) <= max_price]
    return f"Found {len(results)} products:\n" + \
           "\n".join(f"  • {p['product_name']} - CNY{p['price']}" for p in results)

def filter_by_category(category):
    results = [p for p in products_data
               if p['category'].lower() == category.lower()]
    return f"Found {len(results)} products in {category}"

def get_stats():
    total = len(products_data)
    categories = set(p['category'] for p in products_data)
    avg_price = sum(float(p['price']) for p in products_data) / total
    return f"Total: {total} | Categories: {len(categories)} | Avg Price: CNY{avg_price:.2f}"

from mcp.server import Server
from mcp.types import Tool
import mcp.types as types

server = Server("mcp-csv-server")

@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(name="list_products", description="List all products",
             inputSchema={"type": "object", "properties": {}, "required": []}),
        # ... 其他 4 个工具
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "list_products":
        result = list_all_products()
    elif name == "query_product":
        result = query_product_by_id(arguments["product_id"])
    elif name == "search_price_range":
        result = search_by_price_range(arguments["min_price"], arguments["max_price"])
    elif name == "filter_category":
        result = filter_by_category(arguments["category"])
    elif name == "get_stats":
        result = get_stats()
    else:
        result = f"Unknown tool: {name}"
    return [types.TextContent(type="text", text=result)]

def main():
    load_csv()
    server.run_stdio()

if __name__ == "__main__":
    main()

Claude → 调用 search_price_range(min_price=0, max_price=50)
MCP Server → 返回 6 个结果：
  • Claude Opus 4.6 - CNY2.99
  • Claude Sonnet 4.6 - CNY1.99
  • Claude Haiku 4.5 - CNY0.80
  • API Gateway Pro - CNY29.99
  • ClaudeAPI Console - CNY0.00
  • Batch API Access - CNY49.99

{
  "product_id": "P002",
  "product_name": "Claude Sonnet 4.6",
  "category": "AI Model",
  "price": "1.99",
  "stock": "8000",
  "created_date": "2026-03-15"
}

# 启动 MCP Server
import subprocess
server = subprocess.Popen(["python", "server.py"])

# Claude 自动检测到可用工具，用户提问时自动调用
user_input = "Show me all Claude models under 2 dollars"
# Claude → 调用 search_price_range(0, 2) → 返回结果

# .env
CSV_FILE=./products.csv
MCP_PORT=8000
DEBUG=false

from dotenv import load_dotenv
import os

load_dotenv()
csv_file = os.getenv("CSV_FILE", "products.csv")

import sqlite3
conn = sqlite3.connect("products.db")
cursor = conn.cursor()

def search_by_price_range(min_price, max_price):
    cursor.execute(
        "SELECT * FROM products WHERE price BETWEEN ? AND ?",
        (min_price, max_price)
    )
    return cursor.fetchall()

@server.call_tool()
async def handle_call_tool(name, args):
    if name.startswith("write_") or name.startswith("delete_"):
        if not verify_user_permission(user_id):
            return "Permission denied"
    return execute_tool(name, args)

from functools import lru_cache

@lru_cache(maxsize=100)
def get_products_by_category(category):
    return filter_category(category)

def list_products(page=1, page_size=20):
    start = (page - 1) * page_size
    return products_data[start:start+page_size]