"""Agent management and creation.""" from __future__ import annotations import logging import os import re import shutil import tempfile import tomllib from pathlib import Path from typing import TYPE_CHECKING, Any, cast from deepagents import create_deep_agent from deepagents.backends import CompositeBackend, LocalShellBackend from deepagents.backends.filesystem import FilesystemBackend from deepagents.middleware import MemoryMiddleware, SkillsMiddleware # Backwards-compat flag: SDKs before 0.5.4 accept only `list[str]` for # `SkillsMiddleware.sources`; newer SDKs expose the `SkillSource` alias # that permits `(path, label)` tuples. The `skills` module is already # loaded by the `SkillsMiddleware` import above, so the extra lookup # here adds no startup cost. try: from deepagents.middleware.skills import SkillSource as _SkillSource # noqa: F401 except ImportError: _SUPPORTS_SKILL_SOURCE_TUPLES = False else: _SUPPORTS_SKILL_SOURCE_TUPLES = True if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Sequence from deepagents.backends.sandbox import SandboxBackendProtocol from deepagents.middleware.async_subagents import AsyncSubAgent from deepagents.middleware.subagents import CompiledSubAgent, SubAgent from langchain.agents.middleware import InterruptOnConfig from langchain.agents.middleware.types import AgentState from langchain.messages import ToolCall from langchain.tools import BaseTool from langchain_core.language_models import BaseChatModel from langchain_core.messages import ToolMessage from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.prebuilt.tool_node import ToolCallRequest from langgraph.pregel import Pregel from langgraph.runtime import Runtime from langgraph.types import Command from deepagents_code.mcp_tools import MCPServerInfo from deepagents_code.output import OutputFormat from langchain.agents.middleware.types import AgentMiddleware from deepagents_code import theme from deepagents_code._cli_context import CLIContextSchema from deepagents_code._constants import DEFAULT_AGENT_NAME from deepagents_code.config import ( _INHERITED_PYTHONPATH_ENV, _ShellAllowAll, config, console, get_default_coding_instructions, get_glyphs, get_langsmith_project_name, settings, ) from deepagents_code.configurable_model import ConfigurableModelMiddleware from deepagents_code.filesystem_empty_result import _FilesystemEmptyResultMiddleware from deepagents_code.integrations.sandbox_factory import get_default_working_dir from deepagents_code.local_context import ( LocalContextMiddleware, _AsyncExecutableBackend, _ExecutableBackend, ) from deepagents_code.project_utils import ProjectContext, get_server_project_context from deepagents_code.subagents import list_subagents from deepagents_code.unicode_security import ( check_url_safety, detect_dangerous_unicode, format_warning_detail, render_with_unicode_markers, strip_dangerous_unicode, summarize_issues, ) logger = logging.getLogger(__name__) REQUIRE_COMPACT_TOOL_APPROVAL: bool = True """When `True`, `compact_conversation` requires HITL approval like other gated tools.""" def _sanitize_agent_message_name(agent_name: str) -> str: """Return a provider-safe message name for a user-facing agent name. Args: agent_name: Display/storage name for the selected agent. Returns: Name containing only alphanumerics, underscores, and hyphens. """ sanitized = re.sub(r"[^a-zA-Z0-9_-]+", "_", agent_name).strip("_") return sanitized or DEFAULT_AGENT_NAME class ShellAllowListMiddleware(AgentMiddleware): """Validate shell commands against an allow-list without HITL interrupts. When the agent invokes the `execute` shell tool, this middleware checks the command against the configured allow-list **before execution**. Rejected commands are returned as error `ToolMessage` objects — the graph never pauses, so LangSmith traces stay as a single continuous run. Use this middleware in non-interactive mode to avoid the interrupt/resume cycle that fragments traces. """ def __init__(self, allow_list: list[str]) -> None: """Initialize with the shell allow-list to validate commands against. Args: allow_list: Allowed command names (e.g. `["ls", "cat", "grep"]`). Must be a non-empty restrictive list — not `SHELL_ALLOW_ALL`. Raises: ValueError: If `allow_list` is empty. TypeError: If `allow_list` is the `SHELL_ALLOW_ALL` sentinel. """ from deepagents_code.config import SHELL_ALLOW_ALL super().__init__() if not allow_list: msg = "allow_list must not be empty; disable shell access instead" raise ValueError(msg) if isinstance(allow_list, type(SHELL_ALLOW_ALL)): msg = ( "SHELL_ALLOW_ALL should not be used with " "ShellAllowListMiddleware; use auto_approve=True instead" ) raise TypeError(msg) self._allow_list = list(allow_list) def _validate_tool_call(self, request: ToolCallRequest) -> ToolMessage | None: """Return an error tool message when a shell command is not allowed. Args: request: The tool call request being processed. Returns: An error `ToolMessage` when the shell command should be rejected, otherwise `None`. """ from langchain_core.messages import ToolMessage as LCToolMessage from deepagents_code.config import is_shell_command_allowed if request.tool_call["name"] != "execute": return None args = request.tool_call.get("args") or {} command = args.get("command", "") if is_shell_command_allowed(command, self._allow_list): logger.debug("Shell command allowed: %r", command) return None logger.warning("Shell command rejected by allow-list: %r", command) allowed_str = ", ".join(self._allow_list) return LCToolMessage( content=( f"Shell command rejected: `{command}` is not in the allow-list. " f"Allowed commands: {allowed_str}. " f"Please use an allowed command or try another approach." ), name="execute", tool_call_id=request.tool_call["id"], status="error", ) def wrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]], ) -> ToolMessage | Command[Any]: """Reject disallowed shell commands; pass everything else through. Args: request: The tool call request being processed. handler: The next handler in the middleware chain. Returns: The tool execution result, or an error `ToolMessage` for rejected shell commands. """ if (rejection := self._validate_tool_call(request)) is not None: return rejection return handler(request) async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]], ) -> ToolMessage | Command[Any]: """Reject disallowed shell commands; pass everything else through. Args: request: The tool call request being processed. handler: The next handler in the middleware chain. Returns: The tool execution result, or an error `ToolMessage` for rejected shell commands. """ if (rejection := self._validate_tool_call(request)) is not None: return rejection return await handler(request) _INTERPRETER_WRITE_TOOLS: frozenset[str] = frozenset( {"execute", "write_file", "edit_file"} ) """Tools considered write/shell capable for PTC auditing. When `interpreter_ptc="all"` resolves to this set, an INFO log names every write tool that was included so the audit trail is searchable. The `"safe"` preset already excludes them; this is the belt-and-braces check for `"all"`. """ def _resolve_ptc_option( ptc: str | bool | list[str], *, tools: Sequence[BaseTool | Callable | dict[str, Any]], acknowledge_unsafe: bool, auto_approve: bool, ) -> list[str] | None: """Resolve the configured PTC allowlist to a concrete list of tool names. Names are *not* validated against `tools`. The Deep Agents SDK injects the filesystem, `task`, `write_todos`, and `execute` tools via middleware in `create_deep_agent` — *after* this point — so they are absent from `tools` here, and the SDK exposes no importable list of them. `CodeInterpreterMiddleware` matches the resolved names against the live runtime registry and silently ignores any that are absent, so resolution passes names through and lets runtime decide. (Names that match nothing at runtime are dropped, so a typo silently exposes no tool rather than raising.) Args: ptc: Raw `interpreter_ptc` value from settings or CLI. Accepts `False`/`[]`, `"safe"`, `"all"`, or a list of names. A list may include `"safe"`, which expands to `INTERPRETER_PTC_SAFE_PRESET`; `"all"` is rejected inside a list. tools: Tools passed to `create_cli_agent`. Used only to enumerate `"all"`, which is therefore limited to these explicitly-passed tools (the SDK runtime built-ins cannot be enumerated here). acknowledge_unsafe: Mirrors `settings.interpreter_ptc_acknowledge_unsafe`; required when `ptc="all"` and `auto_approve` is `False`. auto_approve: Whether HITL approval is globally disabled. When `True`, `"all"` does not require `acknowledge_unsafe` because every host tool already runs without prompting. Returns: `None` when PTC should be disabled, otherwise a list of tool names suitable for `CodeInterpreterMiddleware(ptc=...)`. Raises: ValueError: For `"all"` inside a list, for `"all"` without `acknowledge_unsafe` outside of `auto_approve`, or for an invalid `ptc` type or string. """ from langchain.tools import BaseTool as _BaseTool if ptc is False or ptc is None or ptc == []: return None live_names: list[str] = [] for tool in tools: if isinstance(tool, _BaseTool): name = tool.name if isinstance(name, str): live_names.append(name) elif isinstance(tool, dict): raw_name = cast("dict[str, Any]", tool).get("name") if isinstance(raw_name, str): live_names.append(raw_name) else: attr = getattr(tool, "name", None) if isinstance(attr, str): live_names.append(attr) live_set: set[str] = set(live_names) if isinstance(ptc, str): normalized = ptc.strip().lower() if normalized == "safe": from deepagents_code.config import INTERPRETER_PTC_SAFE_PRESET # Return the preset as-is; the middleware exposes whichever members # exist in the live registry at runtime (they are SDK built-ins not # present in `tools` here). return sorted(INTERPRETER_PTC_SAFE_PRESET) if normalized == "all": if not auto_approve and not acknowledge_unsafe: msg = ( "interpreter_ptc='all' exposes every host tool to PTC " "calls that bypass HITL approval. Set " "interpreter_ptc_acknowledge_unsafe=True (or use " "auto_approve=True) to opt in." ) raise ValueError(msg) # `all` can only enumerate the tools passed to `create_cli_agent`; # SDK runtime built-ins (filesystem, `task`, …) are injected later # and are not enumerable here. Exposing them under `all` needs an # "expose everything" sentinel in `CodeInterpreterMiddleware` # (tracked in langchain-ai/deepagents#3847). included = sorted(live_set) write_included = sorted(_INTERPRETER_WRITE_TOOLS & live_set) if write_included: logger.info( "interpreter_ptc='all' includes write/shell tools: %s", write_included, ) return included msg = ( f"Invalid interpreter_ptc string {ptc!r}; expected 'safe', 'all', " "or a list of tool names." ) raise ValueError(msg) if isinstance(ptc, list): from deepagents_code.config import INTERPRETER_PTC_SAFE_PRESET if any(name.strip().lower() == "all" for name in ptc): msg = ( "interpreter_ptc list entries cannot include 'all'; use 'all' " "as a standalone value or list explicit tool names (optionally " "with the 'safe' preset)." ) raise ValueError(msg) resolved: list[str] = [] seen: set[str] = set() def _add(name: str) -> None: if name not in seen: seen.add(name) resolved.append(name) for name in ptc: if name.strip().lower() == "safe": for member in sorted(INTERPRETER_PTC_SAFE_PRESET): _add(member) continue _add(name) # Explicit names are passed through unvalidated: the middleware resolves # them against the live runtime registry (which includes the SDK # built-ins absent from `tools`) and drops any that match nothing. absent = sorted(n for n in resolved if n not in live_set) if absent: logger.debug( "interpreter_ptc names not in the build-time toolset (resolved " "at runtime if present): %s", absent, ) return resolved msg = ( "interpreter_ptc must be False, 'safe', 'all', or a list of tool names; " f"got {type(ptc).__name__}." ) raise ValueError(msg) def load_async_subagents(config_path: Path | None = None) -> list[AsyncSubAgent]: """Load async subagent definitions from `config.toml`. Reads the `[async_subagents]` section where each sub-table defines a remote LangGraph deployment: ```toml [async_subagents.researcher] description = "Research agent" url = "https://my-deployment.langsmith.dev" graph_id = "agent" ``` Args: config_path: Path to config file. Defaults to `~/.deepagents/config.toml`. Returns: List of `AsyncSubAgent` specs (empty if section is absent or invalid). """ if config_path is None: config_path = Path.home() / ".deepagents" / "config.toml" if not config_path.exists(): return [] try: with config_path.open("rb") as f: data = tomllib.load(f) except (tomllib.TOMLDecodeError, PermissionError, OSError) as e: logger.warning("Could not read async subagents from %s: %s", config_path, e) console.print( f"[bold yellow]Warning:[/bold yellow] Could not read async subagents " f"from {config_path}: {e}", ) return [] section = data.get("async_subagents") if not isinstance(section, dict): return [] required = {"description", "graph_id"} agents: list[AsyncSubAgent] = [] for name, spec in section.items(): if not isinstance(spec, dict): logger.warning("Skipping async subagent '%s': expected a table", name) continue missing = required - spec.keys() if missing: logger.warning( "Skipping async subagent '%s': missing fields %s", name, missing ) continue agent: AsyncSubAgent = { "name": name, "description": spec["description"], "graph_id": spec["graph_id"], } if "url" in spec and isinstance(spec["url"], str): agent["url"] = spec["url"] if "headers" in spec and isinstance(spec["headers"], dict): agent["headers"] = spec["headers"] agents.append(agent) return agents def _is_agent_dir_entry(entry: Path) -> bool: """Return whether a `~/.deepagents/` entry should be listed as an agent. Filters out symlinks (so dangling links don't masquerade as agents) and dot-prefixed names — `.state/` (app internal state) plus any other hidden directory the user may have placed there. `OSError` from `is_dir`/`is_symlink` propagates so callers can log with the failing entry's name as context. """ if entry.name.startswith("."): return False return entry.is_dir() and not entry.is_symlink() def get_available_agent_names() -> list[str]: """Return a sorted list of available agent names from `~/.deepagents/`. Scans the user's `.deepagents` directory and returns each real subdirectory found there. Symlinks excluded so a dangling link does not masquerade as an agent. Dot-prefixed entries (e.g., `.state/`) are skipped so internal app state never appears as an agent. Filesystem errors (missing parent, permission denied, broken entries) are logged and surfaced as an empty list rather than raised — the caller shows an empty modal instead of crashing mid-render. Returns: Sorted list of agent names. Empty when no agents exist yet or the directory is unreadable (see log for the underlying cause). """ agents_dir = settings.user_deepagents_dir try: entries = list(agents_dir.iterdir()) except FileNotFoundError: return [] except OSError: logger.warning("Could not list agents in %s", agents_dir, exc_info=True) return [] names: list[str] = [] for entry in entries: try: if _is_agent_dir_entry(entry): names.append(entry.name) except OSError: logger.debug( "Skipping unreadable entry in %s: %s", agents_dir, entry.name, exc_info=True, ) return sorted(names) def list_agents(*, output_format: OutputFormat = "text") -> None: """List all available agents. Args: output_format: Output format — `'text'` (Rich) or `'json'`. """ agents_dir = settings.user_deepagents_dir names = get_available_agent_names() if not names: if output_format == "json": from deepagents_code.output import write_json write_json("list", []) return console.print("[yellow]No agents found.[/yellow]") console.print( "[dim]Agents will be created in ~/.deepagents/ " "when you first use them.[/dim]", style=theme.MUTED, ) return if output_format == "json": from deepagents_code.output import write_json agents = [] for name in names: agent_path = agents_dir / name agents.append( { "name": name, "path": str(agent_path), "has_agents_md": (agent_path / "AGENTS.md").exists(), "is_default": name == DEFAULT_AGENT_NAME, } ) write_json("list", agents) return from rich.markup import escape as escape_markup console.print("\n[bold]Available Agents:[/bold]\n", style=theme.PRIMARY) bullet = get_glyphs().bullet for name in names: agent_path = agents_dir / name agent_name = escape_markup(name) is_default = name == DEFAULT_AGENT_NAME default_label = " [dim](default)[/dim]" if is_default else "" if (agent_path / "AGENTS.md").exists(): console.print( f" {bullet} [bold]{agent_name}[/bold]{default_label}", style=theme.PRIMARY, ) else: console.print( f" {bullet} [bold]{agent_name}[/bold]{default_label}" " [dim](incomplete)[/dim]", style=theme.WARNING, ) console.print( f" {escape_markup(str(agent_path))}", style=theme.MUTED, ) console.print() def reset_agent( agent_name: str, source_agent: str | None = None, *, dry_run: bool = False, output_format: OutputFormat = "text", ) -> None: """Reset an agent to default or copy from another agent. Args: agent_name: Name of the agent to reset. source_agent: Copy AGENTS.md from this agent instead of default. dry_run: If `True`, print what would happen without making changes. output_format: Output format — `'text'` (Rich) or `'json'`. Raises: SystemExit: If the source agent is not found. """ agents_dir = settings.user_deepagents_dir agent_dir = agents_dir / agent_name if source_agent: source_dir = agents_dir / source_agent source_md = source_dir / "AGENTS.md" if not source_md.exists(): console.print( f"[bold red]Error:[/bold red] Source agent '{source_agent}' not found " "or has no AGENTS.md\n" " Available agents: dcode agents list" ) raise SystemExit(1) source_content = source_md.read_text() action_desc = f"contents of agent '{source_agent}'" else: source_content = get_default_coding_instructions() action_desc = "default" if dry_run: if output_format == "json": from deepagents_code.output import write_json write_json( "reset", { "agent": agent_name, "reset_to": source_agent or "default", "path": str(agent_dir), "dry_run": True, }, ) return exists = "remove and recreate" if agent_dir.exists() else "create" console.print(f"Would {exists} {agent_dir} with {action_desc} prompt.") console.print("No changes made.", style=theme.MUTED) return if agent_dir.exists(): shutil.rmtree(agent_dir) if output_format != "json": console.print( f"Removed existing agent directory: {agent_dir}", style=theme.WARNING ) agent_dir.mkdir(parents=True, exist_ok=True) agent_md = agent_dir / "AGENTS.md" agent_md.write_text(source_content) if output_format == "json": from deepagents_code.output import write_json write_json( "reset", { "agent": agent_name, "reset_to": source_agent or "default", "path": str(agent_dir), }, ) return console.print( f"{get_glyphs().checkmark} Agent '{agent_name}' reset to {action_desc}", style=theme.PRIMARY, ) console.print(f"Location: {agent_dir}\n", style=theme.MUTED) MODEL_IDENTITY_RE = re.compile(r"### Model Identity\n\n.*?(?=###|\Z)", re.DOTALL) """Matches the `### Model Identity` section in the system prompt, up to the next heading or end of string.""" def build_model_identity_section( name: str | None, provider: str | None = None, context_limit: int | None = None, unsupported_modalities: frozenset[str] = frozenset(), ) -> str: """Build the `### Model Identity` section for the system prompt. Args: name: Model identifier (e.g. `claude-opus-4-6`). provider: Provider identifier (e.g. `anthropic`). context_limit: Max input tokens from the model profile. unsupported_modalities: Input modalities not indicated as supported by the model profile (e.g. `{"audio", "video"}`). Returns: The section text including the heading and trailing newline, or an empty string if `name` is falsy. """ if not name: return "" section = f"### Model Identity\n\nYou are running as model `{name}`" if provider: section += f" (provider: {provider})" section += ".\n" if context_limit: section += f"Your context window is {context_limit:,} tokens.\n" if unsupported_modalities: items = sorted(unsupported_modalities) if len(items) == 1: joined = items[0] elif len(items) == 2: # noqa: PLR2004 joined = f"{items[0]} and {items[1]}" else: joined = ", ".join(items[:-1]) + f", and {items[-1]}" section += ( f"{joined.capitalize()} input may not be available for this model. " "Do not attempt to read or process these content types.\n" ) section += "\n" return section def get_system_prompt( assistant_id: str, sandbox_type: str | None = None, *, interactive: bool = True, cwd: str | Path | None = None, ) -> str: """Get the base system prompt for the agent. Loads the base system prompt template from `system_prompt.md` and interpolates dynamic sections (model identity, working directory, skills path, execution mode, and todo-list guidance for interactive vs headless). Args: assistant_id: The agent identifier for path references sandbox_type: Type of sandbox provider (`'agentcore'`, `'daytona'`, `'langsmith'`, `'modal'`, `'runloop'`). If `None`, agent is operating in local mode. interactive: When `False`, the prompt is tailored for headless non-interactive execution (no human in the loop). cwd: Override the working directory shown in the prompt. Returns: The system prompt string Example: ```txt You are running as model {MODEL} (provider: {PROVIDER}). Your context window is {CONTEXT_WINDOW} tokens. ... {CONDITIONAL SECTIONS} ... ``` """ template = (Path(__file__).parent / "system_prompt.md").read_text() skills_path = f"~/.deepagents/{assistant_id}/skills" if interactive: mode_description = "an interactive TUI on the user's computer" interactive_preamble = ( "The user sends you messages and you respond with text and tool " "calls. Your tools run on the user's machine. The user can see " "your responses and tool outputs in real time, so keep them " "informed — but don't over-explain." ) ambiguity_guidance = ( "- If the request is ambiguous, ask questions before acting.\n" "- If asked how to approach something, explain first, then act." ) todo_guidance = ( "6. When first creating a todo list for a task, ALWAYS ask the user if " "the plan looks good before starting work\n" ' - Create the todos, then ask: "Does this plan ' 'look good?" or similar\n' " - Wait for the user's response before marking the first todo as " "in_progress\n" "7. Update todo status promptly as you complete each item" ) else: mode_description = ( "non-interactive (headless) mode — there is no human operator " "monitoring your output in real time" ) interactive_preamble = ( "You received a single task and must complete it fully and " "autonomously. There is no human available to answer follow-up " "questions, so do NOT ask for clarification — make reasonable " "assumptions and proceed." ) ambiguity_guidance = ( "- Do NOT ask clarifying questions — there is no human to answer " "them. Make reasonable assumptions and proceed.\n" "- If you encounter ambiguity, choose the most reasonable " "interpretation and note your assumption briefly.\n" "- Always use non-interactive command variants — no human is " "available to respond to prompts. Examples: `npm init -y` not " "`npm init`, `apt-get install -y` not `apt-get install`, " "`yes |` or `--no-input`/`--non-interactive` flags where " "available. Never run commands that block waiting for stdin." ) todo_guidance = ( "6. There is no human operator in this mode — do NOT ask the user to " "approve your plan or wait for a reply.\n" " After you create todos for a multi-step task, mark the first item " "`in_progress` immediately and start work.\n" " If the plan needs adjustment, revise the todo list yourself; do " "not block on human confirmation.\n" "7. Update todo status promptly as you complete each item" ) model_identity_section = build_model_identity_section( settings.model_name, provider=settings.model_provider, context_limit=settings.model_context_limit, unsupported_modalities=settings.model_unsupported_modalities, ) # Build working directory section (local vs sandbox) if sandbox_type: working_dir = get_default_working_dir(sandbox_type) working_dir_section = ( f"### Current Working Directory\n\n" f"You are operating in a **remote Linux sandbox** at `{working_dir}`.\n\n" f"All code execution and file operations happen in this sandbox " f"environment.\n\n" f"**Important:**\n" f"- The application is running locally on the user's machine, but you " f"execute code remotely\n" f"- Use `{working_dir}` as your working directory for all operations\n" f"- **You do NOT have access to the user's local filesystem.** Paths " f"like `/Users/...`, `/home//...`, `C:\\...`, etc. do not " f"exist in this sandbox. Never reference or attempt to read/write local " f"paths — all files must be within the sandbox at `{working_dir}`\n" f"- When delegating to subagents, ensure they also use sandbox paths " f"(`{working_dir}/...`), not local paths\n\n" ) else: if cwd is not None: resolved_cwd = Path(cwd) else: try: resolved_cwd = Path.cwd() except OSError: logger.warning( "Could not determine working directory for system prompt", exc_info=True, ) resolved_cwd = Path() cwd = resolved_cwd working_dir_section = ( f"### Current Working Directory\n\n" f"The filesystem backend is currently operating in: `{cwd}`\n\n" f"### File System and Paths\n\n" f"**IMPORTANT - Path Handling:**\n" f"- All file paths must be absolute paths (e.g., `{cwd}/file.txt`)\n" f"- Use the working directory to construct absolute paths\n" f"- Example: To create a file in your working directory, " f"use `{cwd}/research_project/file.md`\n" f"- Never use relative paths - always construct full absolute paths\n\n" ) result = ( template.replace("{mode_description}", mode_description) .replace("{interactive_preamble}", interactive_preamble) .replace("{ambiguity_guidance}", ambiguity_guidance) .replace("{todo_guidance}", todo_guidance) .replace("{model_identity_section}", model_identity_section) .replace("{working_dir_section}", working_dir_section) .replace("{skills_path}", skills_path) ) # Detect unreplaced placeholders (defense-in-depth for template typos) unreplaced = re.findall(r"\{[a-z_]+\}", result) if unreplaced: logger.warning("System prompt contains unreplaced placeholders: %s", unreplaced) return result def _format_write_file_description( tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any] ) -> str: """Format write_file tool call for approval prompt. Returns: Formatted description string for the write_file tool call. """ args = tool_call["args"] file_path = args.get("file_path", "unknown") action = "Overwrite" if Path(file_path).exists() else "Create" return f"Action: {action} file" def _format_edit_file_description( tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any] ) -> str: """Format edit_file tool call for approval prompt. Returns: Formatted description string for the edit_file tool call. """ args = tool_call["args"] replace_all = bool(args.get("replace_all", False)) scope = "all occurrences" if replace_all else "single occurrence" return f"Action: Replace text ({scope})" def _format_web_search_description( tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any] ) -> str: """Format web_search tool call for approval prompt. Returns: Formatted description string for the web_search tool call. """ args = tool_call["args"] query = args.get("query", "unknown") max_results = args.get("max_results", 5) return ( f"Query: {query}\nMax results: {max_results}\n\n" f"{get_glyphs().warning} This will use Tavily API credits" ) def _format_fetch_url_description( tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any] ) -> str: """Format fetch_url tool call for approval prompt. Returns: Formatted description string for the fetch_url tool call. """ args = tool_call["args"] url = str(args.get("url", "unknown")) display_url = strip_dangerous_unicode(url) timeout = args.get("timeout", 30) safety = check_url_safety(url) warning_lines: list[str] = [] if not safety.safe: detail = format_warning_detail(safety.warnings) warning_lines.append(f"{get_glyphs().warning} URL warning: {detail}") if safety.decoded_domain: warning_lines.append( f"{get_glyphs().warning} Decoded domain: {safety.decoded_domain}" ) warning_block = "\n".join(warning_lines) if warning_block: warning_block = f"\n{warning_block}" return ( f"URL: {display_url}\nTimeout: {timeout}s\n\n" f"{get_glyphs().warning} Will fetch and convert web content to markdown" f"{warning_block}" ) def _format_task_description( tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any] ) -> str: """Format task (subagent) tool call for approval prompt. The task tool signature is: task(description: str, subagent_type: str) The description contains all instructions that will be sent to the subagent. Returns: Formatted description string for the task tool call. """ args = tool_call["args"] description = args.get("description", "unknown") subagent_type = args.get("subagent_type", "unknown") # Truncate description if too long for display description_preview = description if len(description) > 500: # noqa: PLR2004 # Subagent description length threshold description_preview = description[:500] + "..." glyphs = get_glyphs() separator = glyphs.box_horizontal * 40 warning_msg = "Subagent will have access to file operations and shell commands" return ( f"Subagent Type: {subagent_type}\n\n" f"{glyphs.warning} {warning_msg} {glyphs.warning}\n\n" f"Task Instructions:\n" f"{separator}\n" f"{description_preview}" ) def _format_execute_description( tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any] ) -> str: """Format execute tool call for approval prompt. Returns: Formatted description string for the execute tool call. """ args = tool_call["args"] command_raw = str(args.get("command", "N/A")) command = strip_dangerous_unicode(command_raw) project_context = get_server_project_context() effective_cwd = ( str(project_context.user_cwd) if project_context is not None else str(Path.cwd()) ) lines = [f"Execute Command: {command}", f"Working Directory: {effective_cwd}"] issues = detect_dangerous_unicode(command_raw) if issues: summary = summarize_issues(issues) lines.append(f"{get_glyphs().warning} Hidden Unicode detected: {summary}") raw_marked = render_with_unicode_markers(command_raw) if len(raw_marked) > 220: # noqa: PLR2004 # UI display truncation threshold raw_marked = raw_marked[:220] + "..." lines.append(f"Raw: {raw_marked}") return "\n".join(lines) def _add_interrupt_on() -> dict[str, InterruptOnConfig]: """Configure human-in-the-loop interrupt settings for all gated tools. Every tool that can have side effects or access external resources (shell execution, file writes/edits, web search, URL fetch, task delegation) is gated behind an approval prompt unless auto-approve is enabled. Returns: Dictionary mapping tool names to their interrupt configuration. """ execute_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": _format_execute_description, # ty: ignore[invalid-argument-type] # Callable description narrower than TypedDict expects } write_file_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": _format_write_file_description, # ty: ignore[invalid-argument-type] # Callable description narrower than TypedDict expects } edit_file_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": _format_edit_file_description, # ty: ignore[invalid-argument-type] # Callable description narrower than TypedDict expects } web_search_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": _format_web_search_description, # ty: ignore[invalid-argument-type] # Callable description narrower than TypedDict expects } fetch_url_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": _format_fetch_url_description, # ty: ignore[invalid-argument-type] # Callable description narrower than TypedDict expects } task_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": _format_task_description, # ty: ignore[invalid-argument-type] # Callable description narrower than TypedDict expects } async_subagent_interrupt_config: InterruptOnConfig = { "allowed_decisions": ["approve", "reject"], "description": "Launch, update, or cancel a remote async subagent.", } interrupt_map: dict[str, InterruptOnConfig] = { "execute": execute_interrupt_config, "write_file": write_file_interrupt_config, "edit_file": edit_file_interrupt_config, "web_search": web_search_interrupt_config, "fetch_url": fetch_url_interrupt_config, "task": task_interrupt_config, "start_async_task": async_subagent_interrupt_config, "update_async_task": async_subagent_interrupt_config, "cancel_async_task": async_subagent_interrupt_config, } if REQUIRE_COMPACT_TOOL_APPROVAL: interrupt_map["compact_conversation"] = { "allowed_decisions": ["approve", "reject"], "description": ( "Offloads older messages to backend storage and " "replaces them with a summary, freeing context " "window space. Recent messages are kept as-is. " "Full history remains available for retrieval." ), } return interrupt_map def _apply_inherited_pythonpath(env: dict[str, str]) -> None: """Re-apply a relayed launch-time `PYTHONPATH` to a shell-command env. `server._build_server_env` strips `PYTHONPATH` from the server interpreter and relays the launch value via `config._INHERITED_PYTHONPATH_ENV`. This restores it as `PYTHONPATH` for the approval-gated `execute` subprocesses, which run in the user's working directory and need the import path. Mutates `env` in place; a no-op when no value was relayed. Args: env: Environment mapping for the shell backend, modified in place. """ inherited = env.pop(_INHERITED_PYTHONPATH_ENV, None) if inherited is not None: env["PYTHONPATH"] = inherited def create_cli_agent( model: str | BaseChatModel, assistant_id: str, *, tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None, sandbox: SandboxBackendProtocol | None = None, sandbox_type: str | None = None, system_prompt: str | None = None, interactive: bool = True, auto_approve: bool = False, interrupt_shell_only: bool = False, shell_allow_list: list[str] | None = None, enable_ask_user: bool = True, enable_memory: bool = True, enable_skills: bool = True, enable_shell: bool = True, enable_interpreter: bool = False, checkpointer: BaseCheckpointSaver | None = None, mcp_server_info: list[MCPServerInfo] | None = None, cwd: str | Path | None = None, project_context: ProjectContext | None = None, async_subagents: list[AsyncSubAgent] | None = None, ) -> tuple[Pregel[Any, Any, Any, Any], CompositeBackend]: """Create a CLI-configured agent with flexible options. This is the main entry point for creating a Deep Agents Code agent, usable both internally and from external code (e.g., benchmarking frameworks). Args: model: LLM model to use (e.g., `'provider:model'`) assistant_id: Agent identifier for memory/state storage tools: Additional tools to provide to agent sandbox: Optional sandbox backend for remote execution (e.g., `ModalSandbox`). If `None`, uses local filesystem + shell. sandbox_type: Type of sandbox provider (`'agentcore'`, `'daytona'`, `'langsmith'`, `'modal'`, `'runloop'`). Used for system prompt generation. system_prompt: Override the default system prompt. If `None`, generates one based on `sandbox_type`, `assistant_id`, and `interactive`. interactive: When `False`, the auto-generated system prompt is tailored for headless non-interactive execution. Ignored when `system_prompt` is provided explicitly. auto_approve: If `True`, no tools trigger human-in-the-loop interrupts — all calls (shell execution, file writes/edits, web search, URL fetch) run automatically. If `False`, tools pause for user confirmation via the approval menu. See `_add_interrupt_on` for the full list of gated tools. interrupt_shell_only: If `True`, all HITL interrupts are disabled; shell commands are validated inline by `ShellAllowListMiddleware` against the configured allow-list instead. Used in non-interactive mode with a restrictive shell allow-list to avoid splitting traces into multiple LangSmith runs. Has no effect when `auto_approve` is `True` (interrupts are already disabled) or when `shell_allow_list` is `SHELL_ALLOW_ALL`. shell_allow_list: Explicit restrictive shell allow-list forwarded from the CLI process. When provided (and `interrupt_shell_only` is `True`), used directly instead of reading `settings.shell_allow_list` (which may not be set in the server subprocess environment). enable_ask_user: Enable `AskUserMiddleware` so the agent can ask clarifying questions. Disabled in non-interactive mode. enable_memory: Enable `MemoryMiddleware` for persistent memory enable_skills: Enable `SkillsMiddleware` for custom agent skills enable_shell: Enable shell execution via `LocalShellBackend` (only in local mode). When enabled, the `execute` tool is available. enable_interpreter: Wire `CodeInterpreterMiddleware` from `langchain-quickjs` into the main agent. Local-mode only — passing a non-`None` `sandbox` while `enable_interpreter=True` raises `ValueError`. Subagents do not receive the interpreter in v1. PTC (`tools.*` host bridge) calls bypass `interrupt_on`/HITL approval, so `settings.interpreter_ptc` is the only effective control over which host tools can be invoked from inside the REPL. `js_eval` itself is intentionally not gated by HITL — per-call approval would be unusably noisy and would not block PTC fan-out anyway. The `"safe"` preset is therefore restricted to tools that are already non-HITL outside the REPL (read-only file inspection); exposing HITL-gated tools — network fetch, subagent dispatch, shell, file writes — requires an explicit list or `interpreter_ptc="all"` with `interpreter_ptc_acknowledge_unsafe=True`. Requires the `quickjs` optional extra (`langchain-quickjs>=0.1.2,<0.2.0`). checkpointer: Optional checkpointer for session persistence. When `None`, the graph is compiled without a checkpointer. mcp_server_info: MCP server metadata to surface in the system prompt. cwd: Override the working directory for the agent's filesystem backend and system prompt. project_context: Explicit project path context for project-sensitive behavior such as project `AGENTS.md` files, skills, subagents, and MCP trust. async_subagents: Remote LangGraph deployments to expose as async subagent tools. Loaded from `[async_subagents]` in `config.toml` or passed directly. Returns: 2-tuple of `(agent_graph, backend)` - `agent_graph`: Configured LangGraph Pregel instance ready for execution - `composite_backend`: `CompositeBackend` for file operations Raises: ValueError: When `enable_interpreter=True` is paired with a non-`None` `sandbox`, when `settings.interpreter_ptc` contains unknown tool names, or when `interpreter_ptc="all"` is used without `auto_approve` or `interpreter_ptc_acknowledge_unsafe`. """ tools = tools or [] effective_cwd = ( Path(cwd) if cwd is not None else (project_context.user_cwd if project_context is not None else None) ) # Setup agent directory for persistent memory (if enabled) if enable_memory or enable_skills: agent_dir = settings.ensure_agent_dir(assistant_id) agent_md = agent_dir / "AGENTS.md" if not agent_md.exists(): # Create empty file for user customizations # Base instructions are loaded fresh from get_system_prompt() agent_md.touch() # Skills directories (if enabled) skills_dir = None user_agent_skills_dir = None project_skills_dir = None project_agent_skills_dir = None if enable_skills: skills_dir = settings.ensure_user_skills_dir(assistant_id) user_agent_skills_dir = settings.get_user_agent_skills_dir() project_skills_dir = ( project_context.project_skills_dir() if project_context is not None else settings.get_project_skills_dir() ) project_agent_skills_dir = ( project_context.project_agent_skills_dir() if project_context is not None else settings.get_project_agent_skills_dir() ) # Load custom subagents from filesystem custom_subagents: list[SubAgent | CompiledSubAgent] = [] restrictive_shell_allow_list: list[str] | None = None if interrupt_shell_only and not auto_approve: # Prefer the explicitly forwarded allow-list (set by the CLI process # and passed through ServerConfig). Fall back to settings only for # direct callers (e.g. benchmarking frameworks) that don't go through # the server subprocess path. if shell_allow_list: restrictive_shell_allow_list = list(shell_allow_list) elif settings.shell_allow_list and not isinstance( settings.shell_allow_list, _ShellAllowAll ): restrictive_shell_allow_list = list(settings.shell_allow_list) else: logger.warning( "interrupt_shell_only=True but no restrictive shell allow-list " "available; falling back to standard HITL interrupts" ) user_agents_dir = settings.get_user_agents_dir(assistant_id) project_agents_dir = ( project_context.project_agents_dir() if project_context is not None else settings.get_project_agents_dir() ) def _subagent_cli_middleware(*, has_explicit_model: bool) -> list[AgentMiddleware]: middleware: list[AgentMiddleware] = [] if not has_explicit_model: middleware.append(ConfigurableModelMiddleware()) if restrictive_shell_allow_list is not None: middleware.append(ShellAllowListMiddleware(restrictive_shell_allow_list)) # Subagents share the on-disk filesystem backend and can edit the user # AGENTS.md, so they get the same managed onboarding-name block guard as # the main agent. Gated on memory because the block only exists when # memory is enabled. if enable_memory: from deepagents_code.memory_guard import ManagedMemoryGuardMiddleware middleware.append( ManagedMemoryGuardMiddleware( [settings.get_user_agent_md_path(assistant_id)] ) ) return middleware for subagent_meta in list_subagents( user_agents_dir=user_agents_dir, project_agents_dir=project_agents_dir, ): # Treat a falsy spec (`None` or `""`) as "no explicit model" so an empty # `model:` in subagent frontmatter inherits the runtime model rather than # being forwarded verbatim to `resolve_model("")`. model_spec = subagent_meta["model"] has_explicit_model = bool(model_spec) subagent: SubAgent = { "name": subagent_meta["name"], "description": subagent_meta["description"], "system_prompt": subagent_meta["system_prompt"], } if model_spec: subagent["model"] = model_spec subagent_middleware = _subagent_cli_middleware( has_explicit_model=has_explicit_model ) if subagent_middleware: subagent["middleware"] = subagent_middleware custom_subagents.append(subagent) from deepagents.middleware.subagents import ( GENERAL_PURPOSE_SUBAGENT, SubAgent as RuntimeSubAgent, ) if not any( subagent["name"] == GENERAL_PURPOSE_SUBAGENT["name"] for subagent in custom_subagents ): general_purpose_subagent: RuntimeSubAgent = { "name": GENERAL_PURPOSE_SUBAGENT["name"], "description": GENERAL_PURPOSE_SUBAGENT["description"], "system_prompt": GENERAL_PURPOSE_SUBAGENT["system_prompt"], "middleware": _subagent_cli_middleware(has_explicit_model=False), } custom_subagents.append(general_purpose_subagent) # Build middleware stack based on enabled features agent_middleware: list[AgentMiddleware[Any, Any]] = [ ConfigurableModelMiddleware(), _FilesystemEmptyResultMiddleware(), ] # Resume state: declares the `_context_tokens` and `_model_spec` channels # and writes them from `after_model` (token count from the latest # `AIMessage.usage_metadata`, model spec from `context["effective_model"]`). # The CLI reads them back from `state_values` on thread resume. from deepagents_code.resume_state import ResumeStateMiddleware agent_middleware.append(ResumeStateMiddleware()) # Add ask_user middleware (must be early so its tool is available) if enable_ask_user: from deepagents_code.ask_user import AskUserMiddleware agent_middleware.append(AskUserMiddleware()) # Add memory middleware if enable_memory: memory_sources = [str(settings.get_user_agent_md_path(assistant_id))] project_agent_md_paths = ( project_context.project_agent_md_paths() if project_context is not None else settings.get_project_agent_md_path() ) memory_sources.extend(str(p) for p in project_agent_md_paths) agent_middleware.append( MemoryMiddleware( backend=FilesystemBackend(virtual_mode=False), sources=memory_sources, ) ) # Protect the machine-managed onboarding-name block in the user # AGENTS.md from being rewritten by agent file edits. The block's # markers are HTML comments stripped before injection, so the model # can't see the boundary and would otherwise clobber it. from deepagents_code.memory_guard import ManagedMemoryGuardMiddleware agent_middleware.append( ManagedMemoryGuardMiddleware( [settings.get_user_agent_md_path(assistant_id)] ) ) # Add skills middleware if enable_skills: # Lowest to highest precedence: # built-in -> user .deepagents -> user .agents # -> project .deepagents -> project .agents # -> user .claude (experimental) -> project .claude (experimental) # Labels disambiguate user- vs project-scoped sources that share a # `.../skills` leaf; the middleware would otherwise derive identical # labels from the parent directory name. sources: list[tuple[str, str]] = [ (str(settings.get_built_in_skills_dir()), "Built-in"), (str(skills_dir), "User Deepagents"), (str(user_agent_skills_dir), "User Agents"), ] if project_skills_dir: sources.append((str(project_skills_dir), "Project Deepagents")) if project_agent_skills_dir: sources.append((str(project_agent_skills_dir), "Project Agents")) # Experimental: Claude Code skill directories user_claude_skills_dir = settings.get_user_claude_skills_dir() if user_claude_skills_dir.exists(): sources.append((str(user_claude_skills_dir), "User Claude")) project_claude_skills_dir = settings.get_project_claude_skills_dir() if project_claude_skills_dir: sources.append((str(project_claude_skills_dir), "Project Claude")) # Backwards-compat: strip labels when the installed SDK is too old # to accept `(path, label)` tuples. Label-based disambiguation # regresses to the pre-alias behavior (user- and project-scoped # `.claude/skills` collapse to the same label), but functionality # is preserved. middleware_sources: Sequence[str | tuple[str, str]] = ( sources if _SUPPORTS_SKILL_SOURCE_TUPLES else [path for path, _ in sources] ) agent_middleware.append( SkillsMiddleware( backend=FilesystemBackend(virtual_mode=False), sources=middleware_sources, ) ) # CONDITIONAL SETUP: Local vs Remote Sandbox if sandbox is None: # ========== LOCAL MODE ========== root_dir = effective_cwd if effective_cwd is not None else Path.cwd() if enable_shell: # Create environment for shell commands # Restore user's original LANGSMITH_PROJECT so their code traces separately shell_env = os.environ.copy() if settings.user_langchain_project: shell_env["LANGSMITH_PROJECT"] = settings.user_langchain_project # Re-apply a launch-time PYTHONPATH that was stripped from the server # interpreter but relayed for approval-gated `execute` commands. _apply_inherited_pythonpath(shell_env) # Use LocalShellBackend for filesystem + shell execution. # The SDK's FilesystemMiddleware exposes per-command timeout # on the execute tool natively. # `inherit_env=False`: `shell_env` is already a complete, curated # copy of `os.environ`. Inheriting again would re-copy `os.environ` # and resurrect the popped carrier var, leaking it into `execute`. backend = LocalShellBackend( root_dir=root_dir, inherit_env=False, env=shell_env, ) else: # No shell access - use plain FilesystemBackend backend = FilesystemBackend(root_dir=root_dir, virtual_mode=False) else: # ========== REMOTE SANDBOX MODE ========== backend = sandbox # Remote sandbox (ModalSandbox, etc.) # Note: Shell middleware not used in sandbox mode # File operations and execute tool are provided by the sandbox backend if enable_interpreter: if sandbox is not None: msg = ( "enable_interpreter=True is not supported with a remote " "sandbox in this release. Disable the sandbox or unset " "enable_interpreter." ) raise ValueError(msg) # Lazy import keeps `dcode -v` fast — see AGENTS.md startup-perf rule. from langchain_quickjs import CodeInterpreterMiddleware, PTCOption ptc_names = _resolve_ptc_option( settings.interpreter_ptc, tools=tools, acknowledge_unsafe=settings.interpreter_ptc_acknowledge_unsafe, auto_approve=auto_approve, ) ptc_option: PTCOption | None = ( cast("PTCOption", list(ptc_names)) if ptc_names is not None else None ) agent_middleware.append( CodeInterpreterMiddleware( tool_name="js_eval", timeout=settings.interpreter_timeout_seconds, memory_limit=settings.interpreter_memory_limit_mb * 1024 * 1024, max_ptc_calls=settings.interpreter_max_ptc_calls, max_result_chars=settings.interpreter_max_result_chars, ptc=ptc_option, ) ) # Local context middleware (git info, directory tree, etc.). if isinstance(backend, (_ExecutableBackend, _AsyncExecutableBackend)): agent_middleware.append( LocalContextMiddleware( backend=backend, mcp_server_info=mcp_server_info, tracing_project=get_langsmith_project_name(), user_tracing_project=settings.user_langchain_project, ) ) # Add shell allow-list middleware when interrupt_shell_only is active. shell_middleware_added = False if restrictive_shell_allow_list is not None: agent_middleware.append(ShellAllowListMiddleware(restrictive_shell_allow_list)) shell_middleware_added = True # Get or use custom system prompt if system_prompt is None: system_prompt = get_system_prompt( assistant_id=assistant_id, sandbox_type=sandbox_type, interactive=interactive, cwd=effective_cwd, ) # Configure interrupt_on based on auto_approve / shell_middleware_added interrupt_on: dict[str, bool | InterruptOnConfig] | None = None if auto_approve or shell_middleware_added: # noqa: SIM108 # if-else clearer than ternary for dual-path config # No HITL interrupts — tools run automatically. # When shell_middleware_added is True, shell validation is handled by # ShellAllowListMiddleware (added above) which rejects disallowed # commands inline as error ToolMessages, keeping the entire run in # a single LangSmith trace. interrupt_on = {} else: # Full HITL for destructive operations interrupt_on = _add_interrupt_on() # ty: ignore[invalid-assignment] # InterruptOnConfig is compatible at runtime # Set up composite backend with routing # For local FilesystemBackend, route large tool results to /tmp to avoid polluting # the working directory. For sandbox backends, no special routing is needed. if sandbox is None: # Local mode: Route large results to a unique temp directory large_results_backend = FilesystemBackend( root_dir=tempfile.mkdtemp(prefix="deepagents_large_results_"), virtual_mode=True, ) conversation_history_backend = FilesystemBackend( root_dir=tempfile.mkdtemp(prefix="deepagents_conversation_history_"), virtual_mode=True, ) composite_backend = CompositeBackend( default=backend, routes={ "/large_tool_results/": large_results_backend, "/conversation_history/": conversation_history_backend, }, ) else: # Sandbox mode: No special routing needed composite_backend = CompositeBackend( default=backend, routes={}, ) from deepagents.middleware.summarization import create_summarization_tool_middleware agent_middleware.append( create_summarization_tool_middleware(model, composite_backend) ) # Create the agent all_subagents: list[SubAgent | CompiledSubAgent | AsyncSubAgent] = [ *custom_subagents, *(async_subagents or []), ] agent = create_deep_agent( model=model, system_prompt=system_prompt, tools=tools, backend=composite_backend, middleware=agent_middleware, interrupt_on=interrupt_on, context_schema=CLIContextSchema, checkpointer=checkpointer, subagents=all_subagents or None, name=_sanitize_agent_message_name(assistant_id), ).with_config(config) return agent, composite_backend