hope-voice-api/agent/skills_mcp.py

"""Skills MCP server — learned procedures for Hope.

Runs over stdio inside the voice container. The agent attaches it via
MCPServerStdio, so the LLM can recall and save skills during a
conversation. Skills live as plain markdown files in SKILLS_DIR
(default /skills), one file per skill, mounted from the host so they
survive rebuilds and are backed up via git.

Skills are *how to do things* — repeatable procedures distilled from
conversations. Facts and preferences belong in memory, not skills.

Tools:
  - skill_save(name, description, steps) -> save a new skill
  - skill_recall(query)    -> matching skills with their full steps
  - skill_list()           -> all skills (name + description line)
  - skill_update(name, new_content) -> replace an existing skill's content
"""

from __future__ import annotations

import os
import re
from datetime import datetime, timezone

from mcp.server.fastmcp import FastMCP

SKILLS_DIR = os.environ.get("SKILLS_DIR", "/skills")

mcp = FastMCP("skills")


def _slugify(name: str) -> str:
    """Turn a skill name into a safe filename slug."""
    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
    return slug or "misc"


def _read_files() -> list[tuple[str, str]]:
    """Return (filename, content) for every .md file in SKILLS_DIR."""
    files: list[tuple[str, str]] = []
    if not os.path.isdir(SKILLS_DIR):
        return files
    for name in sorted(os.listdir(SKILLS_DIR)):
        if not name.endswith(".md"):
            continue
        path = os.path.join(SKILLS_DIR, name)
        try:
            with open(path, encoding="utf-8") as f:
                files.append((name, f.read()))
        except OSError:
            continue
    return files


def _description_line(content: str) -> str:
    """First non-empty line after the H1 title."""
    lines = content.splitlines()
    started = False
    for line in lines:
        if not started:
            if line.startswith("# "):
                started = True
            continue
        if line.strip():
            return line.strip()
    return "(no description)"


@mcp.tool()
def skill_save(name: str, description: str, steps: str) -> str:
    """Save a learned procedure as a reusable skill.

    Use this after successfully completing a multi-step task or when the
    user teaches you a new procedure, so you can follow it next time. Be
    selective: only save repeatable tasks, not one-off facts (those go in
    memory). Returns a short confirmation.
    """
    name = (name or "").strip()
    description = (description or "").strip()
    steps = (steps or "").strip()
    if not name or not description or not steps:
        return "Nothing saved — a name, a description, and steps are all needed."

    os.makedirs(SKILLS_DIR, exist_ok=True)
    slug = _slugify(name)
    path = os.path.join(SKILLS_DIR, f"{slug}.md")

    stamp = datetime.now(timezone.utc).isoformat()
    with open(path, "w", encoding="utf-8") as f:
        f.write(f"# {name}\n\n{description}\n\n## Steps\n{steps}\n\n---\n")
        f.write(f"Created: {stamp}\nSource: conversation-learned\n")

    return f"Saved skill '{name}'."


@mcp.tool()
def skill_recall(query: str) -> str:
    """Search the skill library for a procedure matching a query.

    Use this when you need to perform a task you may have done before —
    it returns the matching skills with their full steps so you can follow
    them. Returns a message saying nothing was found if there's no match.
    """
    q = (query or "").strip().lower()
    if not q:
        return "I don't have a skill for that."

    words = [w for w in re.split(r"\W+", q) if len(w) > 2]
    files = _read_files()
    if not files:
        return "I don't have a skill for that."

    scored: list[tuple[int, str, str]] = []
    for name, content in files:
        lower = content.lower()
        score = 0
        if q in lower:
            score += len(q)
        for w in words:
            score += lower.count(w)
        if score > 0:
            scored.append((score, name, content))

    if not scored:
        return "I don't have a skill for that."

    scored.sort(key=lambda s: s[0], reverse=True)
    parts = []
    for _, name, content in scored[:3]:
        parts.append(content.strip())
    return "\n\n".join(parts)


@mcp.tool()
def skill_list() -> str:
    """List all saved skills with a brief description of each.

    Returns the skill name and its description line, or a message saying
    no skills exist yet.
    """
    files = _read_files()
    if not files:
        return "No skills saved yet."
    lines = [f"{name.removesuffix('.md')}: {_description_line(content)}" for name, content in files]
    return "\n".join(lines)


@mcp.tool()
def skill_update(name: str, new_content: str) -> str:
    """Replace the content of an existing skill.

    Use this to refine a learned procedure after you've improved it. The
    name must match an existing skill (case-insensitive). Returns a
    confirmation, or an error if no such skill exists.
    """
    name = (name or "").strip()
    new_content = (new_content or "").strip()
    if not name or not new_content:
        return "Nothing updated — both a name and new content are needed."

    slug = _slugify(name)
    path = os.path.join(SKILLS_DIR, f"{slug}.md")
    if not os.path.exists(path):
        # Fall back to matching by the H1 title in case the filename differs.
        for fname, content in _read_files():
            first = next((l for l in content.splitlines() if l.strip()), "")
            if first.lstrip("# ").strip().lower() == name.lower():
                path = os.path.join(SKILLS_DIR, fname)
                break
        else:
            return f"No skill named '{name}' found."

    stamp = datetime.now(timezone.utc).isoformat()
    with open(path, "w", encoding="utf-8") as f:
        f.write(new_content.rstrip() + "\n\n---\n")
        f.write(f"Updated: {stamp}\nSource: conversation-learned\n")

    return f"Updated skill '{name}'."


if __name__ == "__main__":
    mcp.run(transport="stdio")