Skip to content

Toolsets API

create_console_toolset

pydantic_ai_backends.toolsets.console.create_console_toolset(id=None, backend=None, include_execute=True, include_background=True, require_write_approval=False, require_execute_approval=True, default_ignore_hidden=True, permissions=None, ask_callback=None, ask_fallback='error', max_retries=1, image_support=False, max_image_bytes=DEFAULT_MAX_IMAGE_BYTES, document_support=False, max_document_bytes=DEFAULT_MAX_DOCUMENT_BYTES, edit_format='str_replace', descriptions=None, profile=DEFAULT_PROFILE)

Create a console toolset for file operations and shell execution.

Works with any backend implementing BackendProtocolLocalBackend, DockerSandbox, StateBackend and so on.

Parameters:

Name Type Description Default
id str | None

Optional unique ID for the toolset.

None
backend BackendProtocol | AsyncBackendProtocol | None

Backend every tool operates on. When omitted, each call reads ctx.deps.backend instead, which requires the agent's deps to satisfy :class:ConsoleDeps. Pass it explicitly when the host owns its own deps type and cannot add a backend field to it — a capability holding a sandbox, for instance.

None
include_execute bool

Include the execute tool. Requires a backend with an execute method.

True
include_background bool

Include the background-shell tools. Requires a backend implementing BackgroundSandboxProtocol.

True
require_write_approval bool

Whether write_file and the edit tool require approval. Ignored when permissions is given.

False
require_execute_approval bool

Whether execute requires approval. Ignored when permissions is given.

True
default_ignore_hidden bool

Default for grep's hidden-file handling.

True
permissions PermissionRuleset | None

Ruleset deciding which tools exist and which need approval: an operation defaulting to "deny" drops its tools entirely, one defaulting to "ask" marks them as requiring approval.

None
max_retries int

Times a tool may retry within one run, with the message fed back to the model — pydantic-ai's own argument validation, and the mistakes toolsets/_failures.py steers on: a missing file, an old_string that is absent or matches twice, a stale read. Past the budget the message is returned rather than raised, so a run never ends on one.

1
image_support bool

Return recognized image files (.png, .jpg, .jpeg, .gif, .webp) as BinaryContent a multimodal model can see, instead of garbled text.

False
max_image_bytes int

Largest image returned; bigger ones yield an error.

DEFAULT_MAX_IMAGE_BYTES
document_support bool

Return recognized documents (.pdf) as BinaryContent for models that understand documents natively. Kept separate from image_support so the two can evolve apart.

False
max_document_bytes int

Largest document returned; bigger ones yield an error.

DEFAULT_MAX_DOCUMENT_BYTES
edit_format EditFormat

"str_replace" matches exact strings; "hashline" tags each line with a content hash so the model references lines by number:hash instead of reproducing text.

'str_replace'
descriptions Mapping[str, str | ToolText] | None

Per-tool text overrides, keyed by tool name: ls, read_file, write_file, edit_file, hashline_edit, glob, grep, execute, run_in_background, read_output, kill_shell, list_shells. A string replaces the tool's description and leaves its argument text alone; a :class:ToolText replaces both. An unknown key raises UserError rather than being ignored, since a silent override is one nobody discovers.

None
profile Profile

How much guidance the descriptions carry. "coding" includes the guidance written for an agent working in a repository — git, dependencies, debugging a failed command — and "agent" leaves it out, which is about 250 tokens a request an agent with a scratch workspace was paying for advice it could not use.

DEFAULT_PROFILE
Example
Python
from dataclasses import dataclass

from pydantic_ai_backends import LocalBackend, create_console_toolset
from pydantic_ai_backends.permissions import DEFAULT_RULESET

@dataclass
class MyDeps:
    backend: LocalBackend

toolset = create_console_toolset()
deps = MyDeps(backend=LocalBackend("/workspace"))

hashline = create_console_toolset(edit_format="hashline")
multimodal = create_console_toolset(image_support=True, document_support=True)
guarded = create_console_toolset(permissions=DEFAULT_RULESET)
Source code in src/pydantic_ai_backends/toolsets/console.py
Python
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
def create_console_toolset(  # noqa: C901
    id: str | None = None,
    backend: BackendProtocol | AsyncBackendProtocol | None = None,
    include_execute: bool = True,
    include_background: bool = True,
    require_write_approval: bool = False,
    require_execute_approval: bool = True,
    default_ignore_hidden: bool = True,
    permissions: PermissionRuleset | None = None,
    ask_callback: AskCallback | None = None,
    ask_fallback: AskFallback = "error",
    max_retries: int = 1,
    image_support: bool = False,
    max_image_bytes: int = DEFAULT_MAX_IMAGE_BYTES,
    document_support: bool = False,
    max_document_bytes: int = DEFAULT_MAX_DOCUMENT_BYTES,
    edit_format: EditFormat = "str_replace",
    descriptions: Mapping[str, str | ToolText] | None = None,
    profile: Profile = DEFAULT_PROFILE,
) -> FunctionToolset[ConsoleDeps]:
    """Create a console toolset for file operations and shell execution.

    Works with any backend implementing `BackendProtocol` — `LocalBackend`,
    `DockerSandbox`, `StateBackend` and so on.

    Args:
        id: Optional unique ID for the toolset.
        backend: Backend every tool operates on. When omitted, each call reads
            `ctx.deps.backend` instead, which requires the agent's deps to
            satisfy :class:`ConsoleDeps`. Pass it explicitly when the host owns
            its own deps type and cannot add a `backend` field to it — a
            capability holding a sandbox, for instance.
        include_execute: Include the `execute` tool. Requires a backend with an
            `execute` method.
        include_background: Include the background-shell tools. Requires a
            backend implementing `BackgroundSandboxProtocol`.
        require_write_approval: Whether `write_file` and the edit tool require
            approval. Ignored when `permissions` is given.
        require_execute_approval: Whether `execute` requires approval. Ignored
            when `permissions` is given.
        default_ignore_hidden: Default for `grep`'s hidden-file handling.
        permissions: Ruleset deciding which tools exist and which need approval:
            an operation defaulting to "deny" drops its tools entirely, one
            defaulting to "ask" marks them as requiring approval.
        max_retries: Times a tool may retry within one run, with the message
            fed back to the model — pydantic-ai's own argument validation, and
            the mistakes `toolsets/_failures.py` steers on: a missing file, an
            `old_string` that is absent or matches twice, a stale read. Past the
            budget the message is returned rather than raised, so a run never
            ends on one.
        image_support: Return recognized image files (`.png`, `.jpg`, `.jpeg`,
            `.gif`, `.webp`) as `BinaryContent` a multimodal model can see,
            instead of garbled text.
        max_image_bytes: Largest image returned; bigger ones yield an error.
        document_support: Return recognized documents (`.pdf`) as
            `BinaryContent` for models that understand documents natively. Kept
            separate from `image_support` so the two can evolve apart.
        max_document_bytes: Largest document returned; bigger ones yield an error.
        edit_format: `"str_replace"` matches exact strings; `"hashline"` tags
            each line with a content hash so the model references lines by
            `number:hash` instead of reproducing text.
        descriptions: Per-tool text overrides, keyed by tool name: `ls`,
            `read_file`, `write_file`, `edit_file`, `hashline_edit`, `glob`,
            `grep`, `execute`, `run_in_background`, `read_output`, `kill_shell`,
            `list_shells`. A string replaces the tool's description and leaves
            its argument text alone; a :class:`ToolText` replaces both. An
            unknown key raises `UserError` rather than being ignored, since a
            silent override is one nobody discovers.
        profile: How much guidance the descriptions carry. `"coding"` includes
            the guidance written for an agent working in a repository — git,
            dependencies, debugging a failed command — and `"agent"` leaves it
            out, which is about 250 tokens a request an agent with a scratch
            workspace was paying for advice it could not use.

    Example:
        ```python
        from dataclasses import dataclass

        from pydantic_ai_backends import LocalBackend, create_console_toolset
        from pydantic_ai_backends.permissions import DEFAULT_RULESET

        @dataclass
        class MyDeps:
            backend: LocalBackend

        toolset = create_console_toolset()
        deps = MyDeps(backend=LocalBackend("/workspace"))

        hashline = create_console_toolset(edit_format="hashline")
        multimodal = create_console_toolset(image_support=True, document_support=True)
        guarded = create_console_toolset(permissions=DEFAULT_RULESET)
        ```
    """
    overrides: Mapping[str, str | ToolText] = descriptions or {}
    unknown = sorted(set(overrides) - OVERRIDE_KEYS)
    if unknown:
        raise UserError(
            f"Unknown tool name(s) in `descriptions`: {', '.join(unknown)}. "
            f"Valid names: {', '.join(sorted(OVERRIDE_KEYS))}."
        )

    # Wrapped once, here, because the closure backend never changes. `guarding`
    # answers with the backend untouched when there is no ruleset or when the
    # backend enforces one of its own, so this is a no-op for every existing
    # caller.
    guarded_backend = (
        None
        if backend is None
        else _guard.guarding(
            backend, permissions, ask_callback=ask_callback, ask_fallback=ask_fallback
        )
    )

    def backend_for(ctx: RunContext[ConsoleDeps]) -> BackendProtocol | AsyncBackendProtocol:
        """The backend this call operates on, with the ruleset applied to it.

        The one place every tool resolves its backend, which is why the guard goes
        here: a ruleset's per-path rules used to reach nothing at all, because the
        toolset only ever read an operation's *default* action at construction
        time. Applying them needs a path, and a path only exists per call.

        The deps backend is wrapped per call rather than once, since it can differ
        between runs. Cheap: the wrapper holds two references and a checker that
        does the same.
        """
        if backend is not None:
            return guarded_backend if guarded_backend is not None else backend
        return _guard.guarding(
            ctx.deps.backend,
            permissions,
            ask_callback=ask_callback,
            ask_fallback=ask_fallback,
        )

    write_approval = _ruleset.requires_approval(permissions, "write", require_write_approval)
    execute_approval = _ruleset.requires_approval(permissions, "execute", require_execute_approval)

    toolset: FunctionToolset[ConsoleDeps] = FunctionToolset(id=id, max_retries=max_retries)

    def described(
        text_id: str,
        tool_name: str | None = None,
        *,
        requires_approval: bool = False,
    ) -> Callable[[_ToolFn], _ToolFn]:
        """Register a tool with the text this configuration gives it.

        One `ToolText` supplies both halves of what the model reads, but they
        travel separately: the description is passed to the decorator, while the
        per-argument text reaches the JSON schema through the function's
        docstring and through nothing else — hence the assignment. It is also
        why the tools below carry a one-line docstring rather than a second copy
        of the argument text, which is a copy that drifts.

        Args:
            text_id: Key in `TOOL_TEXT`. Differs from the tool name only for
                `read_file`, which has one text per edit format.
            tool_name: Name a caller overrides this tool by, when it is not the
                text id.
            requires_approval: Whether the tool call is suspended for a human.
        """
        name = tool_name or text_id
        override = overrides.get(name)
        text = override if isinstance(override, ToolText) else TOOL_TEXT[text_id]
        description = override if isinstance(override, str) else text.render(profile)

        def register(fn: _ToolFn) -> _ToolFn:
            cast("Any", fn).__doc__ = text.docstring()
            registered = toolset.tool(description=description, requires_approval=requires_approval)(
                fn
            )
            return cast("_ToolFn", registered)

        return register

    async def binary_content(
        target: BackendProtocol | AsyncBackendProtocol, path: str
    ) -> Any | None:  # pragma: no cover - exercised through read_file
        """Image or document content for `path`, when either is enabled."""
        if image_support:
            image = await image_content(target, path, max_image_bytes)
            if image is not None:
                return image
        if document_support:
            return await document_content(target, path, max_document_bytes)
        return None

    @described("ls")
    @_degrade_on_error
    async def ls(
        ctx: RunContext[ConsoleDeps],
        path: str = ".",
    ) -> str:
        """List files and directories at the given path."""
        entries = await ensure_async(backend_for(ctx)).ls_info(path)
        if not entries:
            return f"Directory '{path}' is empty or does not exist"

        lines = [f"Contents of {path}:"]
        for entry in entries:
            if entry["is_dir"]:
                lines.append(f"  {entry['name']}/")
            else:
                size = entry.get("size")
                lines.append(f"  {entry['name']}{f' ({size} bytes)' if size is not None else ''}")
        return "\n".join(lines)

    if edit_format == "hashline":

        @described("hashline_read_file", "read_file")
        @_degrade_on_error
        async def read_file(
            ctx: RunContext[ConsoleDeps],
            path: str,
            offset: int = 0,
            limit: int = 2000,
        ) -> Any:
            """Read file content with hashline tags."""
            binary = await binary_content(backend_for(ctx), path)
            if binary is not None:
                return binary

            from pydantic_ai_backends.hashline import format_hashline_output

            backend = ensure_async(backend_for(ctx))
            if not await backend.exists(path):
                return _failures.steer(
                    ctx,
                    f"Error: File '{path}' not found. Check the path with `ls` or "
                    "`glob`, then read it again.",
                )

            raw = await backend.read_bytes(path)
            _tracking.record_read(backend_for(ctx), path, raw)
            return format_hashline_output(raw.decode("utf-8", errors="replace"), offset, limit)

    else:

        @described("read_file")
        @_degrade_on_error
        async def read_file(
            ctx: RunContext[ConsoleDeps],
            path: str,
            offset: int = 0,
            limit: int = 2000,
        ) -> Any:
            """Read file content with line numbers."""
            binary = await binary_content(backend_for(ctx), path)
            if binary is not None:
                return binary

            backend = ensure_async(backend_for(ctx))
            result = await backend.read(path, offset, limit)
            if result.startswith("Error"):
                return _failures.steer(ctx, result)
            await _tracking.record_path_read(backend, backend_for(ctx), path)
            return result

    @described("write_file", requires_approval=write_approval)
    @_degrade_on_error
    async def write_file(
        ctx: RunContext[ConsoleDeps],
        path: str,
        content: str,
    ) -> str:
        """Write content to a file."""
        result = await ensure_async(backend_for(ctx)).write(path, content)
        if result.error:
            return _failures.steer(ctx, f"Error: {result.error}")

        # The agent knows this file's content now, so an immediate edit must not
        # be refused as stale.
        _tracking.record_read(backend_for(ctx), path, content.encode("utf-8"))
        return f"Wrote {len(content.splitlines())} lines to {result.path}"

    if edit_format == "hashline":

        @described("hashline_edit", requires_approval=write_approval)
        @_degrade_on_error
        async def hashline_edit(
            ctx: RunContext[ConsoleDeps],
            path: str,
            start_line: int,
            start_hash: str,
            new_content: str,
            end_line: int | None = None,
            end_hash: str | None = None,
            insert_after: bool = False,
        ) -> str:
            """Edit a file by referencing lines with their content hashes."""
            from pydantic_ai_backends.hashline import apply_hashline_edit_with_summary

            raw_backend = backend_for(ctx)
            backend = ensure_async(raw_backend)

            async with _tracking.edit_lock(raw_backend, path):
                if not await backend.exists(path):
                    return _failures.steer(ctx, f"Error: File '{path}' not found")

                current = (await backend.read_bytes(path)).decode("utf-8", errors="replace")
                new_text, error, summary = apply_hashline_edit_with_summary(
                    current,
                    start_line,
                    start_hash,
                    new_content,
                    end_line,
                    end_hash,
                    insert_after,
                )
                if error:
                    return _failures.steer(ctx, f"Error: {error}")

                written = await backend.write(path, new_text)
                if written.error:
                    return _failures.steer(ctx, f"Error: {written.error}")
                return f"Edited {written.path}: {summary}"

    else:

        @described("edit_file", requires_approval=write_approval)
        @_degrade_on_error
        async def edit_file(
            ctx: RunContext[ConsoleDeps],
            path: str,
            old_string: str,
            new_string: str,
            replace_all: bool = False,
        ) -> str:
            """Edit a file by performing exact string replacement."""
            raw_backend = backend_for(ctx)
            backend = ensure_async(raw_backend)

            # Locked for the same reason `hashline_edit` is: every backend's
            # `edit` is a read, a replace and a write, so two edits to one path
            # in flight together lose one of them. The staleness check belongs
            # inside the lock too — checked outside, it is answered before the
            # other edit's write and passes on content that no longer exists.
            async with _tracking.edit_lock(raw_backend, path):
                stale = await _tracking.staleness_error(backend, raw_backend, path)
                if stale is not None:
                    return _failures.steer(ctx, stale)

                result = await backend.edit(path, old_string, new_string, replace_all)
                if result.error:
                    return _failures.steer(ctx, f"Error: {result.error}")

                # The agent's view is the post-edit content now, so a follow-up
                # edit must not be flagged as stale.
                await _tracking.record_path_read(backend, raw_backend, path)
                return f"Edited {result.path}: replaced {result.occurrences} occurrence(s)"

    @described("glob")
    @_degrade_on_error
    async def glob(
        ctx: RunContext[ConsoleDeps],
        pattern: str,
        path: str = ".",
    ) -> str:
        """Find files matching a glob pattern."""
        entries = await ensure_async(backend_for(ctx)).glob_info(pattern, path)
        if not entries:
            return f"No files matching '{pattern}' in {path}"

        lines = [f"Found {len(entries)} file(s) matching '{pattern}':"]
        lines.extend(f"  {entry['path']}" for entry in entries[:GLOB_RESULT_LIMIT])
        if len(entries) > GLOB_RESULT_LIMIT:
            lines.append(f"  ... and {len(entries) - GLOB_RESULT_LIMIT} more")
        return "\n".join(lines)

    @described("grep")
    @_degrade_on_error
    async def grep(
        ctx: RunContext[ConsoleDeps],
        pattern: str,
        path: str | None = None,
        glob_pattern: str | None = None,
        output_mode: Literal["content", "files_with_matches", "count"] = "files_with_matches",
        ignore_hidden: bool = default_ignore_hidden,
    ) -> str:
        """Search for a regex pattern across files."""
        result = await ensure_async(backend_for(ctx)).grep_raw(
            pattern, path, glob_pattern, ignore_hidden
        )
        if isinstance(result, str):
            return result
        if not result:
            return f"No matches for '{pattern}'"

        matches: list[GrepMatch] = result
        if output_mode == "count":
            return f"Found {len(matches)} match(es) for '{pattern}'"

        if output_mode == "files_with_matches":
            files = sorted({match["path"] for match in matches})
            return _truncated_list(f"Files containing '{pattern}':", files, "more files")

        rendered = [
            f"{m['path']}:{m['line_number']}: {m['line'][:GREP_LINE_WIDTH]}" for m in matches
        ]
        return _truncated_list(f"Matches for '{pattern}':", rendered, "more matches")

    # Exposed for the test suite.
    cast(_ConsoleToolsetTestAttrs, toolset)._console_default_ignore_hidden = default_ignore_hidden
    cast(_ConsoleToolsetTestAttrs, toolset)._console_grep_impl = grep

    if include_execute:

        @described("execute", requires_approval=execute_approval)
        @_degrade_on_error
        async def execute(
            ctx: RunContext[ConsoleDeps],
            command: str,
            timeout: int | None = DEFAULT_EXECUTE_TIMEOUT,
        ) -> str:
            """Execute a shell command in the working directory."""
            target = backend_for(ctx)
            async_backend = ensure_async(target)

            if not hasattr(async_backend, "execute"):
                return "Error: Backend does not support command execution"
            if hasattr(target, "execute_enabled") and not target.execute_enabled:  # pyright: ignore[reportAttributeAccessIssue]
                return "Error: Shell execution is disabled for this backend"

            result = await async_backend.execute(command, timeout)  # pyright: ignore[reportAttributeAccessIssue]

            output = result.output
            if result.truncated:
                output += "\n\n... (output truncated)"
            if result.exit_code is not None and result.exit_code != 0:
                return f"Command failed (exit code {result.exit_code}):\n{output}"
            return str(output)

        # Exposed for the test suite.
        cast(_ConsoleToolsetTestAttrs, toolset)._console_execute_impl = execute

    if include_execute and include_background:

        def background(ctx: RunContext[ConsoleDeps]) -> Any | None:
            """The async background sandbox, or `None` when unsupported."""
            backend = ensure_async(backend_for(ctx))
            return backend if hasattr(backend, "execute_background") else None

        @described("run_in_background", requires_approval=execute_approval)
        @_degrade_on_error
        async def run_in_background(
            ctx: RunContext[ConsoleDeps],
            command: str,
        ) -> str:
            """Start a long-lived command in the background."""
            sandbox = background(ctx)
            if sandbox is None:
                return _NO_BACKGROUND_SUPPORT
            handle = await sandbox.execute_background(command)
            return (
                f"Started background shell {handle.shell_id} (pid {handle.pid}).\n"
                f"Use read_output('{handle.shell_id}') to follow its output and "
                f"kill_shell('{handle.shell_id}') to stop it."
            )

        @described("read_output")
        @_degrade_on_error
        async def read_output(
            ctx: RunContext[ConsoleDeps],
            shell_id: str,
        ) -> str:
            """Read new output from a background shell."""
            sandbox = background(ctx)
            if sandbox is None:
                return _NO_BACKGROUND_SUPPORT

            result = await sandbox.read_background(shell_id)
            status = "running" if result.running else f"exited (code {result.exit_code})"
            body = (result.stdout + result.stderr).strip() or "(no new output)"
            return f"[{result.shell_id}] {status}\n{body}"

        @described("kill_shell", requires_approval=execute_approval)
        @_degrade_on_error
        async def kill_shell(
            ctx: RunContext[ConsoleDeps],
            shell_id: str,
        ) -> str:
            """Stop a background shell."""
            sandbox = background(ctx)
            if sandbox is None:
                return _NO_BACKGROUND_SUPPORT
            if await sandbox.kill_background(shell_id):
                return f"Killed background shell {shell_id}."
            return f"Background shell {shell_id} was already finished or unknown."

        @described("list_shells")
        @_degrade_on_error
        async def list_shells(
            ctx: RunContext[ConsoleDeps],
        ) -> str:
            """List the background shells started this session."""
            sandbox = background(ctx)
            if sandbox is None:
                return _NO_BACKGROUND_SUPPORT

            infos = await sandbox.list_background()
            if not infos:
                return "No background shells."
            return "\n".join(
                f"{i.shell_id}  {'running' if i.running else f'exited({i.exit_code})'}  {i.command}"
                for i in infos
            )

    for tool_name in _denied_tools(permissions):
        toolset.tools.pop(tool_name, None)

    return toolset

get_console_system_prompt

pydantic_ai_backends.toolsets.console.get_console_system_prompt(edit_format='str_replace')

The system prompt describing the console tools.

Parameters:

Name Type Description Default
edit_format EditFormat

Which edit format to describe.

'str_replace'
Source code in src/pydantic_ai_backends/toolsets/console.py
Python
def get_console_system_prompt(edit_format: EditFormat = "str_replace") -> str:
    """The system prompt describing the console tools.

    Args:
        edit_format: Which edit format to describe.
    """
    if edit_format == "hashline":
        return HASHLINE_CONSOLE_PROMPT
    return CONSOLE_SYSTEM_PROMPT

ConsoleDeps

pydantic_ai_backends.toolsets.console.ConsoleDeps

Bases: Protocol

Dependencies that provide a backend for the console tools.

Source code in src/pydantic_ai_backends/toolsets/console.py
Python
@runtime_checkable
class ConsoleDeps(Protocol):
    """Dependencies that provide a backend for the console tools."""

    @property
    def backend(self) -> BackendProtocol | AsyncBackendProtocol:
        """The backend for file operations."""
        ...

backend property

The backend for file operations.

ToolText

pydantic_ai_backends.toolsets.descriptions.ToolText dataclass

Everything the model reads about one tool.

Held as fields rather than one string because the parts have different destinations: summary, usage, coding and returns are composed into the tool's description, while args becomes the per-argument text in its JSON schema. Splitting them is also what lets a host show summary in its own catalogue and know it is the first sentence the model reads, rather than a paraphrase written in another repository.

Source code in src/pydantic_ai_backends/toolsets/descriptions.py
Python
@dataclass(frozen=True)
class ToolText:
    """Everything the model reads about one tool.

    Held as fields rather than one string because the parts have different
    destinations: `summary`, `usage`, `coding` and `returns` are composed into
    the tool's description, while `args` becomes the per-argument text in its
    JSON schema. Splitting them is also what lets a host show `summary` in its
    own catalogue and know it is the first sentence the model reads, rather than
    a paraphrase written in another repository.
    """

    summary: str
    """One sentence: what the tool does. Also what a catalogue should show."""

    usage: str = ""
    """When to use it, when to use another tool, and what it will not do."""

    coding: str = ""
    """Guidance only an agent working in a repository needs.

    Rendered under the `"coding"` profile and omitted under `"agent"`. Anything
    true of any workspace belongs in `usage` instead.
    """

    args: Mapping[str, str] = field(default_factory=dict)
    """One entry per argument, keyed exactly as the parameter is named."""

    returns: str = ""
    """The shape of the result, including its failures and its truncation."""

    def render(self, profile: Profile = DEFAULT_PROFILE) -> str:
        """The description handed to the model.

        Shaped the way pydantic-ai shapes a docstring that has a `Returns:`
        section - the prose inside `<summary>`, the return description inside
        `<returns>` - because that is what every tool built from a docstring
        already sends, and a host registering these beside its own would
        otherwise put two conventions in one tool list. A prose `Returns:`
        paragraph was the first attempt and is what that inconsistency looked
        like. `tests/test_tool_text.py` pins the shape against a tool the
        framework renders itself, so a change there fails here rather than
        drifting quietly.

        Args:
            profile: Which audience to write for.
        """
        parts = [self.summary]
        if self.usage:
            parts.append(self.usage)
        if self.coding and profile == "coding":
            parts.append(self.coding)
        body = "\n\n".join(parts)
        if not self.returns:
            return body
        return (
            f"<summary>{body}</summary>\n"
            f"<returns>\n<description>{self.returns}</description>\n</returns>"
        )

    def docstring(self) -> str:
        """A Google-style docstring carrying the argument text.

        Set on the tool function before it is registered, because per-argument
        descriptions reach the JSON schema through the docstring and through
        nothing else — which is why they cannot simply live in `render`. The
        summary is repeated here for a reader of the generated docstring; the
        model reads `render` instead, since an explicit description wins over
        the docstring's own summary.
        """
        lines = [self.summary]
        if self.args:
            lines.append("")
            lines.append("Args:")
            lines.extend(f"    {name}: {text}" for name, text in self.args.items())
        return "\n".join(lines)

summary instance-attribute

One sentence: what the tool does. Also what a catalogue should show.

usage = '' class-attribute instance-attribute

When to use it, when to use another tool, and what it will not do.

coding = '' class-attribute instance-attribute

Guidance only an agent working in a repository needs.

Rendered under the "coding" profile and omitted under "agent". Anything true of any workspace belongs in usage instead.

args = field(default_factory=dict) class-attribute instance-attribute

One entry per argument, keyed exactly as the parameter is named.

returns = '' class-attribute instance-attribute

The shape of the result, including its failures and its truncation.

render(profile=DEFAULT_PROFILE)

The description handed to the model.

Shaped the way pydantic-ai shapes a docstring that has a Returns: section - the prose inside <summary>, the return description inside <returns> - because that is what every tool built from a docstring already sends, and a host registering these beside its own would otherwise put two conventions in one tool list. A prose Returns: paragraph was the first attempt and is what that inconsistency looked like. tests/test_tool_text.py pins the shape against a tool the framework renders itself, so a change there fails here rather than drifting quietly.

Parameters:

Name Type Description Default
profile Profile

Which audience to write for.

DEFAULT_PROFILE
Source code in src/pydantic_ai_backends/toolsets/descriptions.py
Python
def render(self, profile: Profile = DEFAULT_PROFILE) -> str:
    """The description handed to the model.

    Shaped the way pydantic-ai shapes a docstring that has a `Returns:`
    section - the prose inside `<summary>`, the return description inside
    `<returns>` - because that is what every tool built from a docstring
    already sends, and a host registering these beside its own would
    otherwise put two conventions in one tool list. A prose `Returns:`
    paragraph was the first attempt and is what that inconsistency looked
    like. `tests/test_tool_text.py` pins the shape against a tool the
    framework renders itself, so a change there fails here rather than
    drifting quietly.

    Args:
        profile: Which audience to write for.
    """
    parts = [self.summary]
    if self.usage:
        parts.append(self.usage)
    if self.coding and profile == "coding":
        parts.append(self.coding)
    body = "\n\n".join(parts)
    if not self.returns:
        return body
    return (
        f"<summary>{body}</summary>\n"
        f"<returns>\n<description>{self.returns}</description>\n</returns>"
    )

docstring()

A Google-style docstring carrying the argument text.

Set on the tool function before it is registered, because per-argument descriptions reach the JSON schema through the docstring and through nothing else — which is why they cannot simply live in render. The summary is repeated here for a reader of the generated docstring; the model reads render instead, since an explicit description wins over the docstring's own summary.

Source code in src/pydantic_ai_backends/toolsets/descriptions.py
Python
def docstring(self) -> str:
    """A Google-style docstring carrying the argument text.

    Set on the tool function before it is registered, because per-argument
    descriptions reach the JSON schema through the docstring and through
    nothing else — which is why they cannot simply live in `render`. The
    summary is repeated here for a reader of the generated docstring; the
    model reads `render` instead, since an explicit description wins over
    the docstring's own summary.
    """
    lines = [self.summary]
    if self.args:
        lines.append("")
        lines.append("Args:")
        lines.extend(f"    {name}: {text}" for name, text in self.args.items())
    return "\n".join(lines)

Console Tools

The toolset registers these tools. What each one says — its description and the text describing every argument — is not written beside the function: it lives in TOOL_TEXT, keyed by tool name, and is assigned when the tool is registered. Read TOOL_TEXT["grep"].render() to see exactly what the model is handed.

Python
async def ls(ctx, path: str = ".") -> str: ...
async def read_file(ctx, path: str, offset: int = 0, limit: int = 2000) -> str: ...
async def write_file(ctx, path: str, content: str) -> str: ...
async def edit_file(
    ctx, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> str: ...
async def hashline_edit(
    ctx,
    path: str,
    start_line: int,
    start_hash: str,
    new_content: str,
    end_line: int | None = None,
    end_hash: str | None = None,
    insert_after: bool = False,
) -> str: ...
async def glob(ctx, pattern: str, path: str = ".") -> str: ...
async def grep(
    ctx,
    pattern: str,
    path: str | None = None,
    glob_pattern: str | None = None,
    output_mode: Literal["content", "files_with_matches", "count"] = "files_with_matches",
    ignore_hidden: bool = True,
) -> str: ...
async def execute(ctx, command: str, timeout: int | None = 120) -> str: ...
async def run_in_background(ctx, command: str) -> str: ...
async def read_output(ctx, shell_id: str) -> str: ...
async def kill_shell(ctx, shell_id: str) -> str: ...
async def list_shells(ctx) -> str: ...