Agent Skills (SEP-2640) & SkillController¶

"""Snippet: serving Agent Skills over MCP from a directory."""

from pathlib import Path

from litestar import Litestar

from litestar_mcp import LitestarMCP, MCPConfig, MCPSkillsConfig

SKILLS_ROOT = Path(__file__).resolve().parents[1] / "skills"


def build() -> "Litestar":
    # start-example
    config = MCPConfig(skills=MCPSkillsConfig(paths=[SKILLS_ROOT]))
    app = Litestar(plugins=[LitestarMCP(config)])
    # end-example
    return app

"""Dynamic, request-aware MCP skills example.

Demonstrates composing a static ``FilesystemSkillSource`` with a dynamic
``WorkspaceSkillSource`` that constructs skills in memory with ``Skill.build``,
filters visibility by the incoming ``X-Workspace-Group`` request header, and
lazily serves a secondary ``runbook.md`` file via ``read_file``.
"""

# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "litestar[standard]>=2.0",
#   "litestar-mcp",
#   "uvicorn",
# ]
# ///

from dataclasses import replace
from pathlib import Path
from typing import Any

from litestar import Litestar, Request

from litestar_mcp import (
    LitestarMCP,
    MCPConfig,
    MCPSkillsConfig,
    Skill,
    SkillFile,
)

SKILLS_ROOT = Path(__file__).resolve().parents[1] / "skills"


class WorkspaceSkillSource:
    """In-memory skill source scoped by the ``X-Workspace-Group`` header."""

    def __init__(self) -> "None":
        self._runbook_bytes = (
            b"# Incident Runbook\n\n"
            b"1. Verify service health indicators.\n"
            b"2. Check deployment rollouts for the active workspace.\n"
        )
        lazy_runbook = replace(
            SkillFile.from_content("platform-runbook", "runbook.md", self._runbook_bytes),
            content=None,
        )
        self._skills: dict[str, tuple[str, Skill]] = {
            "platform-runbook": (
                "platform",
                Skill.build(
                    name="platform-runbook",
                    description="Operational runbook for the platform engineering group.",
                    body="Follow the attached runbook when triaging workspace incidents.",
                    files={"runbook.md": lazy_runbook},
                ),
            ),
            "analytics-playbook": (
                "analytics",
                Skill.build(
                    name="analytics-playbook",
                    description="Query optimization playbook for the analytics group.",
                    body="Inspect partition pruning before escalating slow warehouse jobs.",
                ),
            ),
        }

    async def list_skills(self, request: "Request[Any, Any, Any]") -> "list[Skill]":
        """Return only the skills matching the caller's workspace group."""
        group = self._caller_group(request)
        return [skill for required_group, skill in self._skills.values() if required_group == group]

    async def get_skill(
        self,
        request: "Request[Any, Any, Any]",
        name: "str",
    ) -> "Skill | None":
        """Resolve a single skill by name when permitted for the caller's group."""
        entry = self._skills.get(name)
        if entry is None:
            return None
        required_group, skill = entry
        if required_group != self._caller_group(request):
            return None
        return skill

    async def read_file(
        self,
        request: "Request[Any, Any, Any]",
        skill: "Skill",
        file: "SkillFile",
    ) -> "bytes":
        """Serve ``SKILL.md`` from memory and ``runbook.md`` on demand."""
        del request
        if file.content is not None:
            return file.content
        if skill.name == "platform-runbook" and file.relative_path == "runbook.md":
            return self._runbook_bytes
        raise FileNotFoundError(file.relative_path)

    def _caller_group(self, request: "Request[Any, Any, Any] | None") -> "str":
        if request is None:
            return "public"
        return request.headers.get("x-workspace-group", "public").strip().lower()


def build_app() -> "Litestar":
    """Construct a Litestar app combining filesystem and dynamic skill sources."""
    mcp_config = MCPConfig(
        name="Dynamic Skills API",
        skills=MCPSkillsConfig(paths=[SKILLS_ROOT], sources=[WorkspaceSkillSource()]),
    )
    return Litestar(plugins=[LitestarMCP(mcp_config)])


app = build_app()

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

class BaristaSkillController(SkillController):
    """Barista skill providing beverage menu lookup and order preparation tools."""

    name = "barista"
    description = "Specialized skills for querying menu items and preparing coffee orders"
    instructions = (
        "Barista Operational Grounding:\n"
        "- Suggest seasonal single-origin roasts when asked for recommendations.\n"
        "- Confirm milk alternatives (oat, almond, soy) before finalizing espresso drinks.\n"
        "- Never accept orders for items outside the published menu.\n"
    )

    def __init__(self, owner: Any = None) -> None:
        """Initialize the barista skill with catalog menu and empty order list."""
        super().__init__(owner)
        self.menu: dict[str, dict[str, Any]] = {
            "espresso": {"price": 3.50, "roast": "Ethiopian Yirgacheffe", "decaf_available": True},
            "flat_white": {"price": 4.75, "roast": "Guatemalan Antigua", "decaf_available": True},
            "cold_brew": {"price": 5.00, "roast": "Colombian Huila", "decaf_available": False},
        }
        self.orders: list[dict[str, Any]] = []

    @tool
    def get_menu(self) -> dict[str, dict[str, Any]]:
        """Retrieve available coffee drinks and pricing."""
        return self.menu

    @tool
    def place_order(self, beverage: str, milk: str = "whole", shots: int = 2) -> dict[str, Any]:
        """Place a customized beverage order."""
        bev = beverage.lower().replace(" ", "_")
        if bev not in self.menu:
            return {"status": "error", "message": f"'{beverage}' is not on the menu."}

        order_record = {
            "order_id": len(self.orders) + 1,
            "beverage": bev,
            "milk": milk,
            "shots": shots,
            "price": self.menu[bev]["price"],
            "status": "in_queue",
        }
        self.orders.append(order_record)
        return {"status": "success", "order": order_record}