让你的Claude Skills随时可用：将Skills暴露为MCP
的 YAML frontmatter（name + description），不读取完整内容。• 提取 frontmatter 中的 name 和 description。构建，将技能目录动态暴露为 MCP 服务。execute_python_script 脚本执行。• 返回 SKILL.md 或指定子文件的内容。创建 /common-mcp 端点。创建 /coohom-mcp 端点。创建 /re
·
原理
架构流程图
common
bim
coohom
render
SKILL.md
子文件
路径穿越
非 .py 文件
通过
服务启动
扫描 skills_data/ 目录
遍历所有子目录
子目录名称
创建 /common-mcp 端点
创建 /bim-mcp 端点
创建 /coohom-mcp 端点
创建 /render-mcp 端点
注册 MCP 工具
list_skills 列表查询
get_skill_content 内容获取
execute_python_script 脚本执行
解析 SKILL.md frontmatter
返回技能摘要列表
按需加载
返回完整技能内容
返回指定文件内容
安全检查
拒绝执行
执行脚本 30s 超时
返回执行结果
架构概览
本项目基于 MCP Python SDK 构建，将技能目录动态暴露为 MCP 服务。核心设计思路是：
-
1. 服务启动时扫描
skills_data/目录下的所有子目录 -
2. 为每个分类创建独立的 MCP 端点（如
/common-mcp、/bim-mcp） -
3. 每个 MCP 提供三个工具：列表查询、内容获取、脚本执行
核心技术
1. MCP Python SDK
使用 modelcontextprotocol/python-sdk 创建 MCP 服务，基于 FastAPI 提供 HTTP 流式传输能力：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name="技能服务",
streamable_http_path="/", # HTTP 端点路径
)2. 动态路由系统
服务启动时自动扫描 skills_data/ 目录：
skills/
├── skills_data/
│ ├── common/ → /common-mcp 端点
│ ├── bim/ → /bim-mcp 端点
│ ├── coohom/ → /coohom-mcp 端点
│ └── render/ → /render-mcp 端点在 server.py 中通过 _create_skill_mounts() 函数实现：
-
• 遍历
skills_data/下的所有子目录 -
• 为每个目录调用
create_skills_mcp()创建独立的 FastMCP 实例 -
• 使用 Starlette 的
Mount将 MCP 应用挂载到对应路径
3. 渐进式加载策略
采用 Claude Skills 官方推荐的渐进式加载策略：
-
1. 列表阶段：只解析
SKILL.md的 YAML frontmatter（name + description），不读取完整内容 -
2. 详情阶段：通过
get_skill_content()按需加载完整文件内容
4. 安全机制
提供三层安全保障：
|
机制 |
实现方式 |
|---|---|
| 路径穿越防护 |
使用 |
| 文件类型限制 |
只允许执行 |
| 执行超时 |
subprocess.run(timeout=30) 限制最长执行时间 |
| 工作目录隔离 | cwd=str(target_dir)
确保脚本在技能目录下执行 |
MCP 工具
每个分类的 MCP 服务提供以下三个工具：
list_skills
-
• 扫描目录下所有包含
SKILL.md的子目录 -
• 提取 frontmatter 中的 name 和 description
-
• 返回技能摘要列表
get_skill_content
-
• 根据技能名称查找对应目录
-
• 支持模糊匹配（前缀匹配）
-
• 返回 SKILL.md 或指定子文件的内容
execute_python_script
-
• 在技能目录下安全执行 Python 脚本
-
• 支持传递命令行参数
-
• 返回标准输出/错误输出和返回码
使用方法
MCP 端点地址
|
技能分类 |
MCP 端点地址 |
|---|---|
|
通用技能 |
http://your-domain/common-mcp |
|
BIM 技能 |
http://your-domain/bim-mcp |
|
Coohom技能 |
http://your-domain/coohom-mcp |
|
渲染技能 |
http://your-domain/render-mcp |
完整代码
skills.py
"""技能管理 MCP 服务控制器
支持动态创建技能 MCP 服务，采用渐进式加载方式：
1. list_skills - 列出所有技能的 name + description
2. get_skill_content - 获取技能的完整 SKILL.md 内容或子文件
3. execute_python_script - 安全执行指定技能目录下的 Python 脚本
"""
import re
import subprocess
from pathlib import Path
from typing import List, Optional
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
# 禁用 Host 验证，允许所有来源
mcp_security_settings = TransportSecuritySettings(
enable_dns_rebinding_protection=False,
)
class SkillSummary(BaseModel):
"""技能摘要（用于列表展示）"""
name: str = Field(description="技能名称")
description: str = Field(default="", description="技能描述")
class SkillListResponse(BaseModel):
"""技能列表响应"""
data: List[SkillSummary] = Field(description="技能摘要列表")
message: str = Field(default="获取成功", description="操作结果消息")
count: int = Field(default=0, description="技能数量")
class SkillContentResponse(BaseModel):
"""技能内容响应"""
data: str = Field(description="文件内容")
message: str = Field(default="获取成功", description="操作结果消息")
file_path: str = Field(description="文件路径")
class ErrorResponse(BaseModel):
"""错误响应"""
error: str = Field(description="错误消息")
message: str = Field(default="操作失败", description="操作结果消息")
# 脚本执行响应模型
class ExecutionResult(BaseModel):
"""代码执行结果"""
output: str = Field(default="", description="标准输出")
error: str = Field(default="", description="错误输出")
class ScriptExecuteResponse(BaseModel):
"""脚本执行响应"""
data: ExecutionResult = Field(description="执行结果")
message: str = Field(default="执行成功", description="操作结果消息")
success: bool = Field(description="是否执行成功")
return_code: int = Field(default=0, description="返回码")
def _parse_skill_frontmatter(skill_md_path: Path) -> Optional[dict]:
"""解析 SKILL.md 的 YAML frontmatter，提取 name 和 description
Args:
skill_md_path: SKILL.md 文件路径
Returns:
包含 name 和 description 的字典，解析失败返回 None
"""
try:
content = skill_md_path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
return None
# 匹配 YAML frontmatter: --- ... ---
match = re.match(r"^---\s*\n(.+?)\n---", content, re.DOTALL)
if not match:
return None
frontmatter = match.group(1)
result = {}
# 简单解析 key: value 格式（避免引入 yaml 依赖的开销）
for line in frontmatter.split("\n"):
line = line.strip()
if ":" in line:
key, _, value = line.partition(":")
key = key.strip()
value = value.strip()
if key in ("name", "description"):
result[key] = value
if "name" not in result:
return None
return result
def _scan_skills(skill_dir: Path) -> List[SkillSummary]:
"""扫描目录下所有包含 SKILL.md 的子目录，提取技能摘要
Args:
skill_dir: 技能分类目录（如 skills_data/common/）
Returns:
SkillSummary 列表
"""
skills = []
if not skill_dir.exists() or not skill_dir.is_dir():
return skills
for item in sorted(skill_dir.iterdir()):
if not item.is_dir():
continue
skill_md = item / "SKILL.md"
if not skill_md.exists():
continue
parsed = _parse_skill_frontmatter(skill_md)
if parsed:
skills.append(
SkillSummary(
name=parsed.get("name", item.name),
description=parsed.get("description", ""),
)
)
return skills
def create_skills_mcp(skill_dir: str) -> FastMCP:
"""创建技能 MCP 服务
Args:
skill_dir: 技能分类目录路径（如 skills_data/common）
Returns:
FastMCP 实例
"""
skill_path = Path(skill_dir)
category_name = skill_path.name
# 创建 MCP 实例
mcp = FastMCP(
name=f"{category_name}技能服务",
transport_security=mcp_security_settings,
instructions=f"""{category_name} 技能管理服务:
## 工具列表
1. list_skills - 列出所有可用技能
- 返回该分类下所有技能的 name 和 description
- 用于了解有哪些技能可用
2. get_skill_content - 获取技能内容
- 根据技能名称获取 SKILL.md 内容
- 或根据技能名称+相对路径获取子文件内容
## 使用流程
先调用 list_skills 查看可用技能列表，再通过 get_skill_content 获取感兴趣的技能详情。
""",
streamable_http_path="/",
)
def find_skill_dir(target_name: str) -> Optional[Path]:
"""在技能目录中查找目标技能"""
# 精确匹配
exact_match = skill_path / target_name
if exact_match.exists() and (exact_match / "SKILL.md").exists():
return exact_match
# 模糊匹配（前缀）
for item in skill_path.iterdir():
if item.is_dir() and item.name.startswith(target_name):
if (item / "SKILL.md").exists():
return item
return None
@mcp.tool()
async def list_skills() -> list:
"""列出所有可用技能
扫描当前分类目录下所有技能，返回每个技能的名称和描述。
用于快速了解有哪些技能可用，再按需获取详细内容。
Returns:
技能列表响应（包含 name 和 description）
"""
try:
skills = _scan_skills(skill_path)
response = SkillListResponse(
data=skills,
message="获取成功",
count=len(skills),
)
return [response.model_dump()]
except Exception as e:
error = ErrorResponse(
error=f"获取技能列表失败: {str(e)}",
message="获取失败",
)
return [error.model_dump()]
@mcp.tool()
async def get_skill_content(
target_name: str = Field(
description="目标技能名称",
examples=["pdf", "frontend-design", "docx"],
),
relative_path: str = Field(
default="",
description="相对于技能目录的文件路径，空字符串表示获取 SKILL.md",
examples=["", "scripts/helper.py", "README.md"],
),
) -> list:
"""获取技能内容
根据技能名称获取技能文件内容：
- 如果 relative_path 为空，返回 SKILL.md 内容
- 如果指定 relative_path，返回技能目录下对应文件的内容
Args:
target_name: 目标技能名称
relative_path: 相对文件路径
Returns:
技能文件内容
"""
try:
# 查找技能目录
target_dir = find_skill_dir(target_name)
if target_dir is None:
return [ErrorResponse(
error=f"未找到技能: {target_name}",
message="技能不存在",
).model_dump()]
# 确定文件路径
if relative_path == "" or relative_path.lower() == "skill.md":
file_path = target_dir / "SKILL.md"
else:
file_path = target_dir / relative_path
# 安全检查：防止路径穿越
try:
file_path.resolve().relative_to(skill_path.resolve())
except ValueError:
return [ErrorResponse(
error=f"非法路径: {relative_path}",
message="路径不合法",
).model_dump()]
# 检查文件是否存在
if not file_path.exists():
return [ErrorResponse(
error=f"文件不存在: {file_path.relative_to(skill_path)}",
message="文件不存在",
).model_dump()]
# 读取文件内容
try:
content = file_path.read_text(encoding="utf-8")
except UnicodeDecodeError:
content = file_path.read_text(encoding="gbk")
return [SkillContentResponse(
data=content,
message="获取成功",
file_path=str(file_path.relative_to(skill_path)),
).model_dump()]
except Exception as e:
return [ErrorResponse(
error=f"获取内容失败: {str(e)}",
message="获取失败",
).model_dump()]
@mcp.tool()
async def execute_python_script(
target_name: str = Field(
description="目标技能名称",
examples=["pdf", "frontend-design", "docx"],
),
script_path: str = Field(
description="相对于技能目录的 Python 脚本路径",
examples=["scripts/process.py", "tools/helper.py", "main.py"],
),
args: str = Field(
default="",
description="传递给脚本的命令行参数（空格分隔）",
examples=["--input data.json", "--verbose", ""],
),
) -> list:
"""安全执行技能目录下的 Python 脚本
在技能目录下安全执行指定的 Python 脚本，并返回执行结果。
脚本必须在当前技能分类目录内，防止路径穿越攻击。
Args:
target_name: 目标技能名称
script_path: 相对于技能目录的 Python 脚本路径
args: 传递给脚本的命令行参数
Returns:
脚本执行结果
"""
try:
# 查找技能目录
target_dir = find_skill_dir(target_name)
if target_dir is None:
return [ErrorResponse(
error=f"未找到技能: {target_name}",
message="技能不存在",
).model_dump()]
# 构建脚本完整路径
full_script_path = target_dir / script_path
# 安全检查：防止路径穿越
try:
full_script_path.resolve().relative_to(skill_path.resolve())
except ValueError:
return [ErrorResponse(
error=f"非法路径: {script_path}",
message="路径不合法，必须在技能目录内",
).model_dump()]
# 检查文件是否存在且为 .py 文件
if not full_script_path.exists():
return [ErrorResponse(
error=f"脚本文件不存在: {full_script_path.relative_to(skill_path)}",
message="文件不存在",
).model_dump()]
if not full_script_path.suffix == ".py":
return [ErrorResponse(
error=f"只支持执行 .py 文件: {full_script_path.suffix}",
message="文件类型不支持",
).model_dump()]
# 准备命令和参数
cmd = ["python", str(full_script_path)]
if args.strip():
cmd.extend(args.strip().split())
# 执行脚本
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30, # 30 秒超时
cwd=str(target_dir), # 在技能目录下执行
)
# 构建响应
if result.returncode == 0:
return [ScriptExecuteResponse(
data=ExecutionResult(
output=result.stdout,
error=result.stderr,
),
message="执行成功",
success=True,
return_code=result.returncode,
).model_dump()]
else:
return [ScriptExecuteResponse(
data=ExecutionResult(
output=result.stdout,
error=result.stderr,
),
message="执行失败",
success=False,
return_code=result.returncode,
).model_dump()]
except subprocess.TimeoutExpired:
return [ErrorResponse(
error="脚本执行超时（30秒）",
message="执行超时",
).model_dump()]
except Exception as e:
return [ErrorResponse(
error=f"执行脚本失败: {str(e)}",
message="执行失败",
).model_dump()]
return mcpserver.py
"""Skills MCP Server 主入口
支持通过路径参数动态指定技能分类目录，例如：
- /common-mcp -> 对应 skills_data/common 目录
- /bim-mcp -> 对应 skills_data/bim 目录
"""
import os
import sys
import contextlib
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, List, Tuple
from fastapi import FastAPI
from starlette.routing import Mount
from starlette.responses import JSONResponse
# 添加项目根目录到 Python 路径（支持 Docker 和本地运行）
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
# 技能数据根目录
SKILLS_ROOT = PROJECT_ROOT / "skills_data"
# 存储所有 MCP 实例的列表：(category_name, mcp_instance)
_mcp_instances: List[Tuple[str, Any]] = []
def _get_available_categories() -> List[str]:
"""获取 skills_data 下所有可用的技能分类目录"""
if not SKILLS_ROOT.exists():
return []
return sorted([
d.name for d in SKILLS_ROOT.iterdir()
if d.is_dir() and not d.name.startswith(("_", "."))
])
def _create_skill_mounts() -> List:
"""为每个技能分类创建 Mount 对象，并存储 MCP 实例"""
global _mcp_instances
mounts = []
_mcp_instances.clear()
for category_dir in sorted(SKILLS_ROOT.iterdir()):
if not category_dir.is_dir():
continue
# 跳过特殊目录
if category_dir.name.startswith(("_", ".")):
continue
category_name = category_dir.name
mount_path = f"/{category_name}-mcp"
# 创建该分类的 MCP 服务
import importlib.util
import sys
skills_module_path = Path(__file__).parent / "controller" / "skills.py"
spec = importlib.util.spec_from_file_location("skills", skills_module_path)
skills_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(skills_module)
create_skills_mcp = skills_module.create_skills_mcp
skills_mcp = create_skills_mcp(str(category_dir))
# 存储 MCP 实例供 lifespan 使用
_mcp_instances.append((category_name, skills_mcp))
# 创建 Mount
mounts.append(Mount(mount_path, app=skills_mcp.streamable_http_app()))
print(f"已准备技能 MCP: {mount_path} -> {category_dir}")
return mounts
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理，管理所有 MCP 服务器的生命周期"""
print("Skills MCP Server 启动中...")
# 启动所有 MCP 服务的 session_manager
async with contextlib.AsyncExitStack() as stack:
for category_name, mcp_instance in _mcp_instances:
await stack.enter_async_context(mcp_instance.session_manager.run())
print(f"✓ {category_name}-mcp 已启动")
print("所有 FastMCP 服务器启动完成！")
yield
# 关闭时自动清理
print("正在关闭所有 FastMCP 服务器...")
# 创建 FastAPI 应用
app = FastAPI(
title="Skills MCP Server",
description="基于 FastMCP 的技能管理和检索服务，支持动态路由",
version="0.0.1",
lifespan=lifespan,
)
# 预先创建所有技能分类的 Mount 对象并挂载到应用
skill_mounts = _create_skill_mounts()
app.router.routes.extend(skill_mounts)
@app.get("/healthz")
def healthz():
"""健康检查端点"""
return {
"status": "OK",
"service": "Skills MCP Server"
}
@app.get("/")
async def root():
"""根端点，返回服务器信息和可用服务列表"""
categories = _get_available_categories()
return {
"service": "Skills MCP Server",
"status": "running",
"skills_root": str(SKILLS_ROOT),
"available_categories": categories,
"mcp_servers": [
{
"name": category,
"path": f"/{category}-mcp",
"endpoint": f"http://localhost:8000/{category}-mcp"
}
for category in categories
],
"endpoints": {
"health": "/healthz",
"root": "/",
"skills_mcp": "/{category}-mcp",
},
}
@app.get("/{category}-mcp/")
async def skills_mcp_info(category: str):
"""获取技能分类 MCP 信息"""
category_dir = SKILLS_ROOT / category
if not category_dir.exists() or not category_dir.is_dir():
return JSONResponse(
status_code=404,
content={
"error": f"技能分类不存在: {category}",
"available_categories": _get_available_categories(),
},
)
# 统计该分类下的技能数量
skill_count = sum(
1 for item in category_dir.iterdir()
if item.is_dir() and (item / "SKILL.md").exists()
)
return {
"category": category,
"category_path": str(category_dir),
"skill_count": skill_count,
"mcp_endpoint": f"/{category}-mcp",
}
if __name__ == "__main__":
import uvicorn
if not os.path.exists("../logs"):
os.makedirs("../logs")
uvicorn.run(
app,
host="0.0.0.0",
port=8000,
timeout_keep_alive=60,
timeout_graceful_shutdown=30,
log_level="info"
)
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
9
0- 0
已为社区贡献5条内容
所有评论(0)