Skip to content
Lexigram
GitHubDiscord

API Reference

Protocols

Section titled “Protocols”

Protocol for executing skills with lifecycle management.

Handles skill selection, validation, execution, retry, caching, permission checking, and observability.

async def execute(
    skill_name: str,
    params: dict[str, Any],
    user_id: str | None = None,
    session_id: str | None = None
) -> Result[SkillResult, SkillError]

Execute a skill with full lifecycle management.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill to execute
`params`dict[str, Any]Parameters for the skill
`user_id`str | NoneOptional user ID for permission checks
`session_id`str | NoneOptional session ID for context
Returns
TypeDescription
Result[SkillResult, SkillError]Result with SkillResult on success or SkillError on failure

Protocol for a composable skill.

A skill is an async-callable capability that can be registered, discovered, and executed with validation and error handling.

property definition() -> SkillDefinition

Get the skill definition.

Returns
TypeDescription
SkillDefinitionThe skill's definition including parameters and metadata
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Execute the skill.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Result with SkillResult on success or SkillError on failure
def validate(params: dict[str, Any]) -> list[str]

Validate parameters against the definition.

Parameters
ParameterTypeDescription
`params`dict[str, Any]Parameters to validate
Returns
TypeDescription
list[str]List of validation errors (empty if valid)

Protocol for registering and discovering skills.

Maintains a registry of available skills with support for filtering by category and permissions.

def register(skill: SkillProtocol) -> None

Register a skill in the registry.

Parameters
ParameterTypeDescription
`skill`SkillProtocolThe skill to register
def get(name: str) -> SkillProtocol | None

Get a skill by name.

Parameters
ParameterTypeDescription
`name`strThe skill name
Returns
TypeDescription
SkillProtocol | NoneThe skill, or None if not found
def list_skills(
    category: str | None = None,
    permissions: list[str] | None = None
) -> list[SkillDefinition]

List available skills with optional filtering.

Parameters
ParameterTypeDescription
`category`str | NoneFilter by skill category
`permissions`list[str] | NoneFilter to skills that match any of these permissions
Returns
TypeDescription
list[SkillDefinition]List of skill definitions matching the criteria
def get_schemas() -> list[dict[str, Any]]

Get OpenAI function-calling compatible schemas.

Returns
TypeDescription
list[dict[str, Any]]List of skill schemas in OpenAI function calling format

Classes

Section titled “Classes”

Abstract base class for class-based skill definitions.

Subclass this and define a definition class attribute plus an execute method. Parameter validation default uses the JSON schema stored on the definition; override validate for custom rules.

async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Execute the skill with the provided keyword arguments.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Result wrapping SkillResult on success or SkillError on failure.
def validate(params: dict[str, Any]) -> list[str]

Validate params against the skill definition’s parameter schema.

Parameters
ParameterTypeDescription
`params`dict[str, Any]Parameters to validate.
Returns
TypeDescription
list[str]List of validation error messages (empty list means valid).
def to_tool() -> SkillToolAdapter

Convert this skill to a ToolProtocol for use in Lexigram agents.

Returns
TypeDescription
SkillToolAdapterA SkillToolAdapter that implements ToolProtocol.

Execute code in a sandboxed environment and return the output.

This is a stub implementation. For production use, replace with a proper sandboxed execution backend (e.g. Docker, WASM, a remote code execution service).

Required permission: code.execute.

property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the code_execute skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Return a stub result indicating execution is not implemented.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok stub result with ``output=""``, ``stderr=""``, ``exit_code=0``, and ``stub=True``.

Execute read-only SELECT statements and return row dicts.

Only SELECT statements are accepted. A LIMIT clause is automatically appended when the query does not already contain one, capped at max_rows (default 100) to prevent runaway reads.

Parameters
ParameterTypeDescription
`db`A DatabaseProviderProtocol instance.
`max_rows`Maximum rows returned per query.
def __init__(
    db: DatabaseProviderProtocol,
    max_rows: int = _DEFAULT_LIMIT
) -> None

Initialise the skill with a database provider.

Parameters
ParameterTypeDescription
`db`DatabaseProviderProtocolDatabaseProviderProtocol for query execution.
`max_rows`intHard cap on returned rows.
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the database_query skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Execute the SELECT query and return up to max_rows rows.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``rows`` and ``row_count``, or Err if the query is not a SELECT or execution fails.

Return the current UTC date, time, and Unix timestamp.

Parameters

Section titled “Parameters”

timezone : str, optional IANA timezone name (e.g. "America/New_York"). Defaults to "UTC". Only UTC is supported without the zoneinfo stdlib module (Python 3.9+); unknown timezones fall back to UTC.

Example output

{
  "datetime": "2026-01-15T14:32:00+00:00",
  "date": "2026-01-15",
  "time": "14:32:00",
  "timestamp": 1736951520.0,
  "timezone": "UTC"
}
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the current_datetime skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Execute the skill and return the current datetime.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result containing a datetime dict.

Read a file and return its content as a string.

Files are only accessible within base_dir (defaults to the current working directory). Directory traversal attempts return an error.

Required permission: files.read.

def __init__(base_dir: str | None = None) -> None

Initialise with an optional sandbox directory.

Parameters
ParameterTypeDescription
`base_dir`str | NoneBase directory constraint. Defaults to ``os.getcwd()``.
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the file_read skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Read and return file content.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``content``, ``path``, and ``size_bytes``, or Err.

Write text content to a local file.

Files are only writable within base_dir. The parent directory must exist; write will not create intermediate directories.

Required permission: files.write.

def __init__(base_dir: str | None = None) -> None

Initialise with an optional sandbox directory.

Parameters
ParameterTypeDescription
`base_dir`str | NoneBase directory constraint. Defaults to ``os.getcwd()``.
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the file_write skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Write content to the specified path.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``path`` and ``bytes_written``, or Err.

Wraps an async function decorated with ``@skill`` as a SkillProtocol.

Created automatically by the @skill decorator — not intended for direct instantiation.

def __init__(
    fn: Callable[Ellipsis, Awaitable[Any]],
    definition: SkillDefinition,
    param_names: list[str] | None = None
) -> None

Initialise a FunctionSkill.

Parameters
ParameterTypeDescription
`fn`Callable[Ellipsis, Awaitable[Any]]The underlying async callable.
`definition`SkillDefinitionThe skill's definition metadata.
`param_names`list[str] | NoneOrdered list of declared parameter names from ``@skill_param`` decorators (used for schema building).
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionThe SkillDefinition for this function skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Call the wrapped async function and wrap its return value.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Result wrapping SkillResult on success or SkillError on failure.
def validate(params: dict[str, Any]) -> list[str]

Validate params against the JSON schema in the definition.

Parameters
ParameterTypeDescription
`params`dict[str, Any]Parameters to validate.
Returns
TypeDescription
list[str]List of validation error messages.
def to_tool() -> SkillToolAdapter

Convert this skill to a ToolProtocol for use in Lexigram agents.

Returns
TypeDescription
SkillToolAdapterA SkillToolAdapter that implements ToolProtocol.

Make outbound HTTP requests and return the response body.

By default the skill uses urllib.request (stdlib) to avoid requiring an extra HTTP client dependency. For production workloads integrate with the lexigram-http HTTPClientProtocol.

Required permission: http.request.

property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the http_request skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Execute the HTTP request (stdlib urllib, sync-in-thread).

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``status_code``, ``body``, and ``url``, or Err on network/validation failure.

Import MCP tools as Lexigram skills and export skills as MCP tools.

This bridge allows a skill registry to expose its skills to MCP-aware callers and consume MCP tools as if they were first-class skills.

Note

The MCP integration uses lexigram.ai.mcp when available. This class is a stub that gracefully degrades when the MCP package is not installed.

def __init__(registry: SkillRegistry) -> None

Initialise the bridge with the backing registry.

Parameters
ParameterTypeDescription
`registry`SkillRegistryThe skill registry to import into / export from.
async def import_from_mcp(mcp_client: Any) -> int

Register all tools exposed by an MCP client as skills.

Parameters
ParameterTypeDescription
`mcp_client`AnyAn object implementing MCPClientProtocol (from ``lexigram.ai.mcp``).
Returns
TypeDescription
intNumber of tools successfully imported as skills.
def get_mcp_tool_definitions() -> list[dict[str, Any]]

Export all registered skills as MCP tool definitions.

Returns
TypeDescription
list[dict[str, Any]]List of dicts following the MCP tool definition schema.

Evaluate safe arithmetic expressions.

Supports the operators +, -, *, /, //, %, and **. No functions, names, or string literals are permitted.

Example output

{"result": 1024.0, "expression": "2 ** 10"}
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the math_calculate skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Evaluate the expression and return the result.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``result`` and ``expression`` keys, or Err on invalid expression.

Discover and auto-register skills from Python modules.

The scanner walks every attribute of the target module. Attributes that are instances of BaseSkill (or its subclasses, including FunctionSkill) are registered into the provided SkillRegistry.

Example

scanner = ModuleScanner(container)
await scanner.scan(registry, "myapp.skills.builtin")
def __init__(container: ContainerProtocol | None = None) -> None

Initialise the scanner.

Parameters
ParameterTypeDescription
`container`ContainerProtocol | NoneOptional DI container for resolving class-based skills.
async def scan(
    registry: SkillRegistry,
    module_path: str
) -> int

Import module_path and register all discovered skills.

Parameters
ParameterTypeDescription
`registry`SkillRegistryThe SkillRegistry to register discovered skills into.
`module_path`strDotted import path of the module to scan.
Returns
TypeDescription
intNumber of skills successfully registered.
async def scan_package(
    registry: SkillRegistry,
    package_path: str
) -> int

Recursively scan all modules within a package.

Parameters
ParameterTypeDescription
`registry`SkillRegistryThe SkillRegistry to register discovered skills into.
`package_path`strDotted import path of the package to scan.
Returns
TypeDescription
intTotal number of skills registered across all sub-modules.

Execute multiple skills concurrently and aggregate their outputs.

All skills receive the same params. Errors from individual skills do not abort the fan-out; they are collected and returned as part of the aggregated output dict under an "_errors" key.

Example

parallel = ParallelSkills(["sentiment_analysis", "entity_extraction"])
result = await parallel.execute(executor, {"text": "Hello world"})
combined = result.unwrap().output
# combined == {"sentiment_analysis": {...}, "entity_extraction": {...}}
def __init__(skill_names: list[str]) -> None

Initialise with the skills to run in parallel.

Parameters
ParameterTypeDescription
`skill_names`list[str]Names of skills to execute concurrently.
async def execute(
    executor: Any,
    params: dict[str, Any]
) -> Result[SkillResult, SkillError]

Run all skills concurrently and aggregate results.

Parameters
ParameterTypeDescription
`executor`AnySkillExecutorProtocol instance.
`params`dict[str, Any]Input parameters forwarded to every skill.
Returns
TypeDescription
Result[SkillResult, SkillError]A SkillResult whose ``output`` is a dict mapping each skill name to its output (or an error string if that skill failed).

Grant and verify per-user permission sets for skill execution.

Permissions are plain strings (e.g. "files.read", "db.query"). A user with no registered permissions is treated as having no permissions.

Example

checker = PermissionChecker()
checker.grant("user-123", {"files.read", "web.search"})
allowed = checker.check("user-123", {"files.read"})   # True
denied  = checker.check("user-123", {"db.write"})     # False
def __init__() -> None

Initialise an empty permission store.

def grant(
    user_id: str,
    permissions: set[str]
) -> None

Add permissions for a user.

Parameters
ParameterTypeDescription
`user_id`strThe user identifier.
`permissions`set[str]Set of permission strings to grant.
def revoke(
    user_id: str,
    permissions: set[str]
) -> None

Remove permissions for a user.

Parameters
ParameterTypeDescription
`user_id`strThe user identifier.
`permissions`set[str]Set of permission strings to revoke.
def set_permissions(
    user_id: str,
    permissions: set[str]
) -> None

Replace all permissions for a user with the given set.

Parameters
ParameterTypeDescription
`user_id`strThe user identifier.
`permissions`set[str]Complete replacement permission set.
def get_permissions(user_id: str) -> set[str]

Return the current permission set for a user.

Parameters
ParameterTypeDescription
`user_id`strThe user identifier.
Returns
TypeDescription
set[str]Set of permission strings (may be empty).
def check(
    user_id: str,
    required: set[str]
) -> bool

Check whether a user possesses all required permissions.

Parameters
ParameterTypeDescription
`user_id`strThe user to check.
`required`set[str]Permissions that must all be present.
Returns
TypeDescription
bool``True`` if every required permission is granted; ``False`` otherwise (including when *required* is empty — returns ``True``).

Execute skills sequentially, piping each output into the next input.

Each step is a (skill_name, output_mapping) tuple where output_mapping maps keys from the previous output dict to parameter names for the next skill. An empty mapping passes all keys unchanged.

Example

chain = SkillChain([
    ("web_search", {"results": "documents"}),
    ("text_summarize", {}),
])
result = await chain.execute(executor, {"query": "AI trends 2026"})
def __init__(steps: list[tuple[str, dict[str, str]]]) -> None

Initialise the chain with its steps.

Parameters
ParameterTypeDescription
`steps`list[tuple[str, dict[str, str]]]List of ``(skill_name, output_to_input_mapping)`` tuples.
async def execute(
    executor: Any,
    initial_params: dict[str, Any]
) -> Result[SkillResult, SkillError]

Execute the chain from initial_params.

Parameters
ParameterTypeDescription
`executor`AnySkillExecutorProtocol instance used to run each step.
`initial_params`dict[str, Any]Parameters passed to the first skill.
Returns
TypeDescription
Result[SkillResult, SkillError]Result from the final step, or the first error encountered.

Definition of a skill's interface and behavior.

Attributes: name: Unique skill name description: What the skill does parameters_schema: JSON Schema for parameters returns_schema: JSON Schema for return value category: Skill category for organization requires_confirmation: If True, requires user approval before execution cacheable: Whether results can be cached max_retries: Maximum retry attempts on failure timeout_seconds: Execution timeout in seconds permissions: Required permissions to execute

Emitted when a skill execution completes (success or failure).

Consumed by: audit, analytics, skill performance monitoring.

Payload fired when a skill executor completes a skill call.

Payload fired when a skill executor records a failed skill call.

Executes skills with the full production lifecycle.

Handles skill resolution, permission checking, parameter validation, result caching, retry with exponential backoff, timeout enforcement, and observability logging.

def __init__(
    registry: SkillRegistryProtocol,
    *,
    cache: CacheBackendProtocol | None = None,
    permission_checker: PermissionCheckerProtocol | None = None,
    semaphore: asyncio.Semaphore | None = None
) -> None

Initialise the executor.

Parameters
ParameterTypeDescription
`registry`SkillRegistryProtocolSkillRegistryProtocol implementation.
`cache`CacheBackendProtocol | NoneOptional CacheBackendProtocol for caching deterministic results.
`permission_checker`PermissionCheckerProtocol | NoneOptional PermissionCheckerProtocol for access control.
`semaphore`asyncio.Semaphore | NoneOptional semaphore limiting concurrent executions.
async def execute(
    skill_name: str,
    params: dict[str, Any],
    user_id: str | None = None,
    session_id: str | None = None
) -> Result[SkillResult, SkillError]

Execute a skill with full lifecycle management.

Steps: resolve → check permissions → validate → check cache → execute with retry → store cache → return.

Parameters
ParameterTypeDescription
`skill_name`strRegistered name of the skill to execute.
`params`dict[str, Any]Skill parameters.
`user_id`str | NoneOptional caller identity for permission checks.
`session_id`str | NoneOptional session context (passed through to metadata).
Returns
TypeDescription
Result[SkillResult, SkillError]Result wrapping SkillResult on success or SkillError on failure.

Handles loading bundled files and executing skill scripts.

Features:

  • Reading reference.md, forms.md, and other bundled context files
  • Executing scripts (Python, Shell, JavaScript) with parameters
  • Sandboxed execution with path validation
  • Timeout enforcement
  • Environment variable injection
def __init__(
    sandbox: bool = True,
    timeout_seconds: int = 30,
    max_file_size: int = 1024 * 1024
) -> None

Initialize the loader.

Parameters
ParameterTypeDescription
`sandbox`boolWhether to enable sandboxing (default: True).
`timeout_seconds`intScript execution timeout (default: 30s).
`max_file_size`intMaximum file size for context loading (default: 1MB).
async def load_bundled_file(path: Path) -> str | None

Load a bundled context file with size limit.

async def load_bundled_context(
    skill_dir: Path,
    context_files: list[str]
) -> dict[str, str]

Load multiple bundled context files.

async def execute_script(
    script_path: Path,
    params: dict[str, Any]
) -> dict[str, Any]

Execute a skill script with parameters.

Apply a sequence of skills as transformation stages to the same payload.

Unlike SkillChain which passes the output of one skill as the input of the next, a pipeline enriches a single mutable context dict; each stage may read from and write to that shared context.

Example

pipeline = SkillPipeline()
pipeline.add_stage("normalise_text", output_key="clean")
pipeline.add_stage("extract_entities", output_key="entities")
result = await pipeline.execute(executor, {"raw_text": "..."})
# result.unwrap().output == {"raw_text": ..., "clean": ..., "entities": ...}
def __init__() -> None

Initialise an empty pipeline.

def add_stage(
    skill_name: str,
    output_key: str | None = None
) -> None

Append a stage to the pipeline.

Parameters
ParameterTypeDescription
`skill_name`strSkill to invoke at this stage.
`output_key`str | NoneIf given, the skill's output is stored in the shared context under this key. If ``None`` the output dict is merged into the context (ignored for non-dict outputs).
async def execute(
    executor: Any,
    initial_context: dict[str, Any]
) -> Result[SkillResult, SkillError]

Run the pipeline.

Parameters
ParameterTypeDescription
`executor`AnySkillExecutorProtocol instance.
`initial_context`dict[str, Any]Shared context passed to (and enriched by) each stage.
Returns
TypeDescription
Result[SkillResult, SkillError]A SkillResult whose ``output`` is the final enriched context, or the first error encountered.

Payload fired when a skill registry adds a skill definition.

Central registry for all available skills.

Supports registration, lookup by name, category-based listing, permission-filtered listing, and OpenAI function-calling schema export.

def __init__() -> None

Initialise an empty registry.

def register(skill: SkillProtocol) -> None

Register a skill.

Parameters
ParameterTypeDescription
`skill`SkillProtocolThe skill to register.
Raises
ExceptionDescription
SkillAlreadyRegisteredErrorIf a skill with the same name is already registered.
def register_tool(tool: ToolProtocol) -> None

Wrap tool as a ToolSkillAdapter and register it.

Parameters
ParameterTypeDescription
`tool`ToolProtocolAny object satisfying ToolProtocol.
def get(name: str) -> SkillProtocol | None

Return the skill registered under name, or None.

Parameters
ParameterTypeDescription
`name`strSkill name to look up.
Returns
TypeDescription
SkillProtocol | NoneThe skill or None if not found.
def list_skills(
    category: str | None = None,
    permissions: list[str] | None = None
) -> list[SkillDefinition]

List registered skill definitions with optional filters.

Parameters
ParameterTypeDescription
`category`str | NoneReturn only skills in this category.
`permissions`list[str] | NoneReturn only skills whose required permissions are a subset of the supplied permission set.
Returns
TypeDescription
list[SkillDefinition]Matching skill definitions.
def get_schemas() -> list[dict[str, Any]]

Return OpenAI function-calling compatible schemas for all skills.

Returns
TypeDescription
list[dict[str, Any]]List of schema dicts in OpenAI ``tools`` format.

Result of a skill execution.

Attributes: skill_name: Name of the executed skill success: Whether execution succeeded output: The output value (if successful) error: Error message (if failed) duration_ms: Time taken to execute cached: Whether result came from cache metadata: Additional execution metadata

Two-tier result cache for skill executions.

The first tier is an in-process dict; the second tier is an optional CacheBackendProtocol (e.g. Redis). Lookups check the in-process dict first; on a miss they query the backend and populate the in-process dict.

Parameters
ParameterTypeDescription
`backend`Optional async cache backend for cross-process caching.
`ttl_seconds`Time-to-live for backend entries in seconds.
def __init__(
    backend: CacheBackendProtocol | None = None,
    ttl_seconds: int = 3600
) -> None

Initialise the cache.

Parameters
ParameterTypeDescription
`backend`CacheBackendProtocol | NoneOptional CacheBackendProtocol for distributed caching.
`ttl_seconds`intTTL for backend cache entries.
async def get(
    skill_name: str,
    params: dict[str, Any]
) -> SkillResult | None

Return a cached result for the given skill invocation.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill.
`params`dict[str, Any]Parameters the skill was called with.
Returns
TypeDescription
SkillResult | NoneThe cached SkillResult or ``None`` when not cached.
async def set(
    skill_name: str,
    params: dict[str, Any],
    result: SkillResult
) -> None

Store a skill result in both cache tiers.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill.
`params`dict[str, Any]Parameters the skill was called with.
`result`SkillResultThe SkillResult to cache.
def invalidate(
    skill_name: str,
    params: dict[str, Any]
) -> None

Remove a single entry from the local in-process cache.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill.
`params`dict[str, Any]Parameters identifying the entry to remove.
def clear() -> None

Remove all entries from the local in-process cache.

Route an input to one of several skills depending on runtime conditions.

Routes are evaluated in registration order; the first match wins. An optional fallback skill name is used when no route matches.

Example

router = SkillRouter(fallback="default_handler")
router.add_route(
    skill_name="premium_search",
    condition=lambda p: p.get("tier") == "premium",
)
router.add_route(
    skill_name="basic_search",
    condition=lambda p: True,        # catch-all
)
result = await router.execute(executor, {"query": "...", "tier": "free"})
def __init__(fallback: str | None = None) -> None

Initialise the router.

Parameters
ParameterTypeDescription
`fallback`str | NoneSkill name to use when no route matches. If ``None`` and no route matches, the execution returns an error.
def add_route(
    skill_name: str,
    condition: Condition
) -> None

Register a route.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill to invoke if *condition* returns True.
`condition`ConditionCallable receiving the input ``dict`` returning ``bool``.
async def execute(
    executor: Any,
    params: dict[str, Any]
) -> Result[SkillResult, SkillError]

Dispatch params to the first matching skill.

Parameters
ParameterTypeDescription
`executor`AnySkillExecutorProtocol instance.
`params`dict[str, Any]Input parameters used both for condition evaluation and forwarded to the resolved skill.
Returns
TypeDescription
Result[SkillResult, SkillError]Result from the matched skill, or a SkillRoutingError when no route matches and no fallback is configured.

Discovers and registers SKILL.md-based skills.

Features:

  • YAML frontmatter parsing (name, description, version, author, tags)
  • $ARGUMENTS placeholder substitution
  • Instruction-only skills (skill body as instructions)
  • Executable skills (scripts/main.py)
  • Bundled context files (reference.md, forms.md, etc.)
  • Configurable max depth for nested discovery
def __init__(
    loader: SkillLoader | None = None,
    max_depth: int | None = 5
) -> None

Initialize the scanner.

Parameters
ParameterTypeDescription
`loader`SkillLoader | NoneOptional SkillLoader for executing scripts.
`max_depth`int | NoneMaximum directory depth to scan (None for unlimited).
async def scan(
    registry: SkillRegistry,
    root: str | Path
) -> int

Scan a directory tree for SKILL.md folders and register them.

Parameters
ParameterTypeDescription
`registry`SkillRegistryThe SkillRegistry to register discovered skills.
`root`str | PathRoot directory to scan.
Returns
TypeDescription
intNumber of skills successfully registered.

Configuration for the skills subsystem.

Controls execution defaults, caching behaviour, permission enforcement, auto-discovery settings, and which built-in skills are enabled.

Attributes: name: Logical name used for DI registration keys. default_timeout_seconds: Default execution timeout per skill. max_retries: Default maximum retry attempts for skill execution. max_concurrent_executions: Semaphore cap on concurrent executions. cache_enabled: Whether result caching is globally enabled. cache_ttl_seconds: Default TTL for cached skill results. cache_backend: Which cache backend to use (in_memory, cache). enforce_permissions: Whether permission checks are enforced. auto_discover: Whether to auto-scan packages on boot. scan_packages: Fully-qualified package names to scan for skills. enable_builtin: Whether built-in skills are registered on boot. builtin_skills: Names of built-in skills to register.

Application-level façade for the skills subsystem.

Provides a configure factory that creates a pre-configured SkillsProvider ready for registration with the application container.

Usage

from lexigram.ai.skills import SkillsModule
from lexigram.ai.skills.config import SkillsConfig


@module(
    imports=[SkillsModule.configure(
        SkillsConfig(
            enable_builtin=True,
            builtin_skills=["current_datetime", "math_calculate"],
        )
    )]
)
class AppModule(Module):
    pass
def configure(
    cls,
    config: SkillsConfig | None = None,
    enable_tool_bridge: bool = False,
    **kwargs: Any
) -> DynamicModule

Create a SkillsModule with explicit configuration.

Parameters
ParameterTypeDescription
`config`SkillsConfig | NoneSkillsConfig or ``None`` to use defaults.
`enable_tool_bridge`boolRegister the tool-bridge adapter so agent tool registries can invoke skills as first-class tools. Defaults to ``False``. **kwargs: Additional keyword arguments forwarded to SkillsProvider.
Returns
TypeDescription
DynamicModuleA DynamicModule descriptor.
def stub(
    cls,
    config: SkillsConfig | None = None
) -> DynamicModule

Create a SkillsModule suitable for unit and integration testing.

Uses in-memory or no-op skill implementations with minimal side effects. Built-in skill registration is disabled by default to keep the test container lightweight.

Parameters
ParameterTypeDescription
`config`SkillsConfig | NoneOptional SkillsConfig override. Uses safe test defaults when ``None``.
Returns
TypeDescription
DynamicModuleA DynamicModule descriptor.

Register and bootstrap the skills subsystem. 
def __init__(
    config: SkillsConfig | None = None,
    enable_tool_bridge: bool = False,
    **kwargs: Any
) -> None

Initialise the provider with optional configuration.

Parameters
ParameterTypeDescription
`config`SkillsConfig | NoneSkills configuration. Defaults to ``SkillsConfig()``.
`enable_tool_bridge`boolRegister the tool-bridge adapter so agent tool registries can invoke skills as first-class tools. **kwargs: Reserved for future keyword arguments.
async def register(container: ContainerRegistrarProtocol) -> None

Register all skills services into the container.

async def boot(container: ContainerResolverProtocol) -> None

Register built-in and auto-discovered skills into the registry.

async def shutdown() -> None

Perform any cleanup on shutdown (no-op for the skills provider).

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Health check — always healthy (in-process domain provider).

No external backend to ping.

Parameters
ParameterTypeDescription
`timeout`floatIgnored for in-process providers.
Returns
TypeDescription
HealthCheckResultAlways HEALTHY — no external backend to ping.

Return a simple extractive summary of the provided text.

This built-in provides a deterministic word-count truncation summary. For production use, replace or extend with an LLM-backed implementation.

Example output

{"summary": "First 50 words of the input...", "word_count": 200}
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the text_summarize skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Summarise text by truncating to max_words.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``summary`` and ``word_count`` keys.

Stub skill that echoes text with a language annotation.

Replace with an LLM or translation API backend for real translations.

Example output

{"translated": "...", "source_language": "auto", "target_language": "es"}
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the text_translate skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Return the input text unchanged with metadata (stub).

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok result with ``translated``, ``source_language``, and ``target_language`` keys.

Adapts a ``ToolProtocol`` into a ``SkillProtocol``.

Allows existing tool-based registries to participate in the skill system without modification.

def __init__(tool: ToolProtocol) -> None

Wrap tool as a skill.

Parameters
ParameterTypeDescription
`tool`ToolProtocolToolProtocol-compliant object with name, description, parameters_schema, and execute() method.
property definition() -> SkillDefinition

Return the adapted skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition derived from the wrapped tool.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Delegate to the wrapped tool’s execute method.

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Result wrapping SkillResult on success or SkillError on failure.
def validate(params: dict[str, Any]) -> list[str]

Tools have no built-in validation — always returns empty list.

Parameters
ParameterTypeDescription
`params`dict[str, Any]Ignored.
Returns
TypeDescription
list[str]Empty list.

Search the web and return result snippets.

This is a stub implementation. Integrate a real search API (e.g. Brave, Bing, SerpAPI) by overriding execute.

Example output

{
  "query": "AI trends 2026",
  "results": [],
  "stub": true
}
property definition() -> SkillDefinition

Return the skill definition.

Returns
TypeDescription
SkillDefinitionSkillDefinition for the web_search skill.
async def execute(**kwargs: Any) -> Result[SkillResult, SkillError]

Return an empty result set (stub).

Parameters
ParameterTypeDescription
Returns
TypeDescription
Result[SkillResult, SkillError]Ok stub result with empty ``results`` list and ``stub=True``.

Functions

Section titled “Functions”

def skill(
    name: str,
    description: str,
    *,
    category: str = 'general',
    cacheable: bool = False,
    max_retries: int = 0,
    timeout_seconds: float = 30.0,
    requires_confirmation: bool = False,
    permissions: list[str] | None = None
) -> Callable

Convert an async function into a FunctionSkill.

Parameters
ParameterTypeDescription
`name`strUnique skill name.
`description`strWhat the skill does.
`category`strLogical grouping for the skill.
`cacheable`boolWhether results can be cached.
`max_retries`intRetry attempts on failure (0 = no retry).
`timeout_seconds`floatExecution timeout.
`requires_confirmation`boolIf True, consumers may prompt for confirmation.
`permissions`list[str] | NoneRequired permission strings.
Returns
TypeDescription
CallableDecorator that wraps the function as a FunctionSkill.

def skill_param(
    name: str,
    *,
    type: str = 'string',
    description: str = '',
    required: bool = True,
    default: Any = None,
    enum: list[Any] | None = None,
    min_value: float | None = None,
    max_value: float | None = None,
    max_length: int | None = None
) -> Callable

Declare a parameter on a skill function.

Must be applied before @skill. Multiple @skill_param decorators are accumulated in declaration order.

Parameters
ParameterTypeDescription
`name`strParameter name.
`type`strJSON Schema type (e.g. ``"string"``, ``"integer"``, ``"boolean"``).
`description`strHuman-readable description.
`required`boolWhether the parameter is required.
`default`AnyDefault value when not supplied.
`enum`list[Any] | NoneRestricted set of allowed values.
`min_value`float | NoneNumeric minimum.
`max_value`float | NoneNumeric maximum.
`max_length`int | NoneString maximum length.
Returns
TypeDescription
CallableDecorator that annotates the function with parameter metadata.

def validate_non_empty_string(
    value: Any,
    name: str
) -> list[str]

Validate that value is a non-empty string.

Parameters
ParameterTypeDescription
`value`AnyValue to check.
`name`strParameter name for error messages.
Returns
TypeDescription
list[str]List of validation errors.

def validate_params(
    params: dict[str, Any],
    schema: dict[str, Any]
) -> list[str]

Validate params against a JSON Schema object descriptor.

Supports: required fields, type checking, enum, minimum/maximum, minLength/maxLength, and nested object/array types.

Parameters
ParameterTypeDescription
`params`dict[str, Any]The parameter dict supplied by the caller.
`schema`dict[str, Any]JSON Schema ``object`` definition from SkillDefinition.
Returns
TypeDescription
list[str]List of validation error messages. Empty list means valid.

def validate_positive_int(
    value: Any,
    name: str
) -> list[str]

Validate that value is a positive integer.

Parameters
ParameterTypeDescription
`value`AnyValue to check.
`name`strParameter name for error messages.
Returns
TypeDescription
list[str]List of validation errors.

def validate_range(
    value: Any,
    name: str,
    *,
    minimum: float | None = None,
    maximum: float | None = None
) -> list[str]

Validate that value falls within a numeric range.

Parameters
ParameterTypeDescription
`value`AnyValue to check.
`name`strParameter name for error messages.
`minimum`float | NoneInclusive lower bound.
`maximum`float | NoneInclusive upper bound.
Returns
TypeDescription
list[str]List of validation errors.

Exceptions

Section titled “Exceptions”

Raised when a skill is registered under a name that already exists. 
def __init__(skill_name: str) -> None

Initialise with the duplicate skill name.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill that was already registered.

Base for skill execution errors.

Extended in lexigram-ai-skills with specific failures like skill not found, execution failure, parameter validation, etc.

def __init__(
    message: str = 'Skill error',
    **kwargs: Any
) -> None

Raised when a skill execution fails after all retry attempts. 
def __init__(
    message: str,
    cause: Exception | None = None,
    skill_name: str | None = None
) -> None

Initialise with an error message and optional cause.

Parameters
ParameterTypeDescription
`message`strHuman-readable error description.
`cause`Exception | NoneThe underlying exception, if any.
`skill_name`str | NoneName of the skill that failed (optional).

Raised when a requested skill is not registered. 
def __init__(skill_name: str) -> None

Initialise with the missing skill name.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill that was not found.

Raised when the caller lacks required permissions for a skill. 
def __init__(
    skill_name: str,
    required: list[str]
) -> None

Initialise with permission details.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill that requires permissions.
`required`list[str]Required permission strings.

Raised when a SkillRouter finds no matching route. 
def __init__(message: str = 'No matching route') -> None

Initialise with an optional detail message.

Parameters
ParameterTypeDescription
`message`strHuman-readable description of the routing failure.

Raised when skill execution exceeds the configured timeout. 
def __init__(
    skill_name: str,
    timeout_seconds: float
) -> None

Initialise with timeout context.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill that timed out.
`timeout_seconds`floatThe timeout that was exceeded.

Raised when skill parameter validation fails. 
def __init__(
    skill_name: str,
    errors: list[str]
) -> None

Initialise with validation error details.

Parameters
ParameterTypeDescription
`skill_name`strName of the skill that failed validation.
`errors`list[str]List of human-readable validation error messages.