Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-skills/src/lexigram/ai/skills/registry/core.py: 45%

1"""SkillRegistry — central catalog for registering and discovering skills.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.ai.skills.exceptions import SkillAlreadyRegisteredError 

8from lexigram.logging import ( 

9 get_logger, 

10) 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.ai.agents import ToolProtocol 

14 from lexigram.contracts.ai.skills import SkillDefinition, SkillProtocol 

15 

16logger = get_logger(__name__) 

17 

18 

19class SkillRegistry: 

20 """Central registry for all available skills. 

21 

22 Supports registration, lookup by name, category-based listing, 

23 permission-filtered listing, and OpenAI function-calling schema export. 

24 """ 

25 

26 def __init__(self) -> None: 

27 """Initialise an empty registry.""" 

28 self._skills: dict[str, SkillProtocol] = {} 

29 self._categories: dict[str, list[str]] = {} 

30 

31 def register(self, skill: SkillProtocol) -> None: 

32 """Register a skill. 

33 

34 Args: 

35 skill: The skill to register. 

36 

37 Raises: 

38 SkillAlreadyRegisteredError: If a skill with the same name is already registered. 

39 """ 

40 name = skill.definition.name 

41 if name in self._skills: 

42 raise SkillAlreadyRegisteredError(name) 

43 self._skills[name] = skill 

44 cat = skill.definition.category 

45 self._categories.setdefault(cat, []).append(name) 

46 logger.debug("skill_registered", name=name, category=cat) 

47 

48 def register_tool(self, tool: ToolProtocol) -> None: 

49 """Wrap *tool* as a :class:`~lexigram.ai.skills.base.ToolSkillAdapter` and register it. 

50 

51 Args: 

52 tool: Any object satisfying ToolProtocol. 

53 """ 

54 from lexigram.ai.skills.base import ToolSkillAdapter 

55 

56 self.register(ToolSkillAdapter(tool)) 

57 

58 def get(self, name: str) -> SkillProtocol | None: 

59 """Return the skill registered under *name*, or ``None``. 

60 

61 Args: 

62 name: Skill name to look up. 

63 

64 Returns: 

65 The skill or None if not found. 

66 """ 

67 return self._skills.get(name) 

68 

69 def list_skills( 

70 self, 

71 category: str | None = None, 

72 permissions: list[str] | None = None, 

73 ) -> list[SkillDefinition]: 

74 """List registered skill definitions with optional filters. 

75 

76 Args: 

77 category: Return only skills in this category. 

78 permissions: Return only skills whose required permissions are a 

79 subset of the supplied permission set. 

80 

81 Returns: 

82 Matching skill definitions. 

83 """ 

84 skills = list(self._skills.values()) 

85 

86 if category is not None: 

87 names_in_cat = set(self._categories.get(category, [])) 

88 skills = [s for s in skills if s.definition.name in names_in_cat] 

89 

90 if permissions is not None: 

91 user_perms = set(permissions) 

92 skills = [ 

93 s for s in skills if set(s.definition.permissions).issubset(user_perms) 

94 ] 

95 

96 return [s.definition for s in skills] 

97 

98 def get_schemas(self) -> list[dict[str, Any]]: 

99 """Return OpenAI function-calling compatible schemas for all skills. 

100 

101 Returns: 

102 List of schema dicts in OpenAI ``tools`` format. 

103 """ 

104 return [ 

105 { 

106 "type": "function", 

107 "function": { 

108 "name": s.definition.name, 

109 "description": s.definition.description, 

110 "parameters": s.definition.parameters_schema, 

111 }, 

112 } 

113 for s in self._skills.values() 

114 ] 

115 

116 def __len__(self) -> int: 

117 return len(self._skills) 

118 

119 def __contains__(self, name: str) -> bool: 

120 return name in self._skills 

121 

122 

123__all__ = ["SkillRegistry"]