Skip to content

Skills API

dcc_mcp_core.SkillCatalogdcc_mcp_core.SkillScannerdcc_mcp_core.SkillWatcherdcc_mcp_core.SkillMetadatadcc_mcp_core.SkillSummarydcc_mcp_core.ToolDeclarationdcc_mcp_core.parse_skill_mddcc_mcp_core.scan_and_load

dcc_mcp_core.skill(纯 Python):skill_entryskill_successskill_errorskill_warningskill_exceptionrun_main

SkillCatalog

渐进式 Skill 发现与加载。线程安全(所有状态存储在 DashMap/DashSet 中)。

当通过 with_dispatcher() 附加了调度器时,加载 Skill 会自动为每个 Action 注册基于子进程的处理器 — 启用 Skills-First 工作流,Agent 无需手动注册处理器。

python
from dcc_mcp_core import SkillCatalog, ToolRegistry

registry = ToolRegistry()
catalog = SkillCatalog(registry)

构造函数

python
SkillCatalog(registry: ToolRegistry) -> SkillCatalog
参数类型说明
registryToolRegistry用于注册/注销工具的 Action 注册表

方法

方法返回值说明
with_dispatcher(dispatcher)附加 ToolDispatcher;启用 load_skill() 时的自动处理器注册
new_with_dispatcher(registry, dispatcher)创建带 dispatcher 的 catalog(构造器式)
discover(extra_paths=None, dcc_name=None)int扫描并填充目录;返回新发现的 skill 数量
load_skill(skill_name)List[str]加载 Skill;返回注册的 action 名称列表,未找到则报错
load_skills(skill_names)dict批量加载;返回 {name: Ok(actions) or Err(msg)}
unload_skill(skill_name)int卸载 Skill;返回移除的 action 数量,未加载则报错
remove_skill(skill_name)bool从目录中完全移除(已加载则先卸载)
clear()None清空所有 Skill(已加载的先卸载)
search_skills(query=None, tags=None, dcc=None, scope=None, limit=None)List[SkillSummary]统一发现:支持 scope"repo" | "user" | "system" | "admin")和 limit。空调用按 scope 优先级返回顶级 Skill(Admin > System > User > Repo)。
list_skills(status=None)List[SkillSummary]列出 Skill。status:"loaded""unloaded"None 为全部
get_skill_info(skill_name)SkillMetadata | None返回完整元数据,未找到返回 None
is_loaded(skill_name)bool指定 Skill 是否已加载
loaded_count()int已加载 Skill 的数量
__repr__()str字符串表示

示例

python
import os
from dcc_mcp_core import SkillCatalog, ToolRegistry, ToolDispatcher

os.environ["DCC_MCP_SKILL_PATHS"] = "/path/to/skills"

registry = ToolRegistry()
dispatcher = ToolDispatcher(registry)
catalog = SkillCatalog.new_with_dispatcher(registry, dispatcher)

# 发现 Skill
catalog.discover(extra_paths=["/extra/skills"], dcc_name="maya")

# 列出所有已发现的 Skill
for skill in catalog.list_skills():
    status = "loaded" if skill.loaded else "unloaded"
    print(f"  [{status}] {skill.name} v{skill.version}: {skill.description}")

# 搜索
results = catalog.search_skills(query="geometry", tags=["create"])
for s in results:
    print(f"  {s.name}: {s.tool_count} tools → {s.tool_names}")

# 加载 Skill(附加调度器后 Action 自动注册)
actions = catalog.load_skill("maya-geometry")
print(f"已注册 actions: {actions}")

# 获取完整元数据
meta = catalog.get_skill_info("maya-geometry")
if meta:
    print(meta.name, meta.tools)

# 查看已加载数量
print(catalog.loaded_count())

# 卸载
removed = catalog.unload_skill("maya-geometry")
print(f"已移除 {removed} 个 action")

SkillSummary

SkillCatalog.search_skills()list_skills() 返回的轻量级摘要对象。

属性(只读)

属性类型说明
namestrSkill 名称
descriptionstr简短描述
search_hintstr发现关键词提示(来自 SKILL.md search-hint:;回退到 description
tagsList[str]Skill 标签
dccstr目标 DCC(如 "maya"
versionstrSkill 版本
tool_countint声明的工具数量
tool_namesList[str]声明的工具名称列表
loadedbool当前是否已加载

特殊方法

方法说明
__repr__SkillSummary(name='...', loaded=True)

ToolDeclaration

Skill 中的单个工具声明,从 SKILL.md frontmatter 的 tools: 列表解析。

python
from dcc_mcp_core import ToolDeclaration

decl = ToolDeclaration(
    name="create_sphere",
    description="创建多边形球体",
    input_schema='{"type":"object","properties":{"radius":{"type":"number"}}}',
    read_only=False,
    destructive=False,
    idempotent=False,
    defer_loading=True,
    source_file="scripts/create_sphere.py",
)

构造函数

python
ToolDeclaration(
    name: str,
    description: str = "",
    input_schema: str | None = None,    # JSON Schema 字符串
    output_schema: str | None = None,   # JSON Schema 字符串
    read_only: bool = False,
    destructive: bool = False,
    idempotent: bool = False,
    defer_loading: bool = False,
    source_file: str = "",
) -> ToolDeclaration

字段(可读写)

字段类型默认值说明
namestr必填工具名称(在 Skill 内唯一)
descriptionstr""人类可读的描述
read_onlyboolFalse仅读取数据(无副作用)
destructiveboolFalse可能导致破坏性更改
idempotentboolFalse相同参数始终产生相同结果
defer_loadingboolFalse解析 SKILL.md 中的 defer-loading: / defer_loading:,供发现型 UI 使用
source_filestr""脚本文件的显式路径

input_schema 和 output_schema

内部以 JSON 值存储,非字符串。从 Python 构造时传入 JSON 字符串,会自动解析。

渐进式加载信号

tools/list 返回的未加载 skill stub 现在会带 annotations.deferredHint = true。调用 load_skill(...) 后,真实工具会以 deferredHint = false 暴露。


SkillMetadata

从 Skill 的 SKILL.md frontmatter 解析。同时支持 Anthropic Skills、ClawHub/OpenClaw 和 dcc-mcp-core 扩展格式。

构造函数

python
SkillMetadata(
    name: str,
    description: str = "",
    tools: List[str] | None = None,
    dcc: str = "python",
    tags: List[str] | None = None,
    search_hint: str = "",
    scripts: List[str] | None = None,
    skill_path: str = "",
    version: str = "1.0.0",
    depends: List[str] | None = None,
    metadata_files: List[str] | None = None,
) -> SkillMetadata

字段(可读写)

字段类型说明
namestr唯一 Skill 名称
descriptionstr简短描述
search_hintstrsearch_skills 的关键词提示(SKILL.md search-hint: 字段;回退到 description
toolsList[str]frontmatter 中的工具名称
dccstr目标 DCC 应用
tagsList[str]分类标签
scriptsList[str]发现的脚本文件路径
skill_pathstrSkill 目录绝对路径
versionstrSkill 版本
dependsList[str]依赖的 Skill 名称
metadata_filesList[str]metadata/ 目录中的 .md 文件路径

SkillScanner

扫描目录以发现 Skill 技能包,缓存文件修改时间以支持高效的重复扫描。

python
from dcc_mcp_core import SkillScanner

scanner = SkillScanner()

方法

方法返回值说明
scan(extra_paths=None, dcc_name=None, force_refresh=False)List[str]扫描路径以查找 Skill 目录
clear_cache()清除修改时间缓存和已发现列表

属性

属性类型说明
discovered_skillsList[str]之前发现的 Skill 目录路径

SkillWatcher

Skill 目录的热重载监控器。监控文件系统事件,当 SKILL.md 文件更改时自动重新加载技能元数据。

python
from dcc_mcp_core import SkillWatcher

watcher = SkillWatcher(debounce_ms=300)
watcher.watch("/path/to/skills")
skills = watcher.skills()

构造函数

python
SkillWatcher(debounce_ms: int = 300) -> SkillWatcher

方法

方法返回值说明
watch(path)开始递归监控 path。路径不存在则抛出 RuntimeError
unwatch(path)bool停止监控 path。曾被监控返回 True
skills()List[SkillMetadata]当前所有已加载技能的快照
skill_count()int当前已加载技能数量
watched_paths()List[str]当前正在监控的目录路径列表
reload()手动触发完整重载

函数

parse_skill_md

python
parse_skill_md(skill_dir: str) -> SkillMetadata | None

从 Skill 目录解析 SKILL.md。文件缺失或无效时返回 None

scan_skill_paths

python
scan_skill_paths(
    extra_paths: List[str] | None = None,
    dcc_name: str | None = None,
) -> List[str]

便捷包装器:创建 SkillScanner 并返回发现的 Skill 目录路径。

scan_and_load

python
scan_and_load(
    extra_paths: List[str] | None = None,
    dcc_name: str | None = None,
) -> tuple[List[SkillMetadata], List[str]]

完整流水线:扫描目录、加载所有 Skill、按依赖拓扑排序。

返回 (ordered_skills, skipped_dirs)。依赖缺失或有循环则抛出 ValueError

scan_and_load_lenient

python
scan_and_load_lenient(
    extra_paths: List[str] | None = None,
    dcc_name: str | None = None,
) -> tuple[List[SkillMetadata], List[str]]

scan_and_load 相同,但静默跳过依赖缺失的 Skill(通过日志记录警告)。仅循环依赖会抛出 ValueError

resolve_dependencies

python
resolve_dependencies(skills: List[SkillMetadata]) -> List[SkillMetadata]

拓扑排序,每个 Skill 出现在其依赖之后。依赖缺失或有循环则抛出 ValueError

validate_dependencies

python
validate_dependencies(skills: List[SkillMetadata]) -> List[str]

验证所有声明的依赖是否存在。返回错误消息列表(空列表表示无问题)。

expand_transitive_dependencies

python
expand_transitive_dependencies(
    skills: List[SkillMetadata],
    skill_name: str,
) -> List[str]

返回 skill_name 所有传递依赖的名称。依赖缺失或有循环则抛出 ValueError


搜索路径优先级

  1. extra_paths 参数(最高优先级)
  2. DCC_MCP_{APP}_SKILL_PATHS 环境变量(应用专属,如 DCC_MCP_MAYA_SKILL_PATHS
  3. DCC_MCP_SKILL_PATHS 环境变量(全局兜底)
  4. 平台特定 Skill 目录(DCC 特定,通过 get_skills_dir(dcc_name)
  5. 平台特定 Skill 目录(全局,通过 get_skills_dir()

环境变量

变量说明
DCC_MCP_{APP}_SKILL_PATHS应用专属 Skill 路径,如 DCC_MCP_MAYA_SKILL_PATHS(Windows 用 ;,Unix 用 :
DCC_MCP_SKILL_PATHS全局兜底 Skill 路径

create_skill_server

python
create_skill_server(
    app_name: str,
    config: McpHttpConfig | None = None,
    extra_paths: list[str] | None = None,
    dcc_name: str | None = None,
) -> McpHttpServer

Skills-First 工作流的推荐入口(v0.12.12+)。

一次调用即可为指定 DCC 应用创建完整配置的 McpHttpServer,自动完成:

  1. 创建 ToolRegistry + ToolDispatcher
  2. 创建与 dispatcher 连接的 SkillCatalog
  3. DCC_MCP_{APP}_SKILL_PATHSDCC_MCP_SKILL_PATHS 发现 Skills
  4. 返回已配置好的 McpHttpServer

参数:

参数类型说明
app_namestrDCC 名称(如 "maya""blender")— 用于推导环境变量名和 MCP 服务器名
configMcpHttpConfig | NoneHTTP 服务器配置;默认端口 8765
extra_pathslist[str] | None除环境变量外的额外 Skill 目录
dcc_namestr | None覆盖扫描的 DCC 过滤条件(默认与 app_name 相同)

返回值: McpHttpServer — 调用 .start() 开始服务。

示例:

python
import os
from dcc_mcp_core import create_skill_server, McpHttpConfig

os.environ["DCC_MCP_MAYA_SKILL_PATHS"] = "/studio/maya-skills"

server = create_skill_server("maya", McpHttpConfig(port=8765))
handle = server.start()
print(f"服务地址:{handle.mcp_url()}")

get_app_skill_paths_from_env

python
get_app_skill_paths_from_env(app_name: str) -> list[str]

DCC_MCP_{APP_NAME}_SKILL_PATHS 环境变量中返回 Skill 路径列表。

查找时不区分大小写,实际环境变量键名自动转换为大写(如 app_name="maya" 对应 DCC_MCP_MAYA_SKILL_PATHS)。

若环境变量未设置,返回 []

Action 命名规则

SkillCatalog.load_skill() 注册工具时,Action 名称遵循以下格式:

{skill名称(连字符转下划线)}__{工具名称}

示例:

  • Skill maya-geometry,工具 create_spheremaya_geometry__create_sphere
  • Skill blender-utils,工具 render-sceneblender_utils__render_scene

Skill Script 辅助工具(纯 Python)

dcc_mcp_core.skill纯 Python 子模块 — 无需编译扩展即可使用。 Skill 脚本作者可在 DCC 环境内直接导入,即便完整 wheel 包未安装也可正常运行。

python
from dcc_mcp_core.skill import skill_entry, skill_success, skill_error

所有辅助函数返回普通 dict,与 ToolResult 完全兼容。 若 dcc_mcp_core._core 可用,可将 dict 传入 validate_action_result() 获得类型化的 ToolResult 对象。


skill_success

python
skill_success(
    message: str,
    *,
    prompt: str | None = None,
    **context,
) -> dict

返回成功结果 dict。

参数类型说明
messagestr人类可读的执行摘要
promptstr | NoneAgent 下一步操作的提示(可选)
**contextAny附加到 context 的任意键值对
python
return skill_success(
    "时间线已设置为 1–120 帧",
    prompt="查看时间线滑块确认结果。",
    start_frame=1,
    end_frame=120,
)

skill_error

python
skill_error(
    message: str,
    error: str,
    *,
    prompt: str | None = None,
    possible_solutions: list[str] | None = None,
    **context,
) -> dict

返回失败结果 dict。

参数类型说明
messagestr面向用户的错误描述
errorstr技术错误字符串(异常 repr、错误码等)
promptstr | None恢复提示;默认为通用消息
possible_solutionslist[str] | None可操作建议,存储在 context["possible_solutions"]
python
return skill_error(
    "Maya 环境不可用",
    "ImportError: No module named 'maya'",
    prompt="请确保 Maya 已启动再调用此 Skill。",
    possible_solutions=["启动 Maya", "检查 DCC_MCP_MAYA_SKILL_PATHS"],
)

skill_warning

python
skill_warning(
    message: str,
    *,
    warning: str = "",
    prompt: str | None = None,
    **context,
) -> dict

返回成功但带警告的结果(success=Truecontext["warning"] 被设置)。

python
return skill_warning(
    "时间线已设置,end_frame 已截断为场景长度",
    warning="end_frame 9999 > 场景长度 240;已截断为 240",
    prompt="查看时间线滑块。",
    actual_end=240,
)

skill_exception

python
skill_exception(
    exc: BaseException,
    *,
    message: str | None = None,
    prompt: str | None = None,
    include_traceback: bool = True,
    possible_solutions: list[str] | None = None,
    **context,
) -> dict

从异常构建失败结果 dict。自动捕获 error_type 和完整堆栈跟踪(可选)存入 context

python
try:
    do_work()
except Exception as exc:
    return skill_exception(
        exc,
        possible_solutions=["请确认场景已打开"],
    )

@skill_entry

python
@skill_entry
def my_tool(param: str = "default", **kwargs) -> dict:
    ...

为 skill 函数添加标准错误处理的装饰器。

  • 自动捕获 ImportError(DCC 模块缺失)、ExceptionBaseException
  • 每种异常自动转换为规范的错误 dict
  • 直接运行脚本时(__name__ == "__main__"),将 JSON 结果打印到 stdout

完整示例(替代手动 try/except/main() 样板代码):

python
from dcc_mcp_core.skill import skill_entry, skill_success

@skill_entry
def set_timeline(start_frame: float = 1.0, end_frame: float = 120.0, **kwargs):
    """设置 Maya 播放时间线范围。"""
    import maya.cmds as cmds  # ImportError 由装饰器自动捕获

    min_frame = kwargs.get("min_frame", start_frame)
    max_frame = kwargs.get("max_frame", end_frame)

    cmds.playbackOptions(
        min=min_frame, max=max_frame,
        animationStartTime=start_frame, animationEndTime=end_frame,
    )
    return skill_success(
        f"时间线已设置为 {start_frame}{end_frame}",
        prompt="查看时间线滑块确认结果。",
        start_frame=start_frame,
        end_frame=end_frame,
    )

def main(**kwargs):
    """入口点;委托给 set_timeline。"""
    return set_timeline(**kwargs)

if __name__ == "__main__":
    from dcc_mcp_core.skill import run_main
    run_main(main)

run_main

python
run_main(main_fn: Callable[..., dict], argv: list[str] | None = None) -> None

执行 main_fn 并将 JSON 结果打印到 stdout。成功时调用 sys.exit(0),失败时调用 sys.exit(1)

用于 if __name__ == "__main__" 块:

python
if __name__ == "__main__":
    from dcc_mcp_core.skill import run_main
    run_main(main)

从 DCC 专用辅助函数迁移

如果之前使用 dcc_mcp_mayamaya_success / maya_error / maya_from_exception,通用版本直接对应:

旧(DCC 专用)新(通用)
maya_success(msg, prompt=..., **ctx)skill_success(msg, prompt=..., **ctx)
maya_error(msg, error, prompt=..., **ctx)skill_error(msg, error, prompt=..., **ctx)
maya_from_exception(exc_msg, ...)skill_exception(exc, ...)

dict 结构完全相同 — 两者均与 ToolResult 兼容。


结果序列化 — serialize_result / deserialize_result

基于 Rust 实现的 ToolResult 序列化工具。格式通过 SerializeFormat 枚举切换:当前使用 JSON,未来可升级到 MessagePack——调用方代码无需修改。

python
from dcc_mcp_core import (
    serialize_result, deserialize_result, SerializeFormat, success_result
)

SerializeFormat

python
class SerializeFormat:
    Json: SerializeFormat     # UTF-8 JSON 文本(默认)
    MsgPack: SerializeFormat  # 二进制 MessagePack(via rmp-serde)

serialize_result

python
serialize_result(
    result: ToolResult,
    format: SerializeFormat = SerializeFormat.Json,
) -> str | bytes

序列化 ToolResult

format返回类型说明
SerializeFormat.JsonstrUTF-8 JSON 字符串
SerializeFormat.MsgPackbytes二进制 MessagePack
python
arm = success_result("时间线已更新", start_frame=1, end_frame=120)

# JSON(默认)
json_str = serialize_result(arm)
assert isinstance(json_str, str)

# MessagePack
msgpack_bytes = serialize_result(arm, SerializeFormat.MsgPack)
assert isinstance(msgpack_bytes, bytes)

deserialize_result

python
deserialize_result(
    data: str | bytes,
    format: SerializeFormat = SerializeFormat.Json,
) -> ToolResult

str(JSON)或 bytes(MsgPack)反序列化为 ToolResultformat 必须与序列化时使用的格式一致。

python
original = success_result("完成", frame_count=240)
roundtrip = deserialize_result(serialize_result(original))
assert roundtrip.success
assert roundtrip.message == "完成"
assert roundtrip.context["frame_count"] == 240

run_main 的序列化流程

run_main()_core 可用时自动使用 serialize_result,在纯 Python 环境中回退到 json.dumps

result dict
    ↓ validate_action_result()  (类型安全验证)
ToolResult
    ↓ serialize_result(arm, SerializeFormat.Json)   (Rust JSON 写入器)
JSON 字符串 → stdout

未来切换到 MessagePack 时,只需修改 skill.py 中的 _serialize_result()——serialize_result / deserialize_result API 保持稳定。

Released under the MIT License.