hankin
423baff73b
- Docker bridge 网络隔离（8000 端口封死） - Gunicorn 4 Worker 多进程 - Alembic 数据库迁移基线 - 日志轮转 20m×3 - JWT 密钥 + DB 密码 + CORS 收紧 - 3-2-1 备份链路（NAS + R740-B 冷备） - 连接池 pool_pre_ping + pool_recycle=3600
198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
"""
|
|
MCP 工具注册 — 首批业务工具
|
|
每个工具函数签名统一为: async def tool_fn(db, user, params) -> MCPToolResult
|
|
"""
|
|
from __future__ import annotations
|
|
import uuid
|
|
from typing import Any
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.schemas.auth import CurrentUserPayload
|
|
from app.mcp.registry import MCPToolResult, register_tool
|
|
from app.services import customer_service, order_service
|
|
|
|
|
|
@register_tool(
|
|
name="search_customers",
|
|
description="搜索客户列表，支持按名称模糊搜索和等级过滤",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"keyword": {"type": "string", "description": "客户名称关键词"},
|
|
"level": {"type": "string", "enum": ["A", "B", "C"], "description": "客户等级"},
|
|
"page": {"type": "integer", "default": 1},
|
|
"size": {"type": "integer", "default": 10},
|
|
},
|
|
},
|
|
)
|
|
async def search_customers(
|
|
db: AsyncSession, user: CurrentUserPayload, params: dict[str, Any],
|
|
) -> MCPToolResult:
|
|
result = await customer_service.list_customers(
|
|
db, user,
|
|
page=params.get("page", 1),
|
|
size=params.get("size", 10),
|
|
keyword=params.get("keyword"),
|
|
level=params.get("level"),
|
|
)
|
|
return MCPToolResult(
|
|
success=True, response_type="text",
|
|
data=result.model_dump(mode="json"),
|
|
message=f"找到 {result.total} 个客户",
|
|
)
|
|
|
|
|
|
@register_tool(
|
|
name="create_customer",
|
|
description="创建新客户（返回确认卡片，需用户确认后执行）",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string", "description": "客户名称"},
|
|
"level": {"type": "string", "enum": ["A", "B", "C"]},
|
|
"contact": {"type": "string", "description": "联系人"},
|
|
"phone": {"type": "string", "description": "电话"},
|
|
},
|
|
"required": ["name"],
|
|
},
|
|
)
|
|
async def create_customer_tool(
|
|
db: AsyncSession, user: CurrentUserPayload, params: dict[str, Any],
|
|
) -> MCPToolResult:
|
|
# 写操作 → 返回 action_card，由前端确认后再真正执行
|
|
return MCPToolResult(
|
|
success=True, response_type="action_card",
|
|
data={
|
|
"card_type": "create_customer",
|
|
"title": "新建客户确认",
|
|
"summary": f"即将创建客户: {params.get('name', '未知')}",
|
|
"fields": [
|
|
{"label": "客户名称", "value": params.get("name", ""), "editable": True},
|
|
{"label": "客户等级", "value": params.get("level", "C"), "editable": True},
|
|
{"label": "联系人", "value": params.get("contact", ""), "editable": True},
|
|
{"label": "电话", "value": params.get("phone", ""), "editable": True},
|
|
],
|
|
"actions": [
|
|
{"key": "confirm", "label": "确认创建", "style": "primary"},
|
|
{"key": "cancel", "label": "取消", "style": "default"},
|
|
],
|
|
"params": params, # 原始参数，回调时用
|
|
},
|
|
message="请确认以下客户信息",
|
|
)
|
|
|
|
|
|
@register_tool(
|
|
name="calculate_price",
|
|
description="查询客户专属报价（历史成交价追溯 → 标准价兜底）",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"customer_id": {"type": "string", "description": "客户 UUID"},
|
|
"sku_id": {"type": "string", "description": "产品 SKU UUID"},
|
|
},
|
|
"required": ["customer_id", "sku_id"],
|
|
},
|
|
)
|
|
async def calculate_price_tool(
|
|
db: AsyncSession, user: CurrentUserPayload, params: dict[str, Any],
|
|
) -> MCPToolResult:
|
|
result = await order_service.calculate_price(
|
|
db,
|
|
customer_id=uuid.UUID(params["customer_id"]),
|
|
sku_id=uuid.UUID(params["sku_id"]),
|
|
)
|
|
return MCPToolResult(
|
|
success=True, response_type="text",
|
|
data=result.model_dump(mode="json"),
|
|
message=f"SKU {result.sku_code} 报价: ¥{result.unit_price} (来源: {result.price_source})",
|
|
)
|
|
|
|
|
|
@register_tool(
|
|
name="create_order",
|
|
description="创建销售订单（返回确认卡片，需用户确认后执行）",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"customer_id": {"type": "string", "description": "客户 UUID"},
|
|
"items": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"sku_id": {"type": "string"},
|
|
"qty": {"type": "number"},
|
|
"unit_price": {"type": "number"},
|
|
},
|
|
},
|
|
},
|
|
"remark": {"type": "string"},
|
|
},
|
|
"required": ["customer_id", "items"],
|
|
},
|
|
)
|
|
async def create_order_tool(
|
|
db: AsyncSession, user: CurrentUserPayload, params: dict[str, Any],
|
|
) -> MCPToolResult:
|
|
items = params.get("items", [])
|
|
total = sum(i.get("qty", 0) * i.get("unit_price", 0) for i in items)
|
|
return MCPToolResult(
|
|
success=True, response_type="action_card",
|
|
data={
|
|
"card_type": "create_order",
|
|
"title": "创建订单确认",
|
|
"summary": f"共 {len(items)} 项商品，总金额 ¥{total:.2f}",
|
|
"fields": [
|
|
{"label": "客户ID", "value": params.get("customer_id", ""), "editable": False},
|
|
{"label": "商品数", "value": str(len(items)), "editable": False},
|
|
{"label": "总金额", "value": f"¥{total:.2f}", "editable": False},
|
|
{"label": "备注", "value": params.get("remark", ""), "editable": True},
|
|
],
|
|
"actions": [
|
|
{"key": "confirm", "label": "确认建单", "style": "primary"},
|
|
{"key": "cancel", "label": "取消", "style": "default"},
|
|
],
|
|
"params": params,
|
|
},
|
|
message="请确认订单信息",
|
|
)
|
|
|
|
|
|
@register_tool(
|
|
name="search_orders",
|
|
description="搜索订单列表，支持按客户名称、订单号、发货/付款状态筛选",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"keyword": {"type": "string", "description": "客户名称关键词"},
|
|
"order_no": {"type": "string", "description": "订单号模糊搜索"},
|
|
"shipping_state": {
|
|
"type": "string",
|
|
"enum": ["pending", "partial", "shipped"],
|
|
"description": "发货状态",
|
|
},
|
|
"payment_state": {
|
|
"type": "string",
|
|
"enum": ["unpaid", "partial", "cleared"],
|
|
"description": "付款状态",
|
|
},
|
|
"page": {"type": "integer", "default": 1},
|
|
"size": {"type": "integer", "default": 10},
|
|
},
|
|
},
|
|
)
|
|
async def search_orders_tool(
|
|
db: AsyncSession, user: CurrentUserPayload, params: dict[str, Any],
|
|
) -> MCPToolResult:
|
|
result = await order_service.list_orders(
|
|
db, user,
|
|
page=params.get("page", 1),
|
|
size=params.get("size", 10),
|
|
keyword=params.get("keyword"),
|
|
)
|
|
return MCPToolResult(
|
|
success=True, response_type="text",
|
|
data=result.model_dump(mode="json"),
|
|
message=f"找到 {result.total} 个订单",
|
|
)
|