94 lines
2.3 KiB
Python
94 lines
2.3 KiB
Python
import os
|
|
from skillkit import SkillManager
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict, Any
|
|
from appPublic.dictObject import DictObject
|
|
from appPublic.log import debug, exception
|
|
|
|
def find_missing_params(
|
|
input_schema: Dict[str, Any],
|
|
provided_params: Dict[str, Any],
|
|
) -> list[str]:
|
|
"""
|
|
根据 YAML schema 判断缺失的必填参数
|
|
"""
|
|
missing = []
|
|
|
|
for name, spec in input_schema.items():
|
|
if spec.get("required", False) and name not in provided_params:
|
|
missing.append(name)
|
|
|
|
return missing
|
|
|
|
def load_schemas(path) -> Dict[str, Any]:
|
|
"""
|
|
从 YAML 文件中读取 script 输入参数定义
|
|
|
|
返回格式：
|
|
{
|
|
"script": "Slider",
|
|
"inputs": {
|
|
"min": {...},
|
|
"max": {...}
|
|
}
|
|
}
|
|
"""
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Script yaml not found: {yaml_path}")
|
|
|
|
with path.open("r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
return DictObject(**data)
|
|
|
|
class SkillkitWrapper:
|
|
def __init__(self, user_skillsroot, sys_skillsroot=None):
|
|
|
|
self.client = SkillManager(project_skill_dir=user_skillsroot,
|
|
anthropic_config_dir=sys_skillsroot)
|
|
self.client.discover()
|
|
self.schemas = {}
|
|
|
|
def list_skills(self):
|
|
return self.client.list_skills()
|
|
|
|
def load_skill(self, skill_name):
|
|
skill = self.client.load_skill(skill_name)
|
|
schemaspath = skill.base_directory / 'schemas.yaml'
|
|
if schemaspath.exists():
|
|
if not self.schemas.get(skill_name):
|
|
data = load_schemas(schemaspath)
|
|
self.schemas[skill_name] = data
|
|
|
|
return skill
|
|
|
|
def get_script_schema(self, skill_name, script_name):
|
|
skill = self.load_skill(skill_name)
|
|
d = self.schemas.get(skill_name)
|
|
if not d:
|
|
# debug(f'{skill_name=}, {self.schemas=} ,has not schemas.yaml')
|
|
return None
|
|
return d.scripts.get(script_name)
|
|
|
|
def get_script_params(self, skill_name, script_name):
|
|
m = self.get_script_schema(skill_name, script_name)
|
|
if not m:
|
|
return []
|
|
|
|
return m.inputs
|
|
|
|
def get_skill_scripts(self, skill_name):
|
|
skill = self.load_skill(skill_name)
|
|
return skill.scripts
|
|
|
|
def execute_skill_script(self, skill_name, script_name, args={}):
|
|
return self.client.execute_skill_script(
|
|
skill_name=skill_name,
|
|
script_name=script_name,
|
|
arguments=args
|
|
)
|
|
def invoke_skill(self, skill_name: str, script_name: str, params: dict):
|
|
return self.client.invoke_skill(skill_nmae, params)
|
|
|