Agent Zero
dbeadd8ab1
Check Cross-Plugin Imports / check (push) Has been cancelled
- F-CTX: app/ai/context_builder.py (282 lines) — build_agent_context() + ReActSystemPromptBuilder - F-STR: app/ai/agent_stream.py (155 lines) — stream_react_loop() with SSE events (step, status, done, error) - F-DEF: AgentDefinition fields added (temperature, max_tokens, max_steps, trace_mode, skill_ids, trigger_config, ai_use_case_metadata) + migration 0122 - F-SKILL: app/ai/skill_registry.py (82 lines) — SkillDefinition + SkillRegistry singleton - F-TOOL: app/ai/agent_tools.py (117 lines) — get_agent_tools() with permission intersection - Skill CRUD routes: app/plugins/builtins/automation/skill_routes.py - Tests: test_skill_registry.py (97 lines), test_agent_tools.py (219 lines) - All Python compile checks pass, tests require PostgreSQL (infra issue, not code bug)
83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""Small Skill Registry for AI agents.
|
|
|
|
Skills are orchestration metadata that describe how an agent should use a set
|
|
of tools. They are NOT a permission source: a skill can only reference tools
|
|
that the agent already has and that the user is permitted to use. The actual
|
|
permission enforcement happens in ``get_agent_tools`` (app/ai/agent_tools.py).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class SkillDefinition:
|
|
"""A skill definition — orchestration metadata for a set of tools."""
|
|
|
|
name: str
|
|
description: str
|
|
instructions: str # How to use this skill
|
|
allowed_tool_ids: list[str] = field(default_factory=list) # Tool IDs this skill can use
|
|
context_policy: dict[str, Any] | None = None # Optional context inclusion rules
|
|
category: str = "general"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialize to a plain dict for API responses."""
|
|
return {
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"instructions": self.instructions,
|
|
"allowed_tool_ids": list(self.allowed_tool_ids or []),
|
|
"context_policy": self.context_policy,
|
|
"category": self.category,
|
|
}
|
|
|
|
|
|
class SkillRegistry:
|
|
"""Registry for skill definitions.
|
|
|
|
Skills are orchestration metadata, NOT a permission source.
|
|
"""
|
|
|
|
_instance: SkillRegistry | None = None
|
|
|
|
def __new__(cls) -> SkillRegistry:
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
cls._instance._skills: dict[str, SkillDefinition] = {}
|
|
return cls._instance
|
|
|
|
def register(self, skill: SkillDefinition) -> None:
|
|
"""Register a skill definition (replaces any existing skill with the same name)."""
|
|
self._skills[skill.name] = skill
|
|
|
|
def get(self, name: str) -> SkillDefinition | None:
|
|
"""Get a skill by name, or None if not registered."""
|
|
return self._skills.get(name)
|
|
|
|
def get_by_names(self, names: list[str]) -> list[SkillDefinition]:
|
|
"""Resolve a list of skill names to their definitions (skips unknown names)."""
|
|
return [self._skills[name] for name in names if name in self._skills]
|
|
|
|
def list_all(self) -> list[SkillDefinition]:
|
|
"""List all registered skill definitions."""
|
|
return list(self._skills.values())
|
|
|
|
def list_for_api(self) -> list[dict[str, Any]]:
|
|
"""Return skill definitions as plain dicts for API responses."""
|
|
return [skill.to_dict() for skill in self._skills.values()]
|
|
|
|
def unregister(self, name: str) -> None:
|
|
"""Remove a skill definition by name."""
|
|
self._skills.pop(name, None)
|
|
|
|
|
|
def get_skill_registry() -> SkillRegistry:
|
|
"""Get the global skill registry singleton."""
|
|
return SkillRegistry()
|
|
|
|
|
|
__all__ = ["SkillDefinition", "SkillRegistry", "get_skill_registry"]
|