Coverage for src / lexigram / contracts / ai / skills.py: 0%
62 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-15 18:57 +0800
1"""Composable skills/tools system contracts."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from typing import TYPE_CHECKING, Any
8from typing_extensions import Protocol, runtime_checkable
10from lexigram.contracts.ai.exceptions import SkillError
12if TYPE_CHECKING:
13 from lexigram.contracts.core.result import Result
16@dataclass(frozen=True)
17class SkillParameter:
18 """Parameter definition for a skill.
20 Attributes:
21 name: Parameter name
22 type: Parameter type (e.g., "string", "integer", "boolean")
23 description: Human-readable description
24 required: Whether this parameter is required
25 default: Default value if not provided
26 enum: List of allowed values (if restricted)
27 min_value: Numeric minimum (if applicable)
28 max_value: Numeric maximum (if applicable)
29 max_length: String max length (if applicable)
30 """
32 name: str
33 type: str
34 description: str
35 required: bool = True
36 default: Any = None
37 enum: list[Any] | None = None
38 min_value: float | None = None
39 max_value: float | None = None
40 max_length: int | None = None
43@dataclass(frozen=True)
44class SkillDefinition:
45 """Definition of a skill's interface and behavior.
47 Attributes:
48 name: Unique skill name
49 description: What the skill does
50 parameters_schema: JSON Schema for parameters
51 returns_schema: JSON Schema for return value
52 category: Skill category for organization
53 requires_confirmation: If True, requires user approval before execution
54 cacheable: Whether results can be cached
55 max_retries: Maximum retry attempts on failure
56 timeout_seconds: Execution timeout in seconds
57 permissions: Required permissions to execute
58 """
60 name: str
61 description: str
62 parameters_schema: dict[str, Any] = field(default_factory=dict)
63 returns_schema: dict[str, Any] = field(default_factory=dict)
64 category: str = "general"
65 requires_confirmation: bool = False
66 cacheable: bool = False
67 max_retries: int = 0
68 timeout_seconds: float = 30.0
69 permissions: list[str] = field(default_factory=list)
70 metadata: dict[str, Any] = field(default_factory=dict)
73@dataclass(frozen=True)
74class SkillResult:
75 """Result of a skill execution.
77 Attributes:
78 skill_name: Name of the executed skill
79 success: Whether execution succeeded
80 output: The output value (if successful)
81 error: Error message (if failed)
82 duration_ms: Time taken to execute
83 cached: Whether result came from cache
84 metadata: Additional execution metadata
85 """
87 skill_name: str
88 success: bool
89 output: Any = None
90 error: str | None = None
91 duration_ms: float = 0.0
92 cached: bool = False
93 metadata: dict[str, Any] = field(default_factory=dict)
96@runtime_checkable
97class SkillProtocol(Protocol):
98 """Protocol for a composable skill.
100 A skill is an async-callable capability that can be registered,
101 discovered, and executed with validation and error handling.
102 """
104 @property
105 def definition(self) -> SkillDefinition:
106 """Get the skill definition.
108 Returns:
109 The skill's definition including parameters and metadata
110 """
111 ...
113 async def execute(self, **kwargs: Any) -> Result[SkillResult, SkillError]:
114 """Execute the skill.
116 Args:
117 **kwargs: Skill-specific parameters
119 Returns:
120 Result with SkillResult on success or SkillError on failure
121 """
122 ...
124 def validate(self, params: dict[str, Any]) -> list[str]:
125 """Validate parameters against the definition.
127 Args:
128 params: Parameters to validate
130 Returns:
131 List of validation errors (empty if valid)
132 """
133 ...
136@runtime_checkable
137class SkillRegistryProtocol(Protocol):
138 """Protocol for registering and discovering skills.
140 Maintains a registry of available skills with support for
141 filtering by category and permissions.
142 """
144 def register(self, skill: SkillProtocol) -> None:
145 """Register a skill in the registry.
147 Args:
148 skill: The skill to register
149 """
150 ...
152 def get(self, name: str) -> SkillProtocol | None:
153 """Get a skill by name.
155 Args:
156 name: The skill name
158 Returns:
159 The skill, or None if not found
160 """
161 ...
163 def list_skills(
164 self,
165 category: str | None = None,
166 permissions: list[str] | None = None,
167 ) -> list[SkillDefinition]:
168 """List available skills with optional filtering.
170 Args:
171 category: Filter by skill category
172 permissions: Filter to skills that match any of these permissions
174 Returns:
175 List of skill definitions matching the criteria
176 """
177 ...
179 def get_schemas(self) -> list[dict[str, Any]]:
180 """Get OpenAI function-calling compatible schemas.
182 Returns:
183 List of skill schemas in OpenAI function calling format
184 """
185 ...
188@runtime_checkable
189class SkillExecutorProtocol(Protocol):
190 """Protocol for executing skills with lifecycle management.
192 Handles skill selection, validation, execution, retry, caching,
193 permission checking, and observability.
194 """
196 async def execute(
197 self,
198 skill_name: str,
199 params: dict[str, Any],
200 user_id: str | None = None,
201 session_id: str | None = None,
202 ) -> Result[SkillResult, SkillError]:
203 """Execute a skill with full lifecycle management.
205 Args:
206 skill_name: Name of the skill to execute
207 params: Parameters for the skill
208 user_id: Optional user ID for permission checks
209 session_id: Optional session ID for context
211 Returns:
212 Result with SkillResult on success or SkillError on failure
213 """
214 ...
217@runtime_checkable
218class ToolkitProtocol(Protocol):
219 """Protocol for a collection of related skills.
221 A toolkit groups semantically related skills together, such as
222 database operations, web browsing, or file operations.
223 """
225 @property
226 def tools(self) -> tuple[SkillProtocol, ...]:
227 """Get the collection of skills in this toolkit.
229 Returns:
230 Tuple of SkillProtocol instances provided by this toolkit.
231 """
232 ...
234 @property
235 def name(self) -> str:
236 """Get the toolkit name.
238 Returns:
239 Unique identifier for this toolkit.
240 """
241 ...
243 @property
244 def description(self) -> str:
245 """Get the toolkit description.
247 Returns:
248 Human-readable description of what this toolkit provides.
249 """
250 ...
253__all__ = [
254 "SkillDefinition",
255 "SkillError",
256 "SkillExecutorProtocol",
257 "SkillParameter",
258 "SkillProtocol",
259 "SkillRegistryProtocol",
260 "SkillResult",
261 "ToolkitProtocol",
262]