bitwikiorg """StateBackend: Store files in LangGraph agent state (ephemeral)."""
from typing import TYPE_CHECKING
from deepagents.backends.protocol import ( BackendProtocol, EditResult, FileDownloadResponse, FileInfo, FileUploadResponse, GrepMatch, WriteResult, ) from deepagents.backends.utils import ( _glob_search_files, create_file_data, file_data_to_string, format_read_response, grep_matches_from_files, perform_string_replacement, update_file_data, )
if TYPE_CHECKING: from langchain.tools import ToolRuntime
class StateBackend(BackendProtocol): """Backend that stores files in agent state (ephemeral).
Uses LangGraph's state management and checkpointing. Files persist within
a conversation thread but not across threads. State is automatically
checkpointed after each agent step.
Special handling: Since LangGraph state must be updated via Command objects
(not direct mutation), operations return Command objects instead of None.
This is indicated by the uses_state=True flag.
"""
def __init__(self, runtime: "ToolRuntime"):
"""Initialize StateBackend with runtime."""
self.runtime = runtime
def ls_info(self, path: str) -> list[FileInfo]:
"""List files and directories in the specified directory (non-recursive).
Args:
path: Absolute path to directory.
Returns:
List of FileInfo-like dicts for files and directories directly in the directory.
Directories have a trailing / in their path and is_dir=True.
"""
files = self.runtime.state.get("files", {})
infos: list[FileInfo] = []
subdirs: set[str] = set()
# Normalize path to have trailing slash for proper prefix matching
normalized_path = path if path.endswith("/") else path + "/"
for k, fd in files.items():
# Check if file is in the specified directory or a subdirectory
if not k.startswith(normalized_path):
continue
# Get the relative path after the directory
relative = k[len(normalized_path) :]
# If relative path contains '/', it's in a subdirectory
if "/" in relative:
# Extract the immediate subdirectory name
subdir_name = relative.split("/")[0]
subdirs.add(normalized_path + subdir_name + "/")
continue
# This is a file directly in the current directory
size = len("\n".join(fd.get("content", [])))
infos.append(
{
"path": k,
"is_dir": False,
"size": int(size),
"modified_at": fd.get("modified_at", ""),
}
)
# Add directories to the results
for subdir in sorted(subdirs):
infos.append(
{
"path": subdir,
"is_dir": True,
"size": 0,
"modified_at": "",
}
)
infos.sort(key=lambda x: x.get("path", ""))
return infos
def read(
self,
file_path: str,
offset: int = 0,
limit: int = 2000,
) -> str:
"""Read file content with line numbers.
Args:
file_path: Absolute file path.
offset: Line offset to start reading from (0-indexed).
limit: Maximum number of lines to read.
Returns:
Formatted file content with line numbers, or error message.
"""
files = self.runtime.state.get("files", {})
file_data = files.get(file_path)
if file_data is None:
return f"Error: File '{file_path}' not found"
return format_read_response(file_data, offset, limit)
def write(
self,
file_path: str,
content: str,
) -> WriteResult:
"""Create a new file with content.
Returns WriteResult with files_update to update LangGraph state.
"""
files = self.runtime.state.get("files", {})
if file_path in files:
return WriteResult(error=f"Cannot write to {file_path} because it already exists. Read and then make an edit, or write to a new path.")
new_file_data = create_file_data(content)
return WriteResult(path=file_path, files_update={file_path: new_file_data})
def edit(
self,
file_path: str,
old_string: str,
new_string: str,
replace_all: bool = False,
) -> EditResult:
"""Edit a file by replacing string occurrences.
Returns EditResult with files_update and occurrences.
"""
files = self.runtime.state.get("files", {})
file_data = files.get(file_path)
if file_data is None:
return EditResult(error=f"Error: File '{file_path}' not found")
content = file_data_to_string(file_data)
result = perform_string_replacement(content, old_string, new_string, replace_all)
if isinstance(result, str):
return EditResult(error=result)
new_content, occurrences = result
new_file_data = update_file_data(file_data, new_content)
return EditResult(path=file_path, files_update={file_path: new_file_data}, occurrences=int(occurrences))
def grep_raw(
self,
pattern: str,
path: str = "/",
glob: str | None = None,
) -> list[GrepMatch] | str:
files = self.runtime.state.get("files", {})
return grep_matches_from_files(files, pattern, path, glob)
def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
"""Get FileInfo for files matching glob pattern."""
files = self.runtime.state.get("files", {})
result = _glob_search_files(files, pattern, path)
if result == "No files found":
return []
paths = result.split("\n")
infos: list[FileInfo] = []
for p in paths:
fd = files.get(p)
size = len("\n".join(fd.get("content", []))) if fd else 0
infos.append(
{
"path": p,
"is_dir": False,
"size": int(size),
"modified_at": fd.get("modified_at", "") if fd else "",
}
)
return infos
def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
"""Upload multiple files to state.
Args:
files: List of (path, content) tuples to upload
Returns:
List of FileUploadResponse objects, one per input file
"""
raise NotImplementedError(
"StateBackend does not support upload_files yet. You can upload files "
"directly by passing them in invoke if you're storing files in the memory."
)
def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
"""Download multiple files from state.
Args:
paths: List of file paths to download
Returns:
List of FileDownloadResponse objects, one per input path
"""
state_files = self.runtime.state.get("files", {})
responses: list[FileDownloadResponse] = []
for path in paths:
file_data = state_files.get(path)
if file_data is None:
responses.append(FileDownloadResponse(path=path, content=None, error="file_not_found"))
continue
# Convert file data to bytes
content_str = file_data_to_string(file_data)
content_bytes = content_str.encode("utf-8")
responses.append(FileDownloadResponse(path=path, content=content_bytes, error=None))
return responses