37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155 | class SkillsTool(Node):
"""Tool for skills: discover and get content from a skill registry (Dynamiq or FileSystem).
After get, apply the skill's instructions yourself and provide the result in your final answer.
"""
group: Literal[NodeGroup.TOOLS] = NodeGroup.TOOLS
name: str = "SkillsTool"
description: str = (
"Manages skills (instructions only). Use this tool to:\n"
"- List available skills: action='list'\n"
"- Get skill content: action='get', skill_name='...'. "
"For large skills use section='Section title' or line_start/line_end to read only a part.\n\n"
"After get, apply the skill's instructions yourself in your reasoning and provide the result "
"in your final answer. Do not call the tool again with user content to transform; "
"the tool only provides instructions; you produce the output."
)
skill_registry: BaseSkillRegistry = Field(
...,
description="Registry providing skills (Dynamiq or FileSystem).",
)
input_schema: ClassVar[type[SkillsToolInputSchema]] = SkillsToolInputSchema
@property
def to_dict_exclude_params(self):
return super().to_dict_exclude_params | {
"skill_registry": True,
}
def execute(
self, input_data: SkillsToolInputSchema, config: RunnableConfig | None = None, **kwargs
) -> dict[str, Any]:
action = input_data.action
logger.info("SkillsTool - action=%s", action.value)
if action == SkillsToolAction.LIST:
return self._list_skills()
if action == SkillsToolAction.GET:
if not input_data.skill_name:
raise ToolExecutionException("skill_name required for get", recoverable=True)
return self._get_skill(
input_data.skill_name,
section=input_data.section,
line_start=input_data.line_start,
line_end=input_data.line_end,
)
raise ToolExecutionException(f"Unknown action: {action.value}", recoverable=True)
def _list_skills(self) -> dict[str, Any]:
metadata_list = self.skill_registry.get_skills_metadata()
skills_info = [{"name": m.name, "description": m.description} for m in metadata_list]
names = [m.name for m in metadata_list]
logger.info("SkillsTool - list: %d skill(s) %s", len(metadata_list), names)
return {
"content": {
"available_skills": skills_info,
"total": len(metadata_list),
}
}
def _get_skill(
self,
skill_name: str,
section: str | None = None,
line_start: int | None = None,
line_end: int | None = None,
) -> dict[str, Any]:
try:
instructions = self.skill_registry.get_skill_instructions(skill_name)
except Exception as e:
raise ToolExecutionException(f"Failed to get skill '{skill_name}': {e}", recoverable=True) from e
def _content_dict(sliced_instructions: str, section_used: str | None = None) -> dict[str, Any]:
out: dict[str, Any] = {
"name": instructions.name,
"description": instructions.description,
"instructions": sliced_instructions,
}
if section_used is not None:
out["section_used"] = section_used
if instructions.metadata:
out["metadata"] = instructions.metadata
return out
if section is not None or line_start is not None or line_end is not None:
sliced, section_used = extract_skill_content_slice(
instructions.instructions,
section=section,
line_start=line_start,
line_end=line_end,
)
if section is not None and section_used is None:
raise ToolExecutionException(
f"Section '{section}' not found in skill '{skill_name}'.",
recoverable=True,
)
one_line = sliced.replace("\n", " ").strip()
preview = (one_line[:50] + "...") if len(one_line) > 50 else one_line
logger.info(
"SkillsTool - get: skill=%s (section=%s, lines=%s-%s) -> content received (%d chars), preview: %s",
skill_name,
section,
line_start,
line_end,
len(sliced),
preview,
)
return {"content": _content_dict(sliced, section_used)}
one_line = instructions.instructions.replace("\n", " ").strip()
preview = (one_line[:50] + "...") if len(one_line) > 50 else one_line
logger.info(
"SkillsTool - get: skill=%s -> content received (%d chars), preview: %s",
skill_name,
len(instructions.instructions),
preview,
)
return {"content": _content_dict(instructions.instructions)}
|