SkillsMiddleware

SkillsMiddleware

Bases: 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

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/",
    ],
)

PARAMETER	DESCRIPTION
backend	Backend instance for file operations TYPE: BACKEND_TYPES
sources	List of skill source paths. Source names are derived from the last path component. TYPE: list[str]

METHOD	DESCRIPTION
__init__	Initialize the skills middleware.
modify_request	Inject skills documentation into a model request's system message.
before_agent	Load skills metadata before agent execution (synchronous).
abefore_agent	Load skills metadata before agent execution (async).
wrap_model_call	Inject skills documentation into the system prompt.
awrap_model_call	Inject skills documentation into the system prompt (async version).
before_model	Logic to run before the model is called.
abefore_model	Async logic to run before the model is called.
after_model	Logic to run after the model is called.
aafter_model	Async logic to run after the model is called.
after_agent	Logic to run after the agent execution completes.
aafter_agent	Async logic to run after the agent execution completes.
wrap_tool_call	Intercept tool execution for retries, monitoring, or modification.
awrap_tool_call	Intercept and control async tool execution via handler callback.

state_schema class-attribute instance-attribute

state_schema = SkillsState

The schema for state passed to the middleware nodes.

tools instance-attribute

tools: Sequence[BaseTool]

Additional tools registered by the middleware.

name property

name: str

The name of the middleware instance.

Defaults to the class name, but can be overridden for custom naming.

__init__

__init__(*, backend: BACKEND_TYPES, sources: list[str]) -> None

Initialize the skills middleware.

PARAMETER	DESCRIPTION
backend	Backend instance or factory function that takes runtime and returns a backend. Use a factory for StateBackend: lambda rt: StateBackend(rt) TYPE: BACKEND_TYPES
sources	List of skill source paths (e.g., ["/skills/user/", "/skills/project/"]). TYPE: list[str]

modify_request

modify_request(request: ModelRequest) -> ModelRequest

Inject skills documentation into a model request's system message.

PARAMETER	DESCRIPTION
request	Model request to modify TYPE: ModelRequest

RETURNS	DESCRIPTION
ModelRequest	New model request with skills documentation injected into system message

before_agent

before_agent(
    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).

PARAMETER	DESCRIPTION
state	Current agent state. TYPE: SkillsState
runtime	Runtime context. TYPE: Runtime
config	Runnable config. TYPE: RunnableConfig

RETURNS	DESCRIPTION
SkillsStateUpdate | None	State update with skills_metadata populated, or None if already present

abefore_agent async

abefore_agent(
    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).

PARAMETER	DESCRIPTION
state	Current agent state. TYPE: SkillsState
runtime	Runtime context. TYPE: Runtime
config	Runnable config. TYPE: RunnableConfig

RETURNS	DESCRIPTION
SkillsStateUpdate | None	State update with skills_metadata populated, or None if already present

wrap_model_call

wrap_model_call(
    request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse

Inject skills documentation into the system prompt.

PARAMETER	DESCRIPTION
request	Model request being processed TYPE: ModelRequest
handler	Handler function to call with modified request TYPE: Callable[[ModelRequest], ModelResponse]

RETURNS	DESCRIPTION
ModelResponse	Model response from handler

awrap_model_call async

awrap_model_call(
    request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelResponse

Inject skills documentation into the system prompt (async version).

PARAMETER	DESCRIPTION
request	Model request being processed TYPE: ModelRequest
handler	Async handler function to call with modified request TYPE: Callable[[ModelRequest], Awaitable[ModelResponse]]

RETURNS	DESCRIPTION
ModelResponse	Model response from handler

before_model

before_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run before the model is called.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply before model call.

abefore_model async

abefore_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run before the model is called.

PARAMETER	DESCRIPTION
state	The agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply before model call.

after_model

after_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run after the model is called.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after model call.

aafter_model async

aafter_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run after the model is called.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after model call.

after_agent

after_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run after the agent execution completes.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after agent execution.

aafter_agent async

aafter_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run after the agent execution completes.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after agent execution.

wrap_tool_call

wrap_tool_call(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
) -> ToolMessage | Command[Any]

Intercept tool execution for retries, monitoring, or modification.

Async version is awrap_tool_call

Multiple middleware compose automatically (first defined = outermost).

Exceptions propagate unless handle_tool_errors is configured on ToolNode.

PARAMETER	DESCRIPTION
request	Tool call request with call dict, BaseTool, state, and runtime. Access state via request.state and runtime via request.runtime. TYPE: ToolCallRequest
handler	Callable to execute the tool (can be called multiple times). TYPE: Callable[[ToolCallRequest], ToolMessage | Command[Any]]

RETURNS	DESCRIPTION
ToolMessage | Command[Any]	ToolMessage or Command (the final result).

The handler Callable can be invoked multiple times for retry logic.

Each call to handler is independent and stateless.

Examples:

Modify request before execution

def wrap_tool_call(self, request, handler):
    modified_call = {
        **request.tool_call,
        "args": {
            **request.tool_call["args"],
            "value": request.tool_call["args"]["value"] * 2,
        },
    }
    request = request.override(tool_call=modified_call)
    return handler(request)

Retry on error (call handler multiple times)

def wrap_tool_call(self, request, handler):
    for attempt in range(3):
        try:
            result = handler(request)
            if is_valid(result):
                return result
        except Exception:
            if attempt == 2:
                raise
    return result

Conditional retry based on response

def wrap_tool_call(self, request, handler):
    for attempt in range(3):
        result = handler(request)
        if isinstance(result, ToolMessage) and result.status != "error":
            return result
        if attempt < 2:
            continue
        return result

awrap_tool_call async

awrap_tool_call(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
) -> ToolMessage | Command[Any]

Intercept and control async tool execution via handler callback.

The handler callback executes the tool call and returns a ToolMessage or Command. Middleware can call the handler multiple times for retry logic, skip calling it to short-circuit, or modify the request/response. Multiple middleware compose with first in list as outermost layer.

PARAMETER	DESCRIPTION
request	Tool call request with call dict, BaseTool, state, and runtime. Access state via request.state and runtime via request.runtime. TYPE: ToolCallRequest
handler	Async callable to execute the tool and returns ToolMessage or Command. Call this to execute the tool. Can be called multiple times for retry logic. Can skip calling it to short-circuit. TYPE: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]

RETURNS	DESCRIPTION
ToolMessage | Command[Any]	ToolMessage or Command (the final result).

The handler Callable can be invoked multiple times for retry logic.

Each call to handler is independent and stateless.

Examples:

Async retry on error

async def awrap_tool_call(self, request, handler):
    for attempt in range(3):
        try:
            result = await handler(request)
            if is_valid(result):
                return result
        except Exception:
            if attempt == 2:
                raise
    return result

async def awrap_tool_call(self, request, handler):
    if cached := await get_cache_async(request):
        return ToolMessage(content=cached, tool_call_id=request.tool_call["id"])
    result = await handler(request)
    await save_cache_async(request, result)
    return result