DeepAgent module

`DeepAgent`

Bases: Module

A coding agent with filesystem and shell access scoped to a workdir.

DeepAgent is a thin specialization of :class:FunctionCallingAgent that pre-wires up to six workspace tools:

read_file: read a file with line-based pagination, output prefixed with 1-based line numbers (cat -n style).
list_directory: list entries in a directory.
search_files: glob for files and optionally grep their contents (regex). Combines find and grep in one call.
write_file: overwrite or create a file (gated by allow_write).
edit_file: exact-string replacement, one occurrence at a time (gated by allow_write).
run_bash: run a shell command (gated by allow_bash).

The constructor mirrors :class:FunctionCallingAgent — every parameter on that class is accepted here with identical semantics. The only additions are workdir (required) and the safety knobs: allow_write, allow_bash, timeout, max_output_chars. User-supplied tools are appended to the built-in ones.

Security model

File tools (read_file / write_file / edit_file / list_directory) refuse any path that resolves outside the workdir, including .. traversal and absolute paths. Paths are canonicalized via Path.resolve() (which flattens .. and follows existing symlinks) and then prefix-checked against the resolved workdir, so a symlink-inside-workdir pointing to /etc/passwd is also caught. File opens use O_NOFOLLOW where the OS supports it as defense in depth against TOCTOU symlink swaps.

The bash tool is NOT sandboxed. Its cwd is the workdir, but the shell can still read or write any path the host process can. If you're running this on untrusted input, run the host process inside a container or other OS-level isolation; the Python layer cannot make run_bash safe on its own. Disable it with allow_bash=False when you don't need it.

Example:

import synalinks
import asyncio

async def main():
    lm = synalinks.LanguageModel(model="ollama/mistral")

    inputs = synalinks.Input(data_model=synalinks.ChatMessages)
    outputs = await synalinks.DeepAgent(
        workdir="/tmp/my_project",
        language_model=lm,
    )(inputs)
    agent = synalinks.Program(inputs=inputs, outputs=outputs)

    messages = synalinks.ChatMessages(messages=[
        synalinks.ChatMessage(
            role="user",
            content="What's in this directory?",
        )
    ])
    result = await agent(messages)
    print(result.get("messages")[-1].get("content"))

asyncio.run(main())

Parameters:

Name	Type	Description	Default
`workdir`	`str`	Working directory the agent operates in. Required. Must exist. All file paths supplied by the LM are resolved relative to it and rejected if they escape.	required
`allow_write`	`bool`	When `False`, `write_file` and `edit_file` are omitted from the tool set. Defaults to `True`.	`True`
`allow_bash`	`bool`	When `False`, `run_bash` is omitted. Defaults to `True`.	`True`
`timeout`	`float`	Per-command bash timeout in seconds. Defaults to 30.	`30.0`
`max_output_chars`	`int`	Cap on characters returned per stream from `read_file` (single stream) and `run_bash` (stdout and stderr each). Also caps the length of each matching line returned by `search_files`. Defaults to 10000.	`10000`
`max_search_results`	`int`	Cap on entries returned by `search_files` (matching files or matching lines). Defaults to 100.	`100`
`tools`	`list`	Additional :class:`Tool` instances (or plain async functions) to expose alongside the built-in tools. Names must not start with `_` or collide with built-ins.	`None`
`schema`	`dict`	JSON schema for the final answer.	`None`
`data_model`	`DataModel`	DataModel for the final answer. Mutually exclusive with `schema`.	`None`
`language_model`	`LanguageModel`	The language model that drives the agent loop.	`None`
`prompt_template`	`str`	Forwarded to the tool-call generator.	`None`
`examples`	`list`	Few-shot examples for the tool-call generator.	`None`
`instructions`	`str`	Override the default system instructions. When omitted, the default is built from the workdir and the configured permissions.	`None`
`final_instructions`	`str`	Instructions for the final-answer generator. Defaults to `instructions`.	`None`
`temperature`	`float`	LM sampling temperature. Defaults to 0.0.	`0.0`
`use_inputs_schema`	`bool`	Include the input schema in the prompt.	`False`
`use_outputs_schema`	`bool`	Include the output schema in the prompt.	`False`
`reasoning_effort`	`str`	Forwarded to the generators (for reasoning-capable LMs).	`None`
`use_chain_of_thought`	`bool`	When `True`, the tool-call generator emits a `thinking` field per round.	`False`
`autonomous`	`bool`	When `True` (default), the agent runs the tool loop end-to-end. When `False`, returns one step at a time for human-in-the-loop workflows.	`True`
`return_inputs_with_trajectory`	`bool`	When `True` (default), the full message trajectory is included alongside the final answer.	`True`
`max_iterations`	`int`	Maximum number of tool-call rounds. Defaults to 10 (coding tasks tend to need more rounds than RAG / SQL).	`10`
`streaming`	`bool`	Stream the final answer when no `schema` is set. Defaults to `False`.	`False`
`name`	`str`	Module name.	`None`
`description`	`str`	Module description.	`None`

Source code in synalinks/src/modules/agents/deep_agent.py

@synalinks_export(
    [
        "synalinks.modules.DeepAgent",
        "synalinks.DeepAgent",
    ]
)
class DeepAgent(Module):
    """A coding agent with filesystem and shell access scoped to a workdir.

    DeepAgent is a thin specialization of :class:`FunctionCallingAgent`
    that pre-wires up to six workspace tools:

    - ``read_file``: read a file with line-based pagination, output
      prefixed with 1-based line numbers (``cat -n`` style).
    - ``list_directory``: list entries in a directory.
    - ``search_files``: glob for files and optionally grep their
      contents (regex). Combines find and grep in one call.
    - ``write_file``: overwrite or create a file (gated by ``allow_write``).
    - ``edit_file``: exact-string replacement, one occurrence at a time
      (gated by ``allow_write``).
    - ``run_bash``: run a shell command (gated by ``allow_bash``).

    The constructor mirrors :class:`FunctionCallingAgent` — every
    parameter on that class is accepted here with identical semantics.
    The only additions are ``workdir`` (required) and the safety
    knobs: ``allow_write``, ``allow_bash``, ``timeout``,
    ``max_output_chars``. User-supplied ``tools`` are appended to the
    built-in ones.

    ## Security model

    File tools (``read_file`` / ``write_file`` / ``edit_file`` /
    ``list_directory``) refuse any path that resolves outside the
    workdir, including ``..`` traversal and absolute paths. Paths are
    canonicalized via ``Path.resolve()`` (which flattens ``..`` and
    follows existing symlinks) and then prefix-checked against the
    resolved workdir, so a symlink-inside-workdir pointing to
    ``/etc/passwd`` is also caught. File opens use ``O_NOFOLLOW``
    where the OS supports it as defense in depth against TOCTOU
    symlink swaps.

    The bash tool is **NOT sandboxed**. Its ``cwd`` is the workdir,
    but the shell can still read or write any path the host process
    can. If you're running this on untrusted input, run the host
    process inside a container or other OS-level isolation; the
    Python layer cannot make ``run_bash`` safe on its own. Disable
    it with ``allow_bash=False`` when you don't need it.

    Example:

    ```python
    import synalinks
    import asyncio

    async def main():
        lm = synalinks.LanguageModel(model="ollama/mistral")

        inputs = synalinks.Input(data_model=synalinks.ChatMessages)
        outputs = await synalinks.DeepAgent(
            workdir="/tmp/my_project",
            language_model=lm,
        )(inputs)
        agent = synalinks.Program(inputs=inputs, outputs=outputs)

        messages = synalinks.ChatMessages(messages=[
            synalinks.ChatMessage(
                role="user",
                content="What's in this directory?",
            )
        ])
        result = await agent(messages)
        print(result.get("messages")[-1].get("content"))

    asyncio.run(main())
    ```

    Args:
        workdir (str): Working directory the agent operates in.
            Required. Must exist. All file paths supplied by the LM
            are resolved relative to it and rejected if they escape.
        allow_write (bool): When ``False``, ``write_file`` and
            ``edit_file`` are omitted from the tool set. Defaults to
            ``True``.
        allow_bash (bool): When ``False``, ``run_bash`` is omitted.
            Defaults to ``True``.
        timeout (float): Per-command bash timeout in seconds.
            Defaults to 30.
        max_output_chars (int): Cap on characters returned per
            stream from ``read_file`` (single stream) and ``run_bash``
            (stdout and stderr each). Also caps the length of each
            matching line returned by ``search_files``. Defaults to
            10000.
        max_search_results (int): Cap on entries returned by
            ``search_files`` (matching files or matching lines).
            Defaults to 100.
        tools (list): Additional :class:`Tool` instances (or plain
            async functions) to expose alongside the built-in tools.
            Names must not start with ``_`` or collide with built-ins.
        schema (dict): JSON schema for the final answer.
        data_model (DataModel): DataModel for the final answer.
            Mutually exclusive with ``schema``.
        language_model (LanguageModel): The language model that drives
            the agent loop.
        prompt_template (str): Forwarded to the tool-call generator.
        examples (list): Few-shot examples for the tool-call generator.
        instructions (str): Override the default system instructions.
            When omitted, the default is built from the workdir and
            the configured permissions.
        final_instructions (str): Instructions for the final-answer
            generator. Defaults to ``instructions``.
        temperature (float): LM sampling temperature. Defaults to 0.0.
        use_inputs_schema (bool): Include the input schema in the
            prompt.
        use_outputs_schema (bool): Include the output schema in the
            prompt.
        reasoning_effort (str): Forwarded to the generators (for
            reasoning-capable LMs).
        use_chain_of_thought (bool): When ``True``, the tool-call
            generator emits a ``thinking`` field per round.
        autonomous (bool): When ``True`` (default), the agent runs
            the tool loop end-to-end. When ``False``, returns one
            step at a time for human-in-the-loop workflows.
        return_inputs_with_trajectory (bool): When ``True`` (default),
            the full message trajectory is included alongside the
            final answer.
        max_iterations (int): Maximum number of tool-call rounds.
            Defaults to 10 (coding tasks tend to need more rounds
            than RAG / SQL).
        streaming (bool): Stream the final answer when no ``schema``
            is set. Defaults to ``False``.
        name (str): Module name.
        description (str): Module description.
    """

    def __init__(
        self,
        *,
        workdir: str,
        allow_write: bool = True,
        allow_bash: bool = True,
        timeout: float = 30.0,
        max_output_chars: int = 10_000,
        max_search_results: int = 100,
        tools: Optional[List] = None,
        schema=None,
        data_model=None,
        language_model=None,
        prompt_template=None,
        examples=None,
        instructions: Optional[str] = None,
        final_instructions: Optional[str] = None,
        temperature: float = 0.0,
        use_inputs_schema: bool = False,
        use_outputs_schema: bool = False,
        reasoning_effort: Optional[str] = None,
        use_chain_of_thought: bool = False,
        autonomous: bool = True,
        return_inputs_with_trajectory: bool = True,
        max_iterations: int = 10,
        streaming: bool = False,
        name: Optional[str] = None,
        description: Optional[str] = None,
    ):
        super().__init__(name=name, description=description)

        if not workdir:
            raise ValueError("`workdir` is required")
        resolved_workdir = Path(workdir).resolve()
        if not resolved_workdir.exists():
            raise ValueError(f"workdir does not exist: {workdir}")
        if not resolved_workdir.is_dir():
            raise ValueError(f"workdir is not a directory: {workdir}")
        self.workdir = str(resolved_workdir)

        if not isinstance(timeout, (int, float)) or timeout <= 0:
            raise ValueError(f"`timeout` must be a positive number, got {timeout!r}")
        self.timeout = float(timeout)

        if not isinstance(max_output_chars, int) or max_output_chars < 1:
            raise ValueError(
                f"`max_output_chars` must be a positive integer, got {max_output_chars!r}"
            )
        self.max_output_chars = max_output_chars

        if not isinstance(max_search_results, int) or max_search_results < 1:
            raise ValueError(
                f"`max_search_results` must be a positive integer, "
                f"got {max_search_results!r}"
            )
        self.max_search_results = max_search_results

        self.allow_write = bool(allow_write)
        self.allow_bash = bool(allow_bash)

        self.language_model = _get_lm(language_model)

        if not schema and data_model:
            schema = data_model.get_schema()
        self.schema = schema

        if instructions is None:
            instructions = get_default_instructions(
                self.workdir, self.allow_write, self.allow_bash
            )
        self.instructions = instructions
        self.final_instructions = final_instructions

        self.prompt_template = prompt_template
        self.examples = examples
        self.temperature = temperature
        self.use_inputs_schema = use_inputs_schema
        self.use_outputs_schema = use_outputs_schema
        self.reasoning_effort = reasoning_effort
        self.use_chain_of_thought = use_chain_of_thought
        self.autonomous = autonomous
        self.return_inputs_with_trajectory = return_inputs_with_trajectory
        self.max_iterations = max_iterations
        self.streaming = streaming

        builtin_tools = [
            Tool(fn)
            for fn in _build_tools(
                resolved_workdir,
                allow_write=self.allow_write,
                allow_bash=self.allow_bash,
                timeout=self.timeout,
                max_output_chars=self.max_output_chars,
                max_search_results=self.max_search_results,
            )
        ]
        builtin_names = {t.name for t in builtin_tools}

        self.extra_tools = list(tools) if tools else []
        merged_tools = list(builtin_tools)
        for extra in self.extra_tools:
            extra_tool = extra if isinstance(extra, Tool) else Tool(extra)
            if extra_tool.name in builtin_names:
                raise ValueError(
                    f"Tool name {extra_tool.name!r} collides with a built-in "
                    f"deep-agent tool. Rename the additional tool."
                )
            merged_tools.append(extra_tool)
        # Leading-underscore check is centralized in FunctionCallingAgent.

        self.agent = FunctionCallingAgent(
            schema=self.schema,
            language_model=self.language_model,
            prompt_template=self.prompt_template,
            examples=self.examples,
            instructions=self.instructions,
            final_instructions=self.final_instructions,
            temperature=self.temperature,
            use_inputs_schema=self.use_inputs_schema,
            use_outputs_schema=self.use_outputs_schema,
            reasoning_effort=self.reasoning_effort,
            use_chain_of_thought=self.use_chain_of_thought,
            tools=merged_tools,
            autonomous=self.autonomous,
            return_inputs_with_trajectory=self.return_inputs_with_trajectory,
            max_iterations=self.max_iterations,
            streaming=self.streaming,
            name="agent_" + self.name,
        )

    async def call(self, inputs, training=False):
        return await self.agent(inputs, training=training)

    async def compute_output_spec(self, inputs, training=False):
        return await self.agent.compute_output_spec(inputs, training=training)

    def get_config(self):
        config = {
            "workdir": self.workdir,
            "allow_write": self.allow_write,
            "allow_bash": self.allow_bash,
            "timeout": self.timeout,
            "max_output_chars": self.max_output_chars,
            "max_search_results": self.max_search_results,
            "schema": self.schema,
            "prompt_template": self.prompt_template,
            "examples": self.examples,
            "instructions": self.instructions,
            "final_instructions": self.final_instructions,
            "temperature": self.temperature,
            "use_inputs_schema": self.use_inputs_schema,
            "use_outputs_schema": self.use_outputs_schema,
            "reasoning_effort": self.reasoning_effort,
            "use_chain_of_thought": self.use_chain_of_thought,
            "autonomous": self.autonomous,
            "return_inputs_with_trajectory": self.return_inputs_with_trajectory,
            "max_iterations": self.max_iterations,
            "streaming": self.streaming,
            "name": self.name,
            "description": self.description,
        }
        language_model_config = {
            "language_model": serialization_lib.serialize_synalinks_object(
                self.language_model,
            )
        }
        tools_config = {
            "tools": [
                serialization_lib.serialize_synalinks_object(
                    t if isinstance(t, Tool) else Tool(t)
                )
                for t in self.extra_tools
            ]
        }
        return {**config, **language_model_config, **tools_config}

    @classmethod
    def from_config(cls, config):
        language_model = serialization_lib.deserialize_synalinks_object(
            config.pop("language_model")
        )
        tools = [
            serialization_lib.deserialize_synalinks_object(t)
            for t in config.pop("tools", [])
        ]
        return cls(
            language_model=language_model,
            tools=tools,
            **config,
        )

`PathTraversalError`

Bases: ValueError

Raised when a tool argument resolves outside the configured workdir.

Source code in synalinks/src/modules/agents/deep_agent.py

class PathTraversalError(ValueError):
    """Raised when a tool argument resolves outside the configured workdir."""

`get_default_instructions(workdir, allow_write, allow_bash)`

Default system instructions for the deep agent.

Parameters:

Name	Type	Description	Default
`workdir`	`str`	Absolute path of the agent's working directory. Embedded in the prompt so the LM knows where it's operating.	required
`allow_write`	`bool`	Whether write/edit tools are enabled.	required
`allow_bash`	`bool`	Whether the bash tool is enabled.	required

Returns:

Type	Description
`str`	A prompt string describing the tool plan and the safety
`str`	constraints currently in effect.

Source code in synalinks/src/modules/agents/deep_agent.py

def get_default_instructions(
    workdir: str,
    allow_write: bool,
    allow_bash: bool,
) -> str:
    """Default system instructions for the deep agent.

    Args:
        workdir: Absolute path of the agent's working directory.
            Embedded in the prompt so the LM knows where it's
            operating.
        allow_write: Whether write/edit tools are enabled.
        allow_bash: Whether the bash tool is enabled.

    Returns:
        A prompt string describing the tool plan and the safety
        constraints currently in effect.
    """
    capabilities = ["read_file", "list_directory", "search_files"]
    if allow_write:
        capabilities.extend(["write_file", "edit_file"])
    if allow_bash:
        capabilities.append("run_bash")

    extras = []
    if not allow_write:
        extras.append("Write/edit tools are DISABLED — this is a read-only session.")
    if not allow_bash:
        extras.append("Shell execution is DISABLED.")
    constraints = ("\n".join(f"- {line}" for line in extras) + "\n") if extras else ""

    return f"""
You are a software engineering assistant with filesystem and shell access
scoped to a single working directory.

Workdir: {workdir}
Available tools: {capabilities}

Plan:
1. Use `list_directory` to discover what's in the workdir.
2. Use `search_files` to locate files by glob and/or grep their contents.
3. Use `read_file` to read files. Output is line-numbered (``cat -n``
   style). Pages of lines via `offset` / `limit`; raise `offset` to read
   further into the file.
4. {"Use `edit_file` for surgical changes (preferred over `write_file`)." if allow_write else "Reads only — do not propose write operations."}
5. {"Use `run_bash` for builds, tests, and other shell work." if allow_bash else "Shell is disabled — solve tasks with file tools only."}
6. Once you have the answer, stop calling tools and respond.

Constraints:
- All paths must stay inside the workdir. ``..`` traversal and absolute
  paths that escape the workdir are rejected.
{constraints}""".strip()