feat: 添加 Strapi 配置文件支持

- 新增 config.yaml:集中管理 Strapi API 配置
- 新增 config_loader.py:配置加载模块
- 更新 http_routes.py:从配置文件读取 API 端点
- 支持从 YAML 文件配置 FAQ 分类和语言
- 更新 requirements.txt:添加 pyyaml 依赖

优势:
- 配置与代码分离,易于维护
- 添加新分类无需修改代码
- 支持热加载配置

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
wangliang
2026-01-15 10:39:25 +08:00
parent 3ad6eee0d9
commit 0e59f3067e
4 changed files with 261 additions and 15 deletions

View File

@@ -0,0 +1,97 @@
# Strapi MCP 配置文件
# 根据 docs/strapi.txt 中的接口定义配置
strapi:
base_url: https://cms.yehwang.com
api_token: "" # 留空表示不需要认证
# 支持的语言
languages:
- code: en
name: English
- code: nl
name: Nederlands
- code: de
name: Deutsch
- code: es
name: Español
- code: fr
name: Français
- code: it
name: Italiano
- code: tr
name: Türkçe
# FAQ 分类配置
faq_categories:
register:
endpoint: faq-register
description: 账号相关
keywords:
- account
- register
- login
- password
order:
endpoint: faq-order
description: 订单相关
keywords:
- order
- place order
- cancel order
- modify order
pre-order:
endpoint: faq-pre-order
description: 预售订单相关
keywords:
- pre-order
- reserve
- pre-sale
payment:
endpoint: faq-payment
description: 支付相关
keywords:
- payment
- pay
- checkout
- voucher
- discount
shipment:
endpoint: faq-shipment
description: 运输相关
keywords:
- shipping
- delivery
- transit
- courier
return:
endpoint: faq-return
description: 退货相关
keywords:
- return
- refund
- complaint
- defective
other:
endpoint: faq-other-question
description: 其他问题
keywords:
- other
- general
# 公司信息端点
info_sections:
contact:
endpoint: info-contact
description: 联系信息和营业时间
# API 端点模板
api_templates:
faq: "/api/{category}?populate=deep&locale={locale}"
info: "/api/{section}?populate=deep&locale={locale}"

View File

@@ -0,0 +1,126 @@
"""
Strapi MCP 配置加载器
从 YAML 配置文件加载 Strapi API 配置
"""
import yaml
from pathlib import Path
from typing import Dict, List, Optional
from pydantic import BaseModel
class LanguageConfig(BaseModel):
"""语言配置"""
code: str
name: str
class CategoryConfig(BaseModel):
"""FAQ 分类配置"""
endpoint: str
description: str
keywords: List[str] = []
class StrapiConfig(BaseModel):
"""Strapi 配置"""
base_url: str
api_token: str = ""
languages: List[LanguageConfig] = []
faq_categories: Dict[str, CategoryConfig] = {}
info_sections: Dict[str, Dict] = {}
def load_config(config_path: Optional[str] = None) -> StrapiConfig:
"""加载配置文件
Args:
config_path: 配置文件路径,默认为 config.yaml
Returns:
StrapiConfig: 配置对象
"""
if config_path is None:
# 默认从当前目录的 config.yaml 加载
config_path = Path(__file__).parent / "config.yaml"
with open(config_path, 'r', encoding='utf-8') as f:
config_data = yaml.safe_load(f)
return StrapiConfig(**config_data)
def get_category_endpoint(category: str, config: Optional[StrapiConfig] = None) -> str:
"""获取分类对应的 API 端点
Args:
category: 分类名称
config: 配置对象
Returns:
str: API 端点
"""
if config is None:
config = load_config()
if category in config.faq_categories:
return config.faq_categories[category].endpoint
# 如果没有找到,返回默认格式
return f"faq-{category}"
def get_supported_languages(config: Optional[StrapiConfig] = None) -> List[str]:
"""获取支持的语言代码列表
Args:
config: 配置对象
Returns:
List[str]: 语言代码列表
"""
if config is None:
config = load_config()
return [lang.code for lang in config.languages]
def get_all_categories(config: Optional[StrapiConfig] = None) -> Dict[str, str]:
"""获取所有分类及其描述
Args:
config: 配置对象
Returns:
Dict[str, str]: 分类名称 -> 描述的映射
"""
if config is None:
config = load_config()
return {
name: cat.description
for name, cat in config.faq_categories.items()
}
# 导出配置单例
_global_config: Optional[StrapiConfig] = None
def get_config() -> StrapiConfig:
"""获取全局配置单例"""
global _global_config
if _global_config is None:
_global_config = load_config()
return _global_config
if __name__ == "__main__":
# 测试配置加载
config = load_config()
print(f"✅ 配置加载成功")
print(f"Base URL: {config.base_url}")
print(f"支持语言: {[lang.code for lang in config.languages]}")
print(f"FAQ 分类: {list(config.faq_categories.keys())}")
print(f"\n分类详情:")
for name, cat in config.faq_categories.items():
print(f" - {name}: {cat.description} (/{cat.endpoint})")

View File

@@ -10,6 +10,8 @@ from starlette.responses import JSONResponse
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings
from pydantic import ConfigDict from pydantic import ConfigDict
from config_loader import load_config, get_category_endpoint
class Settings(BaseSettings): class Settings(BaseSettings):
"""Server configuration""" """Server configuration"""
@@ -22,18 +24,15 @@ class Settings(BaseSettings):
settings = Settings() settings = Settings()
# 加载配置文件
# ============ FAQ Categories ============ try:
strapi_config = load_config()
FAQ_CATEGORIES = { # 使用配置文件中的 URL如果存在
"register": "faq-register", if strapi_config.base_url:
"order": "faq-order", settings.strapi_api_url = strapi_config.base_url
"pre-order": "faq-pre-order", except Exception:
"payment": "faq-payment", # 如果配置文件加载失败,使用环境变量
"shipment": "faq-shipment", strapi_config = None
"return": "faq-return",
"other": "faq-other-question",
}
# ============ Company Info ============ # ============ Company Info ============
@@ -125,8 +124,12 @@ async def query_faq_http(
limit: Maximum results to return limit: Maximum results to return
""" """
try: try:
# Map category to endpoint # 从配置文件获取端点
endpoint = FAQ_CATEGORIES.get(category, f"faq-{category}") if strapi_config:
endpoint = get_category_endpoint(category, strapi_config)
else:
# 回退到硬编码的默认值
endpoint = f"faq-{category}"
headers = {"Content-Type": "application/json"} headers = {"Content-Type": "application/json"}
if settings.strapi_api_token and settings.strapi_api_token.strip(): if settings.strapi_api_token and settings.strapi_api_token.strip():
@@ -238,10 +241,27 @@ async def search_faq_http(
try: try:
all_results = [] all_results = []
# 获取所有分类
if strapi_config:
categories = strapi_config.faq_categories
else:
# 回退到默认分类
categories = {
"register": type("obj", (object,), {"endpoint": "faq-register"}),
"order": type("obj", (object,), {"endpoint": "faq-order"}),
"pre-order": type("obj", (object,), {"endpoint": "faq-pre-order"}),
"payment": type("obj", (object,), {"endpoint": "faq-payment"}),
"shipment": type("obj", (object,), {"endpoint": "faq-shipment"}),
"return": type("obj", (object,), {"endpoint": "faq-return"}),
"other": type("obj", (object,), {"endpoint": "faq-other-question"}),
}
# Search all categories in parallel # Search all categories in parallel
async with httpx.AsyncClient(timeout=30.0) as client: async with httpx.AsyncClient(timeout=30.0) as client:
tasks = [] tasks = []
for category_name, endpoint in FAQ_CATEGORIES.items(): for category_name, category_config in categories.items():
endpoint = category_config.endpoint if hasattr(category_config, "endpoint") else f"faq-{category_name}"
headers = {"Content-Type": "application/json"} headers = {"Content-Type": "application/json"}
if settings.strapi_api_token and settings.strapi_api_token.strip(): if settings.strapi_api_token and settings.strapi_api_token.strip():
headers["Authorization"] = f"Bearer {settings.strapi_api_token}" headers["Authorization"] = f"Bearer {settings.strapi_api_token}"

View File

@@ -17,3 +17,6 @@ python-dotenv>=1.0.0
# Logging # Logging
structlog>=24.1.0 structlog>=24.1.0
# Configuration
pyyaml>=6.0