Skip to content

API Reference

This section documents the main classes and helpers exposed by Pygent. The content is generated from the package docstrings.

Agent

pygent.agent.Agent(runtime: Runtime = Runtime(), model: Model = _default_model(), model_name: str = DEFAULT_MODEL, persona: Persona = (lambda: DEFAULT_PERSONA)(), system_message_builder: Optional[Callable[[Persona, Optional[List[str]]], str]] = None, system_msg: str = '', history: List[Dict[str, Any]] = list(), history_file: Optional[pathlib.Path] = _default_history_file(), disabled_tools: List[str] = list(), log_file: Optional[pathlib.Path] = _default_log_file(), confirm_bash: bool = _default_confirm_bash(), max_non_tool_replies: Optional[int] = None) dataclass

Interactive assistant handling messages and tool execution.

runtime: Runtime = field(default_factory=Runtime) class-attribute instance-attribute

model: Model = field(default_factory=_default_model) class-attribute instance-attribute

model_name: str = DEFAULT_MODEL class-attribute instance-attribute

persona: Persona = field(default_factory=(lambda: DEFAULT_PERSONA)) class-attribute instance-attribute

system_message_builder: Optional[Callable[[Persona, Optional[List[str]]], str]] = None class-attribute instance-attribute

system_msg: str = '' class-attribute instance-attribute

history: List[Dict[str, Any]] = field(default_factory=list) class-attribute instance-attribute

history_file: Optional[pathlib.Path] = field(default_factory=_default_history_file) class-attribute instance-attribute

disabled_tools: List[str] = field(default_factory=list) class-attribute instance-attribute

log_file: Optional[pathlib.Path] = field(default_factory=_default_log_file) class-attribute instance-attribute

confirm_bash: bool = field(default_factory=_default_confirm_bash) class-attribute instance-attribute

max_non_tool_replies: Optional[int] = None class-attribute instance-attribute

__post_init__() -> None

Initialize defaults after dataclass construction.

Source code in pygent/agent.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __post_init__(self) -> None:
    """Initialize defaults after dataclass construction."""
    self._log_fp = None
    if not self.system_msg:
        self.refresh_system_message()
    if self.history_file and isinstance(self.history_file, (str, pathlib.Path)):
        self.history_file = pathlib.Path(self.history_file)
        if self.history_file.is_file():
            try:
                with self.history_file.open("r", encoding="utf-8") as fh:
                    data = json.load(fh)
            except Exception:
                data = []
            self.history = [
                openai_compat.parse_message(m) if isinstance(m, dict) else m
                for m in data
            ]
    if not self.history:
        self.append_history({"role": "system", "content": self.system_msg})
    if self.log_file is None:
        if hasattr(self.runtime, "base_dir"):
            self.log_file = pathlib.Path(getattr(self.runtime, "base_dir")) / "cli.log"
        else:
            self.log_file = pathlib.Path("cli.log")
    if isinstance(self.log_file, (str, pathlib.Path)):
        self.log_file = pathlib.Path(self.log_file)
        os.environ.setdefault("PYGENT_LOG_FILE", str(self.log_file))
        self.log_file.parent.mkdir(parents=True, exist_ok=True)
        try:
            self._log_fp = self.log_file.open("a", encoding="utf-8")
        except Exception:
            self._log_fp = None

append_history(msg: Any) -> None

Source code in pygent/agent.py 
127
128
129
130
131
132
133
134
135
def append_history(self, msg: Any) -> None:
    self.history.append(msg)
    self._save_history()
    if self._log_fp:
        try:
            self._log_fp.write(json.dumps(self._message_dict(msg)) + "\n")
            self._log_fp.flush()
        except Exception:
            pass

refresh_system_message() -> None

Update the system prompt based on the current tool registry.

Source code in pygent/agent.py 
137
138
139
140
141
142
143
144
145
146
def refresh_system_message(self) -> None:
    """Update the system prompt based on the current tool registry."""
    if self.system_message_builder:
        self.system_msg = self.system_message_builder(
            self.persona, self.disabled_tools
        )
    else:
        self.system_msg = build_system_msg(self.persona, self.disabled_tools)
    if self.history and self.history[0].get("role") == "system":
        self.history[0]["content"] = self.system_msg

step(user_msg: str = None, role: str = 'user') -> openai_compat.Message

Execute one round of interaction with the model.

Source code in pygent/agent.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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
def step(self, user_msg: str = None, role: str = "user") -> openai_compat.Message:
    """Execute one round of interaction with the model."""

    self.refresh_system_message()
    if user_msg:
        self.append_history({"role": role, "content": user_msg})

    status_cm = (
        console.status("[bold cyan]Thinking...", spinner="dots")
        if hasattr(console, "status")
        else nullcontext()
    )
    schemas = [
        s
        for s in tools.TOOL_SCHEMAS
        if s["function"]["name"] not in self.disabled_tools
    ]
    with status_cm:
        assistant_raw = self.model.chat(
            self.history, self.model_name, schemas
        )
    assistant_msg = openai_compat.parse_message(assistant_raw)
    self.append_history(assistant_msg)

    if assistant_msg.tool_calls:
        for call in assistant_msg.tool_calls:
            if self.confirm_bash and call.function.name == "bash":
                args = json.loads(call.function.arguments or "{}")
                cmd = args.get("cmd", "")
                console.print(
                    Panel(
                        f"$ {cmd}",
                        title=f"[bold yellow]{self.persona.name} pending bash[/]",
                        border_style="yellow",
                        box=box.HEAVY_HEAD if box else None,
                        title_align="left",
                    )
                )
                prompt = "Run this command?"
                if questionary:
                    ok = questionary.confirm(prompt, default=True).ask()
                else:  # pragma: no cover - fallback for tests
                    ok_input = console.input(f"{prompt} [Y/n]: ").lower()
                    ok = ok_input == "" or ok_input.startswith("y")
                if not ok:
                    output = f"$ {cmd}\n[bold red]Aborted by user.[/]"
                    self.append_history({"role": "tool", "content": output, "tool_call_id": call.id})
                    console.print(
                        Panel(
                            output,
                            title=f"[bold red]{self.persona.name} tool:{call.function.name}[/]",
                            border_style="red",
                            box=box.ROUNDED if box else None,
                            title_align="left",
                        )
                    )
                    continue
            status_cm = (
                console.status(
                    f"[green]Running {call.function.name}...", spinner="line"
                )
                if hasattr(console, "status")
                else nullcontext()
            )
            with status_cm:
                output = tools.execute_tool(call, self.runtime)
            self.append_history(
                {"role": "tool", "content": output, "tool_call_id": call.id}
            )
            if call.function.name not in {"ask_user", "stop"}:
                display_output = output
                if call.function.name == "read_image" and output.startswith("data:image"):
                    try:
                        args = json.loads(call.function.arguments or "{}")
                        path = args.get("path", "<unknown>")
                        self.append_history(
                            {
                                "role": "user",
                                "content": [
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": output
                                        }
                                    }
                                ]
                            }
                        )
                    except Exception:
                        path = "<unknown>"
                    display_output = f"returned data URL for {path}"
                console.print(
                    Panel(
                        display_output,
                        title=f"[bold bright_blue]{self.persona.name} tool:{call.function.name}[/]",
                        border_style="bright_blue",
                        box=box.ROUNDED if box else None,
                        title_align="left",
                    )
                )
    else:
        markdown_response = Markdown(assistant_msg.content or "") # Ensure content is not None
        console.print(
            Panel(
                markdown_response,
                title=f"[bold green]{self.persona.name} replied[/]",
                title_align="left",
                border_style="green",
                box=box.ROUNDED if box else None,
            )
        )
    return assistant_msg

run_until_stop(user_msg: str, max_steps: int = 20, step_timeout: Optional[float] = None, max_time: Optional[float] = None) -> Optional[openai_compat.Message]

Run steps until stop is called or limits are reached.

The agent also stops if it replies without using a tool more than max_non_tool_replies times consecutively.

Source code in pygent/agent.py 
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
def run_until_stop(
    self,
    user_msg: str,
    max_steps: int = 20,
    step_timeout: Optional[float] = None,
    max_time: Optional[float] = None,
) -> Optional[openai_compat.Message]:
    """Run steps until ``stop`` is called or limits are reached.

    The agent also stops if it replies without using a tool more than
    ``max_non_tool_replies`` times consecutively.
    """

    if step_timeout is None:
        env = os.getenv("PYGENT_STEP_TIMEOUT")
        step_timeout = float(env) if env else None
    if max_time is None:
        env = os.getenv("PYGENT_TASK_TIMEOUT")
        max_time = float(env) if env else None

    msg = user_msg
    start = time.monotonic()
    self._timed_out = False
    last_msg = None
    non_tool_replies = 0
    for _ in range(max_steps):
        if max_time is not None and time.monotonic() - start > max_time:
            self.append_history(
                {"role": "system", "content": f"[timeout after {max_time}s]"}
            )
            self._timed_out = True
            break
        step_start = time.monotonic()
        if msg==user_msg:
          role = 'user'
        else:
          role = 'system'
        assistant_msg = self.step(msg, role=role)
        last_msg = assistant_msg
        if (
            step_timeout is not None
            and time.monotonic() - step_start > step_timeout
        ):
            self.append_history(
                {"role": "system", "content": f"[timeout after {step_timeout}s]"}
            )
            self._timed_out = True
            break
        calls = assistant_msg.tool_calls or []
        if any(c.function.name in ("stop", "ask_user") for c in calls):
            break
        if calls:
            non_tool_replies = 0
        else:
            non_tool_replies += 1
            if (
                self.max_non_tool_replies is not None
                and non_tool_replies >= self.max_non_tool_replies
            ):
                self.append_history(
                    {
                        "role": "system",
                        "content": "[stopped after too many non-tool replies]",
                    }
                )
                break
        msg = None
    return last_msg

close() -> None

Close any open resources.

Source code in pygent/agent.py 
330
331
332
333
334
335
336
def close(self) -> None:
    """Close any open resources."""
    if self._log_fp:
        try:
            self._log_fp.close()
        finally:
            self._log_fp = None

Runtime

pygent.runtime.Runtime(image: Optional[str] = None, use_docker: Optional[bool] = None, initial_files: Optional[list[str]] = None, workspace: Optional[Union[str, Path]] = None, banned_commands: Optional[list[str]] = None, banned_apps: Optional[list[str]] = None)

Executes commands in a Docker container or locally if Docker is unavailable.

If workspace or the environment variable PYGENT_WORKSPACE is set, the given directory is used as the base workspace and kept across sessions.

Create a new execution runtime.

banned_commands and banned_apps can be used to restrict what can be run. Environment variables PYGENT_BANNED_COMMANDS and PYGENT_BANNED_APPS extend these lists using os.pathsep as the delimiter.

Source code in pygent/runtime.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(
    self,
    image: Optional[str] = None,
    use_docker: Optional[bool] = None,
    initial_files: Optional[list[str]] = None,
    workspace: Optional[Union[str, Path]] = None,
    banned_commands: Optional[list[str]] = None,
    banned_apps: Optional[list[str]] = None,
) -> None:
    """Create a new execution runtime.

    ``banned_commands`` and ``banned_apps`` can be used to restrict what
    can be run. Environment variables ``PYGENT_BANNED_COMMANDS`` and
    ``PYGENT_BANNED_APPS`` extend these lists using ``os.pathsep`` as the
    delimiter.
    """
    env_ws = os.getenv("PYGENT_WORKSPACE")
    if workspace is None and env_ws:
        workspace = env_ws
    if workspace is None:
        self.base_dir = Path.cwd() / f"agent_{uuid.uuid4().hex[:8]}"
        self._persistent = False
    else:
        self.base_dir = Path(workspace).expanduser()
        self._persistent = True
    self.base_dir.mkdir(parents=True, exist_ok=True)
    if initial_files is None:
        env_files = os.getenv("PYGENT_INIT_FILES")
        if env_files:
            initial_files = [f.strip() for f in env_files.split(os.pathsep) if f.strip()]
    self._initial_files = initial_files or []
    self.image = image or os.getenv("PYGENT_IMAGE", "python:3.12-slim")
    env_opt = os.getenv("PYGENT_USE_DOCKER")
    if use_docker is None:
        use_docker = (env_opt != "0") if env_opt is not None else True
    self._use_docker = bool(docker) and use_docker
    if self._use_docker:
        try:
            self.client = docker.from_env()
            self.container = self.client.containers.run(
                self.image,
                name=f"pygent-{uuid.uuid4().hex[:8]}",
                command="sleep infinity",
                volumes={str(self.base_dir): {"bind": "/workspace", "mode": "rw"}},
                working_dir="/workspace",
                detach=True,
                tty=True,
                network_disabled=True,
                mem_limit="512m",
                pids_limit=256,
            )
        except Exception:
            self._use_docker = False
    if not self._use_docker:
        self.client = None
        self.container = None

    # populate workspace with initial files
    for fp in self._initial_files:
        src = Path(fp).expanduser()
        dest = self.base_dir / src.name
        if src.is_dir():
            shutil.copytree(src, dest, dirs_exist_ok=True)
        elif src.exists():
            dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy(src, dest)

    env_banned_cmds = os.getenv("PYGENT_BANNED_COMMANDS")
    env_banned_apps = os.getenv("PYGENT_BANNED_APPS")
    self.banned_commands = set(banned_commands or [])
    if env_banned_cmds:
        self.banned_commands.update(c.strip() for c in env_banned_cmds.split(os.pathsep) if c.strip())
    self.banned_apps = set(banned_apps or [])
    if env_banned_apps:
        self.banned_apps.update(a.strip() for a in env_banned_apps.split(os.pathsep) if a.strip())

base_dir = Path.cwd() / f'agent_{uuid.uuid4().hex[:8]}' instance-attribute

image = image or os.getenv('PYGENT_IMAGE', 'python:3.12-slim') instance-attribute

client = docker.from_env() instance-attribute

container = self.client.containers.run(self.image, name=f'pygent-{uuid.uuid4().hex[:8]}', command='sleep infinity', volumes={str(self.base_dir): {'bind': '/workspace', 'mode': 'rw'}}, working_dir='/workspace', detach=True, tty=True, network_disabled=True, mem_limit='512m', pids_limit=256) instance-attribute

banned_commands = set(banned_commands or []) instance-attribute

banned_apps = set(banned_apps or []) instance-attribute

use_docker: bool property

Return True if commands run inside a Docker container.

bash(cmd: str, timeout: int = 600, stream: Optional[Callable[[str], None]] = None) -> str

Run cmd in the sandbox and return its captured output.

If stream is provided, lines are forwarded to it as they are produced. The returned value still contains the full output prefixed with $ cmd just like before.

Source code in pygent/runtime.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def bash(
    self,
    cmd: str,
    timeout: int = 600,
    stream: Optional[Callable[[str], None]] = None,
) -> str:
    """Run ``cmd`` in the sandbox and return its captured output.

    If ``stream`` is provided, lines are forwarded to it as they are
    produced. The returned value still contains the full output prefixed
    with ``$ cmd`` just like before.
    """
    tokens = cmd.split()
    if tokens:
        from pathlib import Path

        if Path(tokens[0]).name in self.banned_commands:
            return f"$ {cmd}\n[error] command '{Path(tokens[0]).name}' disabled"
        for t in tokens:
            if Path(t).name in self.banned_apps:
                return f"$ {cmd}\n[error] application '{Path(t).name}' disabled"

    prefix = f"$ {cmd}\n"
    output_parts = [prefix]
    if stream:
        stream(prefix)

    if self._use_docker and self.container is not None:
        try:
            if stream is not None:
                res = self.container.exec_run(
                    cmd,
                    workdir="/workspace",
                    stream=True,
                    tty=False,
                    stdin=False,
                )
                for chunk in res.output:
                    text = chunk.decode()
                    output_parts.append(text)
                    stream(text)
                return "".join(output_parts)
            res = self.container.exec_run(
                cmd,
                workdir="/workspace",
                demux=True,
                tty=False,
                stdin=False,
                timeout=timeout,
            )
            stdout, stderr = (
                res.output if isinstance(res.output, tuple) else (res.output, b"")
            )
            text = (stdout or b"").decode() + (stderr or b"").decode()
            output_parts.append(text)
            return "".join(output_parts)
        except Exception as exc:
            err = f"[error] {exc}"
            output_parts.append(err)
            if stream:
                stream(err + "\n")
            return "".join(output_parts)

    try:
        proc = subprocess.Popen(
            cmd,
            shell=True,
            cwd=self.base_dir,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            stdin=subprocess.DEVNULL,
        )

        lines: list[str] = []

        def _reader() -> None:
            assert proc.stdout is not None
            for line in proc.stdout:
                lines.append(line)
                if stream:
                    stream(line)

        t = threading.Thread(target=_reader)
        t.start()
        try:
            proc.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            proc.kill()
            t.join()
            lines.append(f"[timeout after {timeout}s]\n")
            if stream:
                stream(f"[timeout after {timeout}s]\n")
            output_parts.extend(lines)
            return "".join(output_parts)
        except Exception as exc:
            proc.kill()
            t.join()
            lines.append(f"[error] {exc}\n")
            if stream:
                stream(f"[error] {exc}\n")
            output_parts.extend(lines)
            return "".join(output_parts)

        t.join()
        output_parts.extend(lines)
        return "".join(output_parts)
    except Exception as exc:
        err = f"[error] {exc}"
        output_parts.append(err)
        if stream:
            stream(err + "\n")
        return "".join(output_parts)

write_file(path: Union[str, Path], content: str) -> str

Source code in pygent/runtime.py 
223
224
225
226
227
def write_file(self, path: Union[str, Path], content: str) -> str:
    p = self.base_dir / path
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content, encoding="utf-8")
    return f"Wrote {p.relative_to(self.base_dir)}"

read_file(path: Union[str, Path], binary: bool = False) -> str

Return the contents of a file relative to the workspace.

Source code in pygent/runtime.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def read_file(self, path: Union[str, Path], binary: bool = False) -> str:
    """Return the contents of a file relative to the workspace."""

    p = self.base_dir / path
    if not p.exists():
        return f"file {p.relative_to(self.base_dir)} not found"
    data = p.read_bytes()
    if binary:
        import base64

        return base64.b64encode(data).decode()
    try:
        return data.decode()
    except UnicodeDecodeError:
        import base64

        return base64.b64encode(data).decode()

upload_file(src: Union[str, Path], dest: Optional[Union[str, Path]] = None) -> str

Copy a local file or directory into the workspace.

Source code in pygent/runtime.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
def upload_file(self, src: Union[str, Path], dest: Optional[Union[str, Path]] = None) -> str:
    """Copy a local file or directory into the workspace."""

    src_path = Path(src).expanduser()
    if not src_path.exists():
        return f"file {src} not found"
    target = self.base_dir / (Path(dest) if dest else src_path.name)
    if src_path.is_dir():
        shutil.copytree(src_path, target, dirs_exist_ok=True)
    else:
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy(src_path, target)
    return f"Uploaded {target.relative_to(self.base_dir)}"

export_file(path: Union[str, Path], dest: Union[str, Path]) -> str

Copy a file or directory from the workspace to a local path.

Source code in pygent/runtime.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
def export_file(self, path: Union[str, Path], dest: Union[str, Path]) -> str:
    """Copy a file or directory from the workspace to a local path."""

    src = self.base_dir / path
    if not src.exists():
        return f"file {path} not found"
    dest_path = Path(dest).expanduser()
    if src.is_dir():
        shutil.copytree(src, dest_path, dirs_exist_ok=True)
    else:
        dest_path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy(src, dest_path)
    return f"Exported {src.relative_to(self.base_dir)}"

cleanup() -> None

Source code in pygent/runtime.py 
275
276
277
278
279
280
281
282
def cleanup(self) -> None:
    if self._use_docker and self.container is not None:
        try:
            self.container.kill()
        finally:
            self.container.remove(force=True)
    if not self._persistent:
        shutil.rmtree(self.base_dir, ignore_errors=True)

TaskManager

pygent.task_manager.TaskManager(agent_factory: Optional[Callable[..., 'Agent']] = None, max_tasks: Optional[int] = None, personas: Optional[list[Persona]] = None)

Launch agents asynchronously and track their progress.

Source code in pygent/task_manager.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    agent_factory: Optional[Callable[..., "Agent"]] = None,
    max_tasks: Optional[int] = None,
    personas: Optional[list[Persona]] = None,
) -> None:
    from .agent import Agent  # local import to avoid circular dependency

    env_max = os.getenv("PYGENT_MAX_TASKS")
    self.max_tasks = max_tasks if max_tasks is not None else int(env_max or "3")
    if agent_factory is None:
        self.agent_factory = lambda p=None: Agent(persona=p)
    else:
        self.agent_factory = agent_factory
    env_personas_json = os.getenv("PYGENT_TASK_PERSONAS_JSON")
    if personas is None and env_personas_json:
        try:
            data = json.loads(env_personas_json)
            if isinstance(data, list):
                personas = [
                    Persona(p.get("name", ""), p.get("description", ""))
                    for p in data
                    if isinstance(p, dict)
                ]
        except Exception:
            personas = None
    env_personas = os.getenv("PYGENT_TASK_PERSONAS")
    if personas is None and env_personas:
        personas = [
            Persona(p.strip(), "")
            for p in env_personas.split(os.pathsep)
            if p.strip()
        ]
    if personas is None:
        personas = [
            Persona(
                os.getenv("PYGENT_PERSONA_NAME", "Pygent"),
                os.getenv("PYGENT_PERSONA", "a sandboxed coding assistant."),
            )
        ]
    self.personas = personas
    self._persona_idx = 0
    self.tasks: Dict[str, Task] = {}
    self._lock = threading.Lock()

max_tasks = max_tasks if max_tasks is not None else int(env_max or '3') instance-attribute

agent_factory = lambda p=None: Agent(persona=p) instance-attribute

personas = personas instance-attribute

tasks: Dict[str, Task] = {} instance-attribute

start_task(prompt: str, parent_rt: Runtime, files: Optional[list[str]] = None, parent_depth: int = 0, step_timeout: Optional[float] = None, task_timeout: Optional[float] = None, persona: Union[Persona, str, None] = None) -> str

Create a new agent and run prompt asynchronously.

persona overrides the default rotation used for delegated tasks.

Source code in pygent/task_manager.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def start_task(
    self,
    prompt: str,
    parent_rt: Runtime,
    files: Optional[list[str]] = None,
    parent_depth: int = 0,
    step_timeout: Optional[float] = None,
    task_timeout: Optional[float] = None,
    persona: Union[Persona, str, None] = None,
) -> str:
    """Create a new agent and run ``prompt`` asynchronously.

    ``persona`` overrides the default rotation used for delegated tasks.
    """

    if parent_depth >= 1:
        raise RuntimeError("nested delegation is not allowed")

    with self._lock:
        active = sum(t.status == "running" for t in self.tasks.values())
        if active >= self.max_tasks:
            raise RuntimeError(f"max {self.max_tasks} tasks reached")

    if step_timeout is None:
        env = os.getenv("PYGENT_STEP_TIMEOUT")
        step_timeout = float(env) if env else 60 * 5  # default 5 minutes
    if task_timeout is None:
        env = os.getenv("PYGENT_TASK_TIMEOUT")
        task_timeout = float(env) if env else 60 * 20  # default 20 minutes

    if persona is None:
        persona = self.personas[self._persona_idx % len(self.personas)]
        self._persona_idx += 1
    elif isinstance(persona, str):
        match = next((p for p in self.personas if p.name == persona), None)
        persona = match or Persona(persona, "")
    try:
        agent = self.agent_factory(persona)
    except TypeError:
        agent = self.agent_factory()

    from .runtime import Runtime
    if getattr(agent, "runtime", None) is not None:
        try:
            agent.runtime.cleanup()
        except Exception:
            pass
    task_dir = parent_rt.base_dir / f"task_{uuid.uuid4().hex[:8]}"
    agent.runtime = Runtime(use_docker=parent_rt.use_docker, workspace=task_dir)
    setattr(agent, "persona", persona)
    if not getattr(agent, "system_msg", None):
        from .system_message import build_system_msg  # lazy import

        agent.system_msg = build_system_msg(persona)
    setattr(agent.runtime, "task_depth", parent_depth + 1)
    if files:
        for fp in files:
            src = parent_rt.base_dir / fp
            dest = agent.runtime.base_dir / fp
            if src.is_dir():
                shutil.copytree(src, dest, dirs_exist_ok=True)
            elif src.exists():
                dest.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy(src, dest)
    task_id = uuid.uuid4().hex[:8]
    task = Task(id=task_id, agent=agent, thread=None)  # type: ignore[arg-type]

    def run() -> None:
        try:
            agent.run_until_stop(
                prompt,
                step_timeout=step_timeout,
                max_time=task_timeout,
            )
            if getattr(agent, "_timed_out", False):
                task.status = f"timeout after {task_timeout}s"
            else:
                task.status = "finished"
        except Exception as exc:  # pragma: no cover - error propagation
            task.status = f"error: {exc}"

    t = threading.Thread(target=run, daemon=True)
    task.thread = t
    with self._lock:
        self.tasks[task_id] = task
    t.start()
    return task_id

status(task_id: str) -> str

Source code in pygent/task_manager.py 
167
168
169
170
171
172
def status(self, task_id: str) -> str:
    with self._lock:
        task = self.tasks.get(task_id)
    if not task:
        return f"Task {task_id} not found"
    return task.status

collect_file(rt: Runtime, task_id: str, path: str, dest: Optional[str] = None) -> str

Copy a file or directory from a task workspace into rt.

Source code in pygent/task_manager.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def collect_file(
    self, rt: Runtime, task_id: str, path: str, dest: Optional[str] = None
) -> str:
    """Copy a file or directory from a task workspace into ``rt``."""

    with self._lock:
        task = self.tasks.get(task_id)
    if not task:
        return f"Task {task_id} not found"
    src = task.agent.runtime.base_dir / path
    if not src.exists():
        return f"file {path} not found"
    dest_path = rt.base_dir / (dest or path)
    if src.is_dir():
        shutil.copytree(src, dest_path, dirs_exist_ok=True)
    else:
        dest_path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy(src, dest_path)
    return f"Retrieved {dest_path.relative_to(rt.base_dir)}"

Tools

pygent.tools

Tool registry and helper utilities.

TOOLS: Dict[str, Callable[..., str]] = {} module-attribute

TOOL_SCHEMAS: List[Dict[str, Any]] = [] module-attribute

BUILTIN_TOOLS = TOOLS.copy() module-attribute

BUILTIN_TOOL_SCHEMAS = deepcopy(TOOL_SCHEMAS) module-attribute

register_tool(name: str, description: str, parameters: Dict[str, Any], func: Callable[..., str]) -> None

Register a new callable tool.

Source code in pygent/tools.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def register_tool(
    name: str, description: str, parameters: Dict[str, Any], func: Callable[..., str]
) -> None:
    """Register a new callable tool."""
    if name in TOOLS:
        raise ValueError(f"tool {name} already registered")
    TOOLS[name] = func
    TOOL_SCHEMAS.append(
        {
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": parameters,
            },
        }
    )

tool(name: str, description: str, parameters: Dict[str, Any])

Decorator for registering a tool.

Source code in pygent/tools.py 
48
49
50
51
52
53
54
55
def tool(name: str, description: str, parameters: Dict[str, Any]):
    """Decorator for registering a tool."""

    def decorator(func: Callable[..., str]) -> Callable[..., str]:
        register_tool(name, description, parameters, func)
        return func

    return decorator

execute_tool(call: Any, rt: Runtime) -> str

Dispatch a tool call.

Any exception raised by the tool is caught and returned as an error string so callers don't crash the CLI.

Source code in pygent/tools.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def execute_tool(call: Any, rt: Runtime) -> str:  # pragma: no cover
    """Dispatch a tool call.

    Any exception raised by the tool is caught and returned as an error
    string so callers don't crash the CLI.
    """

    name = call.function.name
    try:
        args: Dict[str, Any] = json.loads(call.function.arguments or "{}")
    except Exception as exc:  # pragma: no cover - defensive
        return f"[error] invalid arguments for {name}: {exc}"

    func = TOOLS.get(name)
    if func is None:
        return f"⚠️ unknown tool {name}"

    try:
        return func(rt, **args)
    except Exception as exc:  # pragma: no cover - tool errors
        return f"[error] {exc}"

clear_tools() -> None

Remove all registered tools globally.

Source code in pygent/tools.py 
252
253
254
255
def clear_tools() -> None:
    """Remove all registered tools globally."""
    TOOLS.clear()
    TOOL_SCHEMAS.clear()

reset_tools() -> None

Restore the default built-in tools.

Source code in pygent/tools.py 
258
259
260
261
262
def reset_tools() -> None:
    """Restore the default built-in tools."""
    clear_tools()
    TOOLS.update(BUILTIN_TOOLS)
    TOOL_SCHEMAS.extend(deepcopy(BUILTIN_TOOL_SCHEMAS))

remove_tool(name: str) -> None

Unregister a specific tool.

Source code in pygent/tools.py 
265
266
267
268
269
270
271
272
273
274
def remove_tool(name: str) -> None:
    """Unregister a specific tool."""
    if name not in TOOLS:
        raise ValueError(f"tool {name} not registered")
    del TOOLS[name]
    for i, schema in enumerate(TOOL_SCHEMAS):
        func = schema.get("function", {})
        if func.get("name") == name:
            TOOL_SCHEMAS.pop(i)
            break

Models

pygent.models

CUSTOM_MODEL: Optional[Model] = None module-attribute

Model

Bases: Protocol

Protocol for chat models used by :class:~pygent.agent.Agent.

chat(messages: List[Dict[str, Any]], model: str, tools: Any) -> Message

Return the assistant message for the given prompt.

Source code in pygent/models.py 
20
21
22
def chat(self, messages: List[Dict[str, Any]], model: str, tools: Any) -> Message:
    """Return the assistant message for the given prompt."""
    ...

OpenAIModel

Default model using the OpenAI-compatible API.

chat(messages: List[Dict[str, Any]], model: str, tools: Any) -> Message

Source code in pygent/models.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def chat(self, messages: List[Dict[str, Any]], model: str, tools: Any) -> Message:
    try:
        serialized = [
            asdict(m) if is_dataclass(m) else m
            for m in messages
        ]
        resp = openai.chat.completions.create(
            model=model,
            messages=serialized,
            tools=tools,
            tool_choice="auto",
        )
        return resp.choices[0].message
    except Exception as exc:
        raise APIError(str(exc)) from exc

set_custom_model(model: Optional[Model]) -> None

Set a global custom model used by :class:~pygent.agent.Agent.

Source code in pygent/models.py 
49
50
51
52
53
def set_custom_model(model: Optional[Model]) -> None:
    """Set a global custom model used by :class:`~pygent.agent.Agent`."""

    global CUSTOM_MODEL
    CUSTOM_MODEL = model

Config

pygent.config

Utilities for loading configuration files.

DEFAULT_CONFIG_FILES = [Path('pygent.toml'), Path.home() / '.pygent.toml'] module-attribute

load_snapshot(path: Union[str, os.PathLike[str]]) -> Path

Load environment variables and history from a snapshot directory.

Source code in pygent/config.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def load_snapshot(path: Union[str, os.PathLike[str]]) -> Path:
    """Load environment variables and history from a snapshot directory."""

    dest = Path(path)
    env_file = dest / "env.json"
    if env_file.is_file():
        try:
            data = json.loads(env_file.read_text())
        except Exception:
            data = {}
        for k, v in data.items():
            os.environ.setdefault(k, str(v))
    ws = dest / "workspace"
    os.environ["PYGENT_WORKSPACE"] = str(ws)
    hist = dest / "history.json"
    if hist.is_file():
        os.environ["PYGENT_HISTORY_FILE"] = str(hist)
    log = dest / "cli.log"
    if log.is_file():
        os.environ["PYGENT_LOG_FILE"] = str(log)
    return ws

run_py_config(path: Union[str, os.PathLike[str]] = 'config.py') -> None

Execute a Python configuration file if it exists.

Source code in pygent/config.py 
44
45
46
47
48
49
50
51
52
def run_py_config(path: Union[str, os.PathLike[str]] = "config.py") -> None:
    """Execute a Python configuration file if it exists."""
    p = Path(path)
    if not p.is_file():
        return
    spec = importlib.util.spec_from_file_location("pygent_config", p)
    if spec and spec.loader:
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

load_config(path: Optional[Union[str, os.PathLike[str]]] = None) -> Dict[str, Any]

Load configuration from a TOML file and set environment variables.

Environment variables already set take precedence over file values. Returns the configuration dictionary.

Source code in pygent/config.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def load_config(path: Optional[Union[str, os.PathLike[str]]] = None) -> Dict[str, Any]:
    """Load configuration from a TOML file and set environment variables.

    Environment variables already set take precedence over file values.
    Returns the configuration dictionary.
    """
    config: Dict[str, Any] = {}
    paths = [Path(path)] if path else DEFAULT_CONFIG_FILES
    for p in paths:
        if p.is_file():
            with p.open("rb") as fh:
                try:
                    data = tomllib.load(fh)
                except Exception:
                    continue
            config.update(data)
    # update environment without overwriting existing values
    if "persona" in config and "PYGENT_PERSONA" not in os.environ:
        os.environ["PYGENT_PERSONA"] = str(config["persona"])
    if "persona_name" in config and "PYGENT_PERSONA_NAME" not in os.environ:
        os.environ["PYGENT_PERSONA_NAME"] = str(config["persona_name"])
    if "task_personas" in config:
        personas = config["task_personas"]
        if isinstance(personas, list) and personas and isinstance(personas[0], Mapping):
            if "PYGENT_TASK_PERSONAS_JSON" not in os.environ:
                os.environ["PYGENT_TASK_PERSONAS_JSON"] = json.dumps(personas)
            if "PYGENT_TASK_PERSONAS" not in os.environ:
                os.environ["PYGENT_TASK_PERSONAS"] = os.pathsep.join(
                    str(p.get("name", "")) for p in personas
                )
        elif "PYGENT_TASK_PERSONAS" not in os.environ:
            if isinstance(personas, list):
                os.environ["PYGENT_TASK_PERSONAS"] = os.pathsep.join(
                    str(p) for p in personas
                )
            else:
                os.environ["PYGENT_TASK_PERSONAS"] = str(personas)
    if "initial_files" in config and "PYGENT_INIT_FILES" not in os.environ:
        if isinstance(config["initial_files"], list):
            os.environ["PYGENT_INIT_FILES"] = os.pathsep.join(
                str(p) for p in config["initial_files"]
            )
        else:
            os.environ["PYGENT_INIT_FILES"] = str(config["initial_files"])
    if "banned_commands" in config and "PYGENT_BANNED_COMMANDS" not in os.environ:
        banned = config["banned_commands"]
        if isinstance(banned, list):
            os.environ["PYGENT_BANNED_COMMANDS"] = os.pathsep.join(str(c) for c in banned)
        else:
            os.environ["PYGENT_BANNED_COMMANDS"] = str(banned)
    if "banned_apps" in config and "PYGENT_BANNED_APPS" not in os.environ:
        apps = config["banned_apps"]
        if isinstance(apps, list):
            os.environ["PYGENT_BANNED_APPS"] = os.pathsep.join(str(a) for a in apps)
        else:
            os.environ["PYGENT_BANNED_APPS"] = str(apps)
    return config

Errors

pygent.errors

PygentError

Bases: Exception

Base error for the Pygent package.

APIError

Bases: PygentError

Raised when the OpenAI API call fails.

OpenAI Compatibility (openai_compat)

pygent.openai_compat

Lightweight client compatible with the OpenAI HTTP API.

OPENAI_BASE_URL = os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1') module-attribute

OPENAI_API_KEY = os.getenv('OPENAI_API_KEY', '') module-attribute

chat = _Chat() module-attribute

ToolCallFunction(name: str, arguments: str) dataclass

name: str instance-attribute

arguments: str instance-attribute

ToolCall(id: str, type: str, function: ToolCallFunction) dataclass

id: str instance-attribute

type: str instance-attribute

function: ToolCallFunction instance-attribute

Message(role: str, content: Optional[str] = None, tool_calls: Optional[List[ToolCall]] = None) dataclass

role: str instance-attribute

content: Optional[str] = None class-attribute instance-attribute

tool_calls: Optional[List[ToolCall]] = None class-attribute instance-attribute

Choice(message: Message) dataclass

message: Message instance-attribute

ChatCompletion(choices: List[Choice]) dataclass

choices: List[Choice] instance-attribute

parse_message(raw: Any) -> Message

Return a :class:Message from raw data.

Accepts dictionaries and objects from the official OpenAI client.

Source code in pygent/openai_compat.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def parse_message(raw: Any) -> Message:
    """Return a :class:`Message` from ``raw`` data.

    Accepts dictionaries and objects from the official OpenAI client.
    """
    if isinstance(raw, Message):
        return raw
    if isinstance(raw, dict):
        tool_calls = []
        for tc in raw.get("tool_calls", []) or []:
            func_data = tc.get("function", {})
            func = ToolCallFunction(
                name=func_data.get("name", ""),
                arguments=func_data.get("arguments", ""),
            )
            tool_calls.append(
                ToolCall(
                    id=tc.get("id", ""),
                    type=tc.get("type", ""),
                    function=func,
                )
            )
        return Message(
            role=raw.get("role", ""),
            content=raw.get("content"),
            tool_calls=tool_calls or None,
        )
    if hasattr(raw, "model_dump"):
        return parse_message(raw.model_dump())
    if hasattr(raw, "to_dict"):
        return parse_message(raw.to_dict())
    raise TypeError(f"Unsupported message type: {type(raw)!r}")

Commands

pygent.commands

COMMANDS: Dict[str, Command] = {'/cmd': Command(cmd_cmd, description='Run a raw shell command in the sandbox.', usage='/cmd <command>'), '/cp': Command(cmd_cp, description='Copy a file into the workspace.', usage='/cp SRC [DEST]'), '/new': Command(cmd_new, description='Restart the conversation with a fresh history.', usage='/new'), '/help': Command(cmd_help, description='Display available commands.', usage='/help [command]'), '/save': Command(cmd_save, description='Save workspace and environment to DIR for later use.', usage='/save DIR'), '/tools': Command(cmd_tools, description='Enable/disable tools at runtime or list them.', usage='/tools [list|enable NAME|disable NAME]'), '/banned': Command(cmd_banned, description='List or modify banned commands.', usage='/banned [list|add CMD|remove CMD]'), '/confirm-bash': Command(cmd_confirm_bash, description='Toggle confirmation before running bash commands.', usage='/confirm-bash [on|off]')} module-attribute

Command(handler: Callable[[Agent, str], Optional[Agent]], description: str | None = None, usage: str | None = None)

CLI command definition.

Source code in pygent/commands.py 
34
35
36
37
def __init__(self, handler: Callable[[Agent, str], Optional[Agent]], description: str | None = None, usage: str | None = None):
    self.handler = handler
    self.description = description or (handler.__doc__ or "")
    self.usage = usage

handler = handler instance-attribute

description = description or (handler.__doc__ or '') instance-attribute

usage = usage instance-attribute

__call__(agent: Agent, arg: str) -> Optional[Agent]

Source code in pygent/commands.py 
39
40
def __call__(self, agent: Agent, arg: str) -> Optional[Agent]:
    return self.handler(agent, arg)

cmd_cmd(agent: Agent, arg: str) -> None

Run a raw shell command in the sandbox.

Source code in pygent/commands.py 
43
44
45
46
47
48
49
50
def cmd_cmd(agent: Agent, arg: str) -> None:
    """Run a raw shell command in the sandbox."""
    console = Console()

    def _stream(line: str) -> None:
        console.print(line, end="")

    agent.runtime.bash(arg, stream=_stream)

cmd_cp(agent: Agent, arg: str) -> None

Copy a file into the workspace.

Source code in pygent/commands.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def cmd_cp(agent: Agent, arg: str) -> None:
    """Copy a file into the workspace."""
    parts = arg.split()
    console = Console()
    if not parts:
        console.print("Usage: /cp SRC [DEST]", style="bold red")
        return
    src = parts[0]
    dest = parts[1] if len(parts) > 1 else None
    try:
        msg = agent.runtime.upload_file(src, dest)
        console.print(msg)
    except Exception as e:
        console.print(f"Error: {e}", style="bold red")

cmd_new(agent: Agent, arg: str) -> Agent

Restart the conversation with a fresh history.

Source code in pygent/commands.py 
69
70
71
72
73
74
75
76
77
def cmd_new(agent: Agent, arg: str) -> Agent:
    """Restart the conversation with a fresh history."""
    persistent = agent.runtime._persistent
    use_docker = agent.runtime.use_docker
    workspace = agent.runtime.base_dir if persistent else None
    agent.runtime.cleanup()
    console = Console()
    console.print("Starting a new session.", style="green")
    return Agent(runtime=Runtime(use_docker=use_docker, workspace=workspace))

cmd_help(agent: Agent, arg: str) -> None

Display available commands.

Source code in pygent/commands.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def cmd_help(agent: Agent, arg: str) -> None:
    """Display available commands."""
    console = Console()

    if arg:
        command_name = arg if arg.startswith('/') else f'/{arg}'
        cmd = COMMANDS.get(command_name)
        if not cmd:
            console.print(f"No help available for {arg}", style="bold red")
            return
        if Table and Text:
            table = Table(title=f"Help: {command_name}", show_header=False, box=None, padding=(0, 2))
            table.add_row(Text("Description:", style="bold cyan"), cmd.description)
            if cmd.usage:
                table.add_row(Text("Usage:", style="bold cyan"), cmd.usage)
            else:
                table.add_row(Text("Usage:", style="bold cyan"), command_name)
            console.print(table)
        else:  # plain fallback
            print(f"{command_name} - {cmd.description}")
            print(f"Usage: {cmd.usage or command_name}")
        return

    if Table and Text:
        table = Table(title="Available Commands", title_style="bold magenta", show_header=True, header_style="bold cyan")
        table.add_column("Command", style="dim", width=15)
        table.add_column("Description")
        table.add_column("Usage", width=30)

        for name, command in sorted(COMMANDS.items()):
            usage = command.usage or name
            table.add_row(name, command.description, usage)
        table.add_row("/exit", "Quit the session.", "/exit")

        console.print(table)
    else:
        print("Available Commands:")
        for name, command in sorted(COMMANDS.items()):
            usage = command.usage or name
            print(f"{name} - {command.description} ({usage})")
        print("/exit - quit the session (/exit)")

cmd_save(agent: Agent, arg: str) -> None

Save workspace and environment to DIR for later use.

Source code in pygent/commands.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def cmd_save(agent: Agent, arg: str) -> None:
    """Save workspace and environment to ``DIR`` for later use."""
    if not arg:
        print("usage: /save DIR")
        return
    dest = Path(arg).expanduser()
    dest.mkdir(parents=True, exist_ok=True)
    agent.runtime.export_file(".", dest / "workspace")
    if agent.history_file and agent.history_file.exists():
        shutil.copy(agent.history_file, dest / "history.json")
    env = {k: v for k, v in os.environ.items() if k.startswith(("PYGENT_", "OPENAI_"))}
    (dest / "env.json").write_text(json.dumps(env, indent=2), encoding="utf-8")
    if agent.log_file and Path(agent.log_file).exists():
        shutil.copy(agent.log_file, dest / "cli.log")
    print(f"Saved environment to {dest}")

cmd_tools(agent: Agent, arg: str) -> None

Enable/disable tools at runtime or list them.

Source code in pygent/commands.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def cmd_tools(agent: Agent, arg: str) -> None:
    """Enable/disable tools at runtime or list them."""
    parts = arg.split()
    if not parts or parts[0] == "list":
        for name in sorted(tools.TOOLS):
            suffix = " (disabled)" if name in agent.disabled_tools else ""
            print(f"{name}{suffix}")
        return
    if len(parts) != 2 or parts[0] not in {"enable", "disable"}:
        print("usage: /tools [list|enable NAME|disable NAME]")
        return
    action, name = parts
    if action == "enable":
        if name in agent.disabled_tools:
            agent.disabled_tools.remove(name)
            agent.refresh_system_message()
            print(f"Enabled {name}")
        else:
            print(f"{name} already enabled")
    else:
        if name not in agent.disabled_tools:
            agent.disabled_tools.append(name)
            agent.refresh_system_message()
            print(f"Disabled {name}")
        else:
            print(f"{name} already disabled")

cmd_banned(agent: Agent, arg: str) -> None

List or modify banned commands.

Source code in pygent/commands.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def cmd_banned(agent: Agent, arg: str) -> None:
    """List or modify banned commands."""
    parts = arg.split()
    if not parts or parts[0] == "list":
        for name in sorted(agent.runtime.banned_commands):
            print(name)
        return
    if len(parts) != 2 or parts[0] not in {"add", "remove"}:
        print("usage: /banned [list|add CMD|remove CMD]")
        return
    action, name = parts
    if action == "add":
        agent.runtime.banned_commands.add(name)
        print(f"Added {name}")
    else:
        if name in agent.runtime.banned_commands:
            agent.runtime.banned_commands.remove(name)
            print(f"Removed {name}")
        else:
            print(f"{name} not banned")

cmd_confirm_bash(agent: Agent, arg: str) -> None

Show or toggle confirmation for bash commands.

Source code in pygent/commands.py 
190
191
192
193
194
195
196
197
198
199
200
201
def cmd_confirm_bash(agent: Agent, arg: str) -> None:
    """Show or toggle confirmation for bash commands."""
    arg = arg.strip().lower()
    if not arg:
        status = "on" if agent.confirm_bash else "off"
        print(status)
        return
    if arg not in {"on", "off"}:
        print("usage: /confirm-bash [on|off]")
        return
    agent.confirm_bash = arg == "on"
    print("confirmation " + ("enabled" if agent.confirm_bash else "disabled"))

register_command(name: str, handler: Callable[[Agent, str], Optional[Agent]], description: str | None = None, usage: str | None = None) -> None

Register a custom CLI command.

Source code in pygent/commands.py 
204
205
206
207
208
209
210
211
212
213
def register_command(
    name: str,
    handler: Callable[[Agent, str], Optional[Agent]],
    description: str | None = None,
    usage: str | None = None,
) -> None:
    """Register a custom CLI command."""
    if name in COMMANDS:
        raise ValueError(f"command {name} already registered")
    COMMANDS[name] = Command(handler, description, usage)