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_msg: str = lambda: build_system_msg(DEFAULT_PERSONA)(), 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()) 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_msg: str = field(default_factory=lambda: build_system_msg(DEFAULT_PERSONA)) 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

__post_init__() -> None

Initialize defaults after dataclass construction.

Source code in pygent/agent.py

def __post_init__(self) -> None:
    """Initialize defaults after dataclass construction."""
    self._log_fp = None
    if not self.system_msg:
        self.system_msg = build_system_msg(self.persona, self.disabled_tools)
    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

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

def refresh_system_message(self) -> None:
    """Update the system prompt based on the current tool registry."""
    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)

Execute one round of interaction with the model.

Source code in pygent/agent.py

def step(self, user_msg: str):
    """Execute one round of interaction with the model."""

    self.refresh_system_message()
    self.append_history({"role": "user", "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>")
                    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.

Source code in pygent/agent.py

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."""

    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
    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()
        assistant_msg = self.step(msg)
        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
        msg = "ask_user"

    return last_msg

close() -> None

Close any open resources.

Source code in pygent/agent.py

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

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) -> str

Run a command in the container or locally and return the output.

The executed command is always included in the returned string so the caller can display what was run.

Source code in pygent/runtime.py

def bash(self, cmd: str, timeout: int = 600) -> str:
    """Run a command in the container or locally and return the output.

    The executed command is always included in the returned string so the
    caller can display what was run.
    """
    tokens = cmd.split()
    if tokens:
        from pathlib import Path

        if Path(tokens[0]).name in self.banned_commands:
            return f"$ {cmd}\n[error] command '{tokens[0]}' disabled"
        for t in tokens:
            if Path(t).name in self.banned_apps:
                return f"$ {cmd}\n[error] application '{Path(t).name}' disabled"
    if self._use_docker and self.container is not None:
        try:
            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"")
            )
            output = (stdout or b"").decode() + (stderr or b"").decode()
            return f"$ {cmd}\n{output}"
        except Exception as exc:
            return f"$ {cmd}\n[error] {exc}"
    try:
        proc = subprocess.run(
            cmd,
            shell=True,
            cwd=self.base_dir,
            capture_output=True,
            text=True,
            stdin=subprocess.DEVNULL,
            timeout=timeout,
        )
        return f"$ {cmd}\n{proc.stdout + proc.stderr}"
    except subprocess.TimeoutExpired:
        return f"$ {cmd}\n[timeout after {timeout}s]"
    except Exception as exc:
        return f"$ {cmd}\n[error] {exc}"

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

Source code in pygent/runtime.py

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

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

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

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

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

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

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 .agent 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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

def cmd_cmd(agent: Agent, arg: str) -> None:
    """Run a raw shell command in the sandbox."""
    output = agent.runtime.bash(arg)
    console = Console()
    console.print(output)

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

Copy a file into the workspace.

Source code in pygent/commands.py

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

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

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

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

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

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

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

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)