skills

description	Imported skill skills from langchain
name	skills
signature	d40a671a8330cf2365b13b6fba9f8422bc36e0d8a4ab277ea07be6d3e893f944
source	/a0/tmp/skills_research/langchain/libs/deepagents/deepagents/middleware/skills.py

"""Skills middleware for loading and exposing agent skills to the system prompt.

This module implements Anthropic's agent skills pattern with progressive disclosure, loading skills from backend storage via configurable sources.

Architecture

Skills are loaded from one or more sources - paths in a backend where skills are organized. Sources are loaded in order, with later sources overriding earlier ones when skills have the same name (last one wins). This enables layering: base -> user -> project -> team skills.

The middleware uses backend APIs exclusively (no direct filesystem access), making it portable across different storage backends (filesystem, state, remote storage, etc.).

For StateBackend (ephemeral/in-memory), use a factory function:

SkillsMiddleware(backend=lambda rt: StateBackend(rt), ...)

Skill Structure

Each skill is a directory containing a SKILL.md file with YAML frontmatter:

/skills/user/web-research/
├── SKILL.md          # Required: YAML frontmatter + markdown instructions
└── helper.py         # Optional: supporting files

SKILL.md format:

---
name: web-research
description: Structured approach to conducting thorough web research
license: MIT
---

# Web Research Skill

## When to Use
- User asks you to research a topic
...

Skill Metadata (SkillMetadata)

Parsed from YAML frontmatter per Agent Skills specification:

• name: Skill identifier (max 64 chars, lowercase alphanumeric and hyphens)
• description: What the skill does (max 1024 chars)
• path: Backend path to the SKILL.md file
• Optional: license, compatibility, metadata, allowed_tools

Sources

Sources are simply paths to skill directories in the backend. The source name is derived from the last component of the path (e.g., "/skills/user/" -> "user").

Example sources:

[
    "/skills/user/",
    "/skills/project/"
]

Path Conventions

All paths use POSIX conventions (forward slashes) via PurePosixPath:

• Backend paths: "/skills/user/web-research/SKILL.md"
• Virtual, platform-independent
• Backends handle platform-specific conversions as needed

Usage

from deepagents.backends.state import StateBackend
from deepagents.middleware.skills import SkillsMiddleware

middleware = SkillsMiddleware(
    backend=my_backend,
    sources=[
        "/skills/base/",
        "/skills/user/",
        "/skills/project/",
    ],
)

"""

from future import annotations

import logging import re from pathlib import PurePosixPath from typing import TYPE_CHECKING, Annotated

import yaml from langchain.agents.middleware.types import PrivateStateAttr

if TYPE_CHECKING: from deepagents.backends.protocol import BACKEND_TYPES, BackendProtocol

from collections.abc import Awaitable, Callable from typing import NotRequired, TypedDict

from langchain.agents.middleware.types import ( AgentMiddleware, AgentState, ModelRequest, ModelResponse, ) from langchain_core.runnables import RunnableConfig from langgraph.prebuilt import ToolRuntime from langgraph.runtime import Runtime

from deepagents.middleware._utils import append_to_system_message

logger = logging.getLogger(name)

Security: Maximum size for SKILL.md files to prevent DoS attacks (10MB)

MAX_SKILL_FILE_SIZE = 10 * 1024 * 1024

Agent Skills specification constraints (https://agentskills.io/specification)

MAX_SKILL_NAME_LENGTH = 64 MAX_SKILL_DESCRIPTION_LENGTH = 1024

class SkillMetadata(TypedDict): """Metadata for a skill per Agent Skills specification (https://agentskills.io/specification)."""

name: str
"""Skill identifier (max 64 chars, lowercase alphanumeric and hyphens)."""

description: str
"""What the skill does (max 1024 chars)."""

path: str
"""Path to the SKILL.md file."""

license: str | None
"""License name or reference to bundled license file."""

compatibility: str | None
"""Environment requirements (max 500 chars)."""

metadata: dict[str, str]
"""Arbitrary key-value mapping for additional metadata."""

allowed_tools: list[str]
"""Space-delimited list of pre-approved tools. (Experimental)"""

class SkillsState(AgentState): """State for the skills middleware."""

skills_metadata: NotRequired[Annotated[list[SkillMetadata], PrivateStateAttr]]
"""List of loaded skill metadata from all configured sources."""

class SkillsStateUpdate(TypedDict): """State update for the skills middleware."""

skills_metadata: list[SkillMetadata]
"""List of loaded skill metadata to merge into state."""

def _validate_skill_name(name: str, directory_name: str) -> tuple[bool, str]: """Validate skill name per Agent Skills specification.

Requirements per spec:
- Max 64 characters
- Lowercase alphanumeric and hyphens only (a-z, 0-9, -)
- Cannot start or end with hyphen
- No consecutive hyphens
- Must match parent directory name

Args:
    name: Skill name from YAML frontmatter
    directory_name: Parent directory name

Returns:
    (is_valid, error_message) tuple. Error message is empty if valid.
"""
if not name:
    return False, "name is required"
if len(name) > MAX_SKILL_NAME_LENGTH:
    return False, "name exceeds 64 characters"
# Pattern: lowercase alphanumeric, single hyphens between segments, no start/end hyphen
if not re.match(r"^[a-z0-9]+(-[a-z0-9]+)*$", name):
    return False, "name must be lowercase alphanumeric with single hyphens only"
if name != directory_name:
    return False, f"name '{name}' must match directory name '{directory_name}'"
return True, ""

def _parse_skill_metadata( content: str, skill_path: str, directory_name: str, ) -> SkillMetadata | None: """Parse YAML frontmatter from SKILL.md content.

Extracts metadata per Agent Skills specification from YAML frontmatter delimited
by --- markers at the start of the content.

Args:
    content: Content of the SKILL.md file
    skill_path: Path to the SKILL.md file (for error messages and metadata)
    directory_name: Name of the parent directory containing the skill

Returns:
    SkillMetadata if parsing succeeds, None if parsing fails or validation errors occur
"""
if len(content) > MAX_SKILL_FILE_SIZE:
    logger.warning("Skipping %s: content too large (%d bytes)", skill_path, len(content))
    return None

# Match YAML frontmatter between --- delimiters
frontmatter_pattern = r"^---\s*\n(.*?)\n---\s*\n"
match = re.match(frontmatter_pattern, content, re.DOTALL)

if not match:
    logger.warning("Skipping %s: no valid YAML frontmatter found", skill_path)
    return None

frontmatter_str = match.group(1)

# Parse YAML using safe_load for proper nested structure support
try:
    frontmatter_data = yaml.safe_load(frontmatter_str)
except yaml.YAMLError as e:
    logger.warning("Invalid YAML in %s: %s", skill_path, e)
    return None

if not isinstance(frontmatter_data, dict):
    logger.warning("Skipping %s: frontmatter is not a mapping", skill_path)
    return None

# Validate required fields
name = frontmatter_data.get("name")
description = frontmatter_data.get("description")

if not name or not description:
    logger.warning("Skipping %s: missing required 'name' or 'description'", skill_path)
    return None

# Validate name format per spec (warn but continue loading for backwards compatibility)
is_valid, error = _validate_skill_name(str(name), directory_name)
if not is_valid:
    logger.warning(
        "Skill '%s' in %s does not follow Agent Skills specification: %s. Consider renaming for spec compliance.",
        name,
        skill_path,
        error,
    )

# Validate description length per spec (max 1024 chars)
description_str = str(description).strip()
if len(description_str) > MAX_SKILL_DESCRIPTION_LENGTH:
    logger.warning(
        "Description exceeds %d characters in %s, truncating",
        MAX_SKILL_DESCRIPTION_LENGTH,
        skill_path,
    )
    description_str = description_str[:MAX_SKILL_DESCRIPTION_LENGTH]

if frontmatter_data.get("allowed-tools"):
    allowed_tools = frontmatter_data.get("allowed-tools").split(" ")
else:
    allowed_tools = []

return SkillMetadata(
    name=str(name),
    description=description_str,
    path=skill_path,
    metadata=frontmatter_data.get("metadata", {}),
    license=frontmatter_data.get("license", "").strip() or None,
    compatibility=frontmatter_data.get("compatibility", "").strip() or None,
    allowed_tools=allowed_tools,
)

def _list_skills(backend: BackendProtocol, source_path: str) -> list[SkillMetadata]: """List all skills from a backend source.

Scans backend for subdirectories containing SKILL.md files, downloads their content,
parses YAML frontmatter, and returns skill metadata.

Expected structure:
    source_path/
    ├── skill-name/
    │   ├── SKILL.md        # Required
    │   └── helper.py       # Optional

Args:
    backend: Backend instance to use for file operations
    source_path: Path to the skills directory in the backend

Returns:
    List of skill metadata from successfully parsed SKILL.md files
"""
base_path = source_path

skills: list[SkillMetadata] = []
items = backend.ls_info(base_path)
# Find all skill directories (directories containing SKILL.md)
skill_dirs = []
for item in items:
    if not item.get("is_dir"):
        continue
    skill_dirs.append(item["path"])

if not skill_dirs:
    return []

# For each skill directory, check if SKILL.md exists and download it
skill_md_paths = []
for skill_dir_path in skill_dirs:
    # Construct SKILL.md path using PurePosixPath for safe, standardized path operations
    skill_dir = PurePosixPath(skill_dir_path)
    skill_md_path = str(skill_dir / "SKILL.md")
    skill_md_paths.append((skill_dir_path, skill_md_path))

paths_to_download = [skill_md_path for _, skill_md_path in skill_md_paths]
responses = backend.download_files(paths_to_download)

# Parse each downloaded SKILL.md
for (skill_dir_path, skill_md_path), response in zip(skill_md_paths, responses, strict=True):
    if response.error:
        # Skill doesn't have a SKILL.md, skip it
        continue

    if response.content is None:
        logger.warning("Downloaded skill file %s has no content", skill_md_path)
        continue

    try:
        content = response.content.decode("utf-8")
    except UnicodeDecodeError as e:
        logger.warning("Error decoding %s: %s", skill_md_path, e)
        continue

    # Extract directory name from path using PurePosixPath
    directory_name = PurePosixPath(skill_dir_path).name

    # Parse metadata
    skill_metadata = _parse_skill_metadata(
        content=content,
        skill_path=skill_md_path,
        directory_name=directory_name,
    )
    if skill_metadata:
        skills.append(skill_metadata)

return skills

async def _alist_skills(backend: BackendProtocol, source_path: str) -> list[SkillMetadata]: """List all skills from a backend source (async version).

Scans backend for subdirectories containing SKILL.md files, downloads their content,
parses YAML frontmatter, and returns skill metadata.

Expected structure:
    source_path/
    ├── skill-name/
    │   ├── SKILL.md        # Required
    │   └── helper.py       # Optional

Args:
    backend: Backend instance to use for file operations
    source_path: Path to the skills directory in the backend

Returns:
    List of skill metadata from successfully parsed SKILL.md files
"""
base_path = source_path

skills: list[SkillMetadata] = []
items = await backend.als_info(base_path)
# Find all skill directories (directories containing SKILL.md)
skill_dirs = []
for item in items:
    if not item.get("is_dir"):
        continue
    skill_dirs.append(item["path"])

if not skill_dirs:
    return []

# For each skill directory, check if SKILL.md exists and download it
skill_md_paths = []
for skill_dir_path in skill_dirs:
    # Construct SKILL.md path using PurePosixPath for safe, standardized path operations
    skill_dir = PurePosixPath(skill_dir_path)
    skill_md_path = str(skill_dir / "SKILL.md")
    skill_md_paths.append((skill_dir_path, skill_md_path))

paths_to_download = [skill_md_path for _, skill_md_path in skill_md_paths]
responses = await backend.adownload_files(paths_to_download)

# Parse each downloaded SKILL.md
for (skill_dir_path, skill_md_path), response in zip(skill_md_paths, responses, strict=True):
    if response.error:
        # Skill doesn't have a SKILL.md, skip it
        continue

    if response.content is None:
        logger.warning("Downloaded skill file %s has no content", skill_md_path)
        continue

    try:
        content = response.content.decode("utf-8")
    except UnicodeDecodeError as e:
        logger.warning("Error decoding %s: %s", skill_md_path, e)
        continue

    # Extract directory name from path using PurePosixPath
    directory_name = PurePosixPath(skill_dir_path).name

    # Parse metadata
    skill_metadata = _parse_skill_metadata(
        content=content,
        skill_path=skill_md_path,
        directory_name=directory_name,
    )
    if skill_metadata:
        skills.append(skill_metadata)

return skills

SKILLS_SYSTEM_PROMPT = """

Skills System

You have access to a skills library that provides specialized capabilities and domain knowledge.

{skills_locations}

Available Skills:

{skills_list}

How to Use Skills (Progressive Disclosure):

Skills follow a progressive disclosure pattern - you see their name and description above, but only read full instructions when needed:

1. Recognize when a skill applies: Check if the user's task matches a skill's description
2. Read the skill's full instructions: Use the path shown in the skill list above
3. Follow the skill's instructions: SKILL.md contains step-by-step workflows, best practices, and examples
4. Access supporting files: Skills may include helper scripts, configs, or reference docs - use absolute paths

When to Use Skills:

• User's request matches a skill's domain (e.g., "research X" -> web-research skill)
• You need specialized knowledge or structured workflows
• A skill provides proven patterns for complex tasks

Executing Skill Scripts: Skills may contain Python scripts or other executable files. Always use absolute paths from the skill list.

Example Workflow:

User: "Can you research the latest developments in quantum computing?"

1. Check available skills -> See "web-research" skill with its path
2. Read the skill using the path shown
3. Follow the skill's research workflow (search -> organize -> synthesize)
4. Use any helper scripts with absolute paths

Remember: Skills make you more capable and consistent. When in doubt, check if a skill exists for the task! """

class SkillsMiddleware(AgentMiddleware): """Middleware for loading and exposing agent skills to the system prompt.

Loads skills from backend sources and injects them into the system prompt
using progressive disclosure (metadata first, full content on demand).

Skills are loaded in source order with later sources overriding earlier ones.

Example:
    ```python
    from deepagents.backends.filesystem import FilesystemBackend

    backend = FilesystemBackend(root_dir="/path/to/skills")
    middleware = SkillsMiddleware(
        backend=backend,
        sources=[
            "/path/to/skills/user/",
            "/path/to/skills/project/",
        ],
    )
    ```

Args:
    backend: Backend instance for file operations
    sources: List of skill source paths. Source names are derived from the last path component.
"""

state_schema = SkillsState

def __init__(self, *, backend: BACKEND_TYPES, sources: list[str]) -> None:
    """Initialize the skills middleware.

    Args:
        backend: Backend instance or factory function that takes runtime and returns a backend.
                 Use a factory for StateBackend: `lambda rt: StateBackend(rt)`
        sources: List of skill source paths (e.g., ["/skills/user/", "/skills/project/"]).
    """
    self._backend = backend
    self.sources = sources
    self.system_prompt_template = SKILLS_SYSTEM_PROMPT

def _get_backend(self, state: SkillsState, runtime: Runtime, config: RunnableConfig) -> BackendProtocol:
    """Resolve backend from instance or factory.

    Args:
        state: Current agent state.
        runtime: Runtime context for factory functions.
        config: Runnable config to pass to backend factory.

    Returns:
        Resolved backend instance
    """
    if callable(self._backend):
        # Construct an artificial tool runtime to resolve backend factory
        tool_runtime = ToolRuntime(
            state=state,
            context=runtime.context,
            stream_writer=runtime.stream_writer,
            store=runtime.store,
            config=config,
            tool_call_id=None,
        )
        backend = self._backend(tool_runtime)
        if backend is None:
            raise AssertionError("SkillsMiddleware requires a valid backend instance")
        return backend

    return self._backend

def _format_skills_locations(self) -> str:
    """Format skills locations for display in system prompt."""
    locations = []
    for i, source_path in enumerate(self.sources):
        name = PurePosixPath(source_path.rstrip("/")).name.capitalize()
        suffix = " (higher priority)" if i == len(self.sources) - 1 else ""
        locations.append(f"**{name} Skills**: `{source_path}`{suffix}")
    return "\n".join(locations)

def _format_skills_list(self, skills: list[SkillMetadata]) -> str:
    """Format skills metadata for display in system prompt."""
    if not skills:
        paths = [f"{source_path}" for source_path in self.sources]
        return f"(No skills available yet. You can create skills in {' or '.join(paths)})"

    lines = []
    for skill in skills:
        lines.append(f"- **{skill['name']}**: {skill['description']}")
        lines.append(f"  -> Read `{skill['path']}` for full instructions")

    return "\n".join(lines)

def modify_request(self, request: ModelRequest) -> ModelRequest:
    """Inject skills documentation into a model request's system message.

    Args:
        request: Model request to modify

    Returns:
        New model request with skills documentation injected into system message
    """
    skills_metadata = request.state.get("skills_metadata", [])
    skills_locations = self._format_skills_locations()
    skills_list = self._format_skills_list(skills_metadata)

    skills_section = self.system_prompt_template.format(
        skills_locations=skills_locations,
        skills_list=skills_list,
    )

    new_system_message = append_to_system_message(request.system_message, skills_section)

    return request.override(system_message=new_system_message)

def before_agent(self, state: SkillsState, runtime: Runtime, config: RunnableConfig) -> SkillsStateUpdate | None:
    """Load skills metadata before agent execution (synchronous).

    Runs before each agent interaction to discover available skills from all
    configured sources. Re-loads on every call to capture any changes.

    Skills are loaded in source order with later sources overriding
    earlier ones if they contain skills with the same name (last one wins).

    Args:
        state: Current agent state.
        runtime: Runtime context.
        config: Runnable config.

    Returns:
        State update with `skills_metadata` populated, or `None` if already present
    """
    # Skip if skills_metadata is already present in state (even if empty)
    if "skills_metadata" in state:
        return None

    # Resolve backend (supports both direct instances and factory functions)
    backend = self._get_backend(state, runtime, config)
    all_skills: dict[str, SkillMetadata] = {}

    # Load skills from each source in order
    # Later sources override earlier ones (last one wins)
    for source_path in self.sources:
        source_skills = _list_skills(backend, source_path)
        for skill in source_skills:
            all_skills[skill["name"]] = skill

    skills = list(all_skills.values())
    return SkillsStateUpdate(skills_metadata=skills)

async def abefore_agent(self, state: SkillsState, runtime: Runtime, config: RunnableConfig) -> SkillsStateUpdate | None:
    """Load skills metadata before agent execution (async).

    Runs before each agent interaction to discover available skills from all
    configured sources. Re-loads on every call to capture any changes.

    Skills are loaded in source order with later sources overriding
    earlier ones if they contain skills with the same name (last one wins).

    Args:
        state: Current agent state.
        runtime: Runtime context.
        config: Runnable config.

    Returns:
        State update with `skills_metadata` populated, or `None` if already present
    """
    # Skip if skills_metadata is already present in state (even if empty)
    if "skills_metadata" in state:
        return None

    # Resolve backend (supports both direct instances and factory functions)
    backend = self._get_backend(state, runtime, config)
    all_skills: dict[str, SkillMetadata] = {}

    # Load skills from each source in order
    # Later sources override earlier ones (last one wins)
    for source_path in self.sources:
        source_skills = await _alist_skills(backend, source_path)
        for skill in source_skills:
            all_skills[skill["name"]] = skill

    skills = list(all_skills.values())
    return SkillsStateUpdate(skills_metadata=skills)

def wrap_model_call(
    self,
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Inject skills documentation into the system prompt.

    Args:
        request: Model request being processed
        handler: Handler function to call with modified request

    Returns:
        Model response from handler
    """
    modified_request = self.modify_request(request)
    return handler(modified_request)

async def awrap_model_call(
    self,
    request: ModelRequest,
    handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
) -> ModelResponse:
    """Inject skills documentation into the system prompt (async version).

    Args:
        request: Model request being processed
        handler: Async handler function to call with modified request

    Returns:
        Model response from handler
    """
    modified_request = self.modify_request(request)
    return await handler(modified_request)

all = ["SkillMetadata", "SkillsMiddleware"]