Spaces:
Running
Running
| """ | |
| NeuraPrompt — Skill tools (list / load / save) | |
| ============================================= | |
| Sync functions; agent wraps them with asyncio.to_thread via the registry. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from typing import Optional | |
| log = logging.getLogger("agent.tools.skills") | |
| def _builtins(): | |
| try: | |
| from ..skills_catalog import BUILTIN_SKILLS, get_builtin | |
| except ImportError: | |
| from skills_catalog import BUILTIN_SKILLS, get_builtin # type: ignore | |
| return BUILTIN_SKILLS, get_builtin | |
| def _memory(): | |
| try: | |
| from .. import agent_memory as mem | |
| except ImportError: | |
| import agent_memory as mem # type: ignore | |
| return mem | |
| def list_skills(user_id: str = "") -> str: | |
| """List builtin + user skill names with when_to_use hints.""" | |
| BUILTIN_SKILLS, _ = _builtins() | |
| lines = ["## Builtin skills"] | |
| for sk in BUILTIN_SKILLS: | |
| lines.append(f"- **{sk['name']}**: {sk.get('when_to_use', '')}") | |
| if user_id: | |
| mem = _memory() | |
| try: | |
| # Prefer richer listing if available | |
| db = mem._get_db() if hasattr(mem, "_get_db") else None | |
| if db is not None: | |
| skills = list( | |
| db.skills.find( | |
| {"user_id": user_id}, | |
| {"skill_name": 1, "when_to_use": 1, "skill_description": 1, "_id": 0}, | |
| ) | |
| ) | |
| if skills: | |
| lines.append("\n## Your learned skills") | |
| for s in skills[:20]: | |
| name = s.get("skill_name", "?") | |
| when = s.get("when_to_use") or s.get("skill_description", "")[:120] | |
| lines.append(f"- **{name}**: {when}") | |
| else: | |
| names = mem.get_skills_list(user_id) or [] | |
| if names: | |
| lines.append("\n## Your learned skills") | |
| for n in names: | |
| lines.append(f"- **{n}**") | |
| except Exception as e: | |
| log.warning(f"[skill_tools] list user skills failed: {e}") | |
| return "\n".join(lines) | |
| def load_skill(name: str, user_id: str = "") -> str: | |
| """Load full skill body: builtins first, then Mongo user skills.""" | |
| if not name or not str(name).strip(): | |
| return "Error: skill name is required." | |
| _, get_builtin = _builtins() | |
| builtin = get_builtin(name) | |
| if builtin: | |
| return ( | |
| f"# Skill: {builtin['name']}\n" | |
| f"**When to use:** {builtin.get('when_to_use', '')}\n\n" | |
| f"{builtin.get('body', '')}" | |
| ) | |
| if user_id: | |
| mem = _memory() | |
| try: | |
| detail = None | |
| if hasattr(mem, "get_skill_detail"): | |
| detail = mem.get_skill_detail(user_id, name) | |
| if detail: | |
| return f"# Skill: {name}\n\n{detail}" | |
| except Exception as e: | |
| log.warning(f"[skill_tools] load user skill failed: {e}") | |
| return f"Skill '{name}' not found. Use list_skills to see available skills." | |
| def save_user_skill( | |
| user_id: str, | |
| name: str, | |
| when_to_use: str = "", | |
| body: str = "", | |
| ) -> str: | |
| """Save / update a user skill with richer fields.""" | |
| if not user_id: | |
| return "Error: user_id is required to save a skill." | |
| if not name or not name.strip(): | |
| return "Error: skill name is required." | |
| mem = _memory() | |
| try: | |
| if hasattr(mem, "save_skill"): | |
| # Prefer enhanced signature | |
| try: | |
| is_new = mem.save_skill( | |
| user_id, | |
| name.strip(), | |
| body or when_to_use or name, | |
| when_to_use=when_to_use or "", | |
| skill_body=body or "", | |
| ) | |
| except TypeError: | |
| is_new = mem.save_skill( | |
| user_id, | |
| name.strip(), | |
| body or when_to_use or name, | |
| ) | |
| if is_new: | |
| return f"Saved new skill '{name}'." | |
| return f"Updated skill '{name}'." | |
| return "Error: skill storage unavailable." | |
| except Exception as e: | |
| log.warning(f"[skill_tools] save_user_skill failed: {e}") | |
| return f"Error saving skill: {e}" | |
| __all__ = ["list_skills", "load_skill", "save_user_skill"] | |