API Reference

This section documents the main classes and helpers exposed by Pygent.

Core API

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

Source code in pygent/agent.py

def __post_init__(self) -> None:
    self._log_fp = None
    if not self.system_msg:
        self.refresh_system_message()
    self._restore_history_file()
    if not self.history:
        self.append_history({"role": "system", "content": self.system_msg})
    self._init_log_file()

append_history(msg: Any) -> None

Source code in pygent/agent.py

def append_history(self, msg: Any) -> None:
    self.history.append(deepcopy(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

Source code in pygent/agent.py

def refresh_system_message(self) -> None:
    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

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

    with _status("[bold cyan]Thinking...", spinner="dots"):
        assistant_raw = self.model.chat(self.history, self.model_name, self._active_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:
            self._execute_tool_call(call)
    else:
        _safe_print(
            Panel(
                Markdown(assistant_msg.content or ""),
                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: Optional[str] = user_msg
    start = time.monotonic()
    self._timed_out = False
    last_msg = None
    non_tool_replies = 0

    for idx 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()
        role = "user" if idx == 0 else "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

Source code in pygent/agent.py

def close(self) -> None:
    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 Docker, or locally when explicitly requested.

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 = self.base_dir.resolve()
    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 = use_docker
    self.client = None
    self.container = None
    if use_docker and (docker is None or not hasattr(docker, "from_env")):
        if not self._persistent:
            shutil.rmtree(self.base_dir, ignore_errors=True)
        raise RuntimeError("Docker is required. Install pygent[docker] and start Docker, "
                           "or explicitly select use_docker=False for trusted local execution.")
    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 as exc:
            if self.client is not None:
                self.client.close()
            if not self._persistent:
                shutil.rmtree(self.base_dir, ignore_errors=True)
            raise RuntimeError("Docker startup failed; local execution was not enabled") from exc
    if not self._use_docker:
        self.client = None
        self.container = None

    try:
        # populate workspace with initial files
        for fp in self._initial_files:
            src = Path(fp).expanduser()
            dest = self.base_dir / src.name
            self.check_copy_tree(src)
            self.check_copy_tree(dest)
            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)

    except Exception:
        self.cleanup()
        raise

    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 = self.base_dir.resolve() 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: float = 600, stream: Optional[Callable[[str], None]] = None, max_output_chars: int = 64000) -> str

Run a shell command with bounded capture and an execution timeout.

Output is streamed in chunks. On POSIX the local process group is killed on timeout (including ordinary shell children). Docker images must provide sh and GNU timeout. Local mode is trusted execution, not a sandbox.

Source code in pygent/runtime.py

def bash(
    self, cmd: str, timeout: float = 600,
    stream: Optional[Callable[[str], None]] = None,
    max_output_chars: int = 64_000,
) -> str:
    """Run a shell command with bounded capture and an execution timeout.

    Output is streamed in chunks. On POSIX the local process group is killed
    on timeout (including ordinary shell children). Docker images must provide
    ``sh`` and GNU ``timeout``. Local mode is trusted execution, not a sandbox.
    """
    if not math.isfinite(timeout) or timeout < 0:
        raise ValueError("timeout must be finite and nonnegative")
    if not isinstance(max_output_chars, int) or max_output_chars < 1:
        raise ValueError("max_output_chars must be a positive integer")
    if timeout == 0:
        return f"$ {cmd}\n[timeout after 0s]"
    tokens = cmd.split()
    if tokens:
        if Path(tokens[0]).name in self.banned_commands:
            return f"$ {cmd}\n[error] command '{Path(tokens[0]).name}' disabled"
        for token in tokens:
            if Path(token).name in self.banned_apps:
                return f"$ {cmd}\n[error] application '{Path(token).name}' disabled"
    parts: list[str] = []
    captured = 0
    truncated = False

    def emit(text: str) -> None:
        nonlocal captured, truncated
        remaining = max_output_chars - captured
        chunk = text[:remaining]
        if chunk:
            parts.append(chunk)
            captured += len(chunk)
            if stream:
                stream(chunk)
        if len(text) > remaining and not truncated:
            truncated = True
            parts.append("\n[output truncated]\n")
            if stream:
                stream("\n[output truncated]\n")

    emit(f"$ {cmd}\n")
    if self._use_docker and self.container is not None:
        try:
            result = self.container.exec_run(
                ["timeout", "--signal=KILL", str(timeout), "sh", "-lc", cmd],
                workdir="/workspace", stream=True, tty=False, stdin=False,
            )
            for chunk in result.output:
                emit(chunk.decode("utf-8", errors="replace"))
        except Exception as exc:
            emit(f"[error] {exc}")
        return "".join(parts)

    proc = subprocess.Popen(
        cmd, shell=True, cwd=self.base_dir, stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
        start_new_session=(os.name == "posix"),
    )
    reader_errors: list[Exception] = []

    def read_output() -> None:
        import codecs
        decoder = codecs.getincrementaldecoder("utf-8")("replace")
        try:
            assert proc.stdout is not None
            while True:
                chunk = os.read(proc.stdout.fileno(), 4096)
                if not chunk:
                    break
                emit(decoder.decode(chunk))
            emit(decoder.decode(b"", final=True))
        except Exception as exc:
            reader_errors.append(exc)

    reader = threading.Thread(target=read_output, daemon=True)
    reader.start()
    timed_out = False
    try:
        proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True
    finally:
        # Also remove background descendants after their parent shell exits.
        if os.name == "posix":
            try:
                os.killpg(proc.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        elif proc.poll() is None:
            proc.kill()
        proc.wait()
        reader.join(timeout=2)
        if not reader.is_alive() and proc.stdout is not None:
            proc.stdout.close()
    if reader_errors:
        raise reader_errors[0]
    if timed_out:
        emit(f"[timeout after {timeout}s]\n")
    return "".join(parts)

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

Resolve a workspace-relative path, rejecting escapes and symlinks.

This is a file API boundary, not an OS sandbox against concurrent hostile filesystem mutation. Shell commands require separate OS isolation.

Source code in pygent/runtime.py

def resolve_path(self, path: Union[str, Path]) -> Path:
    """Resolve a workspace-relative path, rejecting escapes and symlinks.

    This is a file API boundary, not an OS sandbox against concurrent hostile
    filesystem mutation. Shell commands require separate OS isolation.
    """
    value = Path(path)
    if value.is_absolute():
        raise ValueError("workspace paths must be relative")
    candidate = self.base_dir / value
    if candidate.is_symlink() or any(parent.is_symlink() for parent in candidate.parents):
        raise ValueError("symlinks are not allowed in workspace paths")
    resolved = candidate.resolve()
    if not resolved.is_relative_to(self.base_dir):
        raise ValueError("path escapes the workspace")
    return resolved

check_copy_tree(path: Path) -> None staticmethod

Reject symlinks before a recursive copy, including nested entries.

Source code in pygent/runtime.py

@staticmethod
def check_copy_tree(path: Path) -> None:
    """Reject symlinks before a recursive copy, including nested entries."""
    if path.is_symlink() or any(p.is_symlink() for p in path.parents):
        raise ValueError("symlinks are not allowed in copied paths")
    if path.is_dir():
        for root, directories, files in os.walk(path):
            for name in directories + files:
                if (Path(root) / name).is_symlink():
                    raise ValueError("symlinks are not allowed in copied directories")

__enter__() -> Runtime

Source code in pygent/runtime.py

def __enter__(self) -> Runtime:
    return self

__exit__(*exc: object) -> None

Source code in pygent/runtime.py

def __exit__(self, *exc: object) -> None:
    self.cleanup()

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.resolve_path(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.resolve_path(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.resolve_path(Path(dest) if dest else src_path.name)
    self.check_copy_tree(src_path)
    self.check_copy_tree(target)
    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.resolve_path(path)
    if not src.exists():
        return f"file {path} not found"
    dest_path = Path(dest).expanduser()
    self.check_copy_tree(src)
    self.check_copy_tree(dest_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"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)
            self.container = None
    if self.client is not None:
        self.client.close()
        self.client = None
    if not self._persistent:
        shutil.rmtree(self.base_dir, ignore_errors=True)

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

    if not isinstance(args, dict):
        return f"[error] arguments for {name} must be a JSON object"

    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(client: Any = None)

Default model using the OpenAI-compatible API.

Optionally inject a configured client (credentials, timeout and retries).

Source code in pygent/models.py

def __init__(self, client: Any = None) -> None:
    """Optionally inject a configured client (credentials, timeout and retries)."""
    self.client = client

client = client instance-attribute

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
        ]
        kwargs = {"model": model, "messages": serialized}
        if tools:
            kwargs.update(tools=tools, tool_choice="auto")
        client = self.client if self.client is not None else openai
        resp = client.chat.completions.create(**kwargs)
        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."""
    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

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)

Optional legacy API

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")
    self._default_factory = agent_factory is None
    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.RLock()

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

Atomically enforce admission limits and register a background task.

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:
    """Atomically enforce admission limits and register a background task."""
    with self._lock:
        return self._start_task(prompt, parent_rt, files, parent_depth,
                                step_timeout, task_timeout, persona)

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.resolve_path(path)
    if not src.exists():
        return f"file {path} not found"
    dest_path = rt.resolve_path(dest or path)
    Runtime.check_copy_tree(src)
    Runtime.check_copy_tree(dest_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)}"

Task tools module

pygent.task_tools

get_task_manager() -> TaskManager

Return the lazily-created manager used by task tools.

Source code in pygent/task_tools.py

def get_task_manager() -> TaskManager:
    """Return the lazily-created manager used by task tools."""
    return _get_manager()

set_task_manager(manager: TaskManager) -> None

Override the manager instance (used by tests and integration layers).

Source code in pygent/task_tools.py

def set_task_manager(manager: TaskManager) -> None:
    """Override the manager instance (used by tests and integration layers)."""
    global _task_manager
    _task_manager = manager

register_task_tools() -> None

Register task-related tools.

Source code in pygent/task_tools.py

def register_task_tools() -> None:
    """Register task-related tools."""
    register_tool(
        "delegate_task",
        "Create a background task using a new agent and return its ID.",
        {
            "type": "object",
            "properties": {
                "prompt": {"type": "string", "description": "Instruction for the sub-agent"},
                "files": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Files to copy to the sub-agent before starting",
                },
                "persona": {"type": "string", "description": "Persona for the sub-agent"},
                "timeout": {"type": "number", "description": "Max seconds for the task"},
                "step_timeout": {"type": "number", "description": "Time limit per step"},
            },
            "required": ["prompt"],
        },
        lambda rt, **kwargs: _delegate_task(rt, **kwargs),
    )

    register_tool(
        "delegate_persona_task",
        "Create a background task with a specific persona and return its ID.",
        {
            "type": "object",
            "properties": {
                "prompt": {"type": "string", "description": "Instruction for the sub-agent"},
                "persona": {"type": "string", "description": "Persona for the sub-agent"},
                "files": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Files to copy to the sub-agent before starting",
                },
                "timeout": {"type": "number", "description": "Max seconds for the task"},
                "step_timeout": {"type": "number", "description": "Time limit per step"},
            },
            "required": ["prompt", "persona"],
        },
        lambda rt, **kwargs: _delegate_persona_task(rt, **kwargs),
    )

    register_tool(
        "list_personas",
        "Return the available personas for delegated agents.",
        {"type": "object", "properties": {}},
        lambda rt, **kwargs: _list_personas(rt, **kwargs),
    )

    register_tool(
        "task_status",
        "Check the status of a delegated task.",
        {
            "type": "object",
            "properties": {"task_id": {"type": "string"}},
            "required": ["task_id"],
        },
        lambda rt, **kwargs: _task_status(rt, **kwargs),
    )

    register_tool(
        "collect_file",
        "Retrieve a file or directory from a delegated task.",
        {
            "type": "object",
            "properties": {
                "task_id": {"type": "string"},
                "path": {"type": "string"},
                "dest": {"type": "string"},
            },
            "required": ["task_id", "path"],
        },
        lambda rt, **kwargs: _collect_file(rt, **kwargs),
    )

    register_tool(
        "download_file",
        "Return the contents of a file from the workspace.",
        {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "binary": {"type": "boolean"},
            },
            "required": ["path"],
        },
        lambda rt, **kwargs: _download_file(rt, **kwargs),
    )

Harness

pygent.harness

Small, synchronous, provider-neutral agent loop without terminal side effects.

Tools are explicitly supplied per harness. Runs own their transcript and counters; model clients, tool functions and event sinks must support the application's chosen concurrency model. Time limits and cancellation are cooperative between calls.

Status = Literal['completed', 'max_steps', 'max_tool_calls', 'timeout', 'cancelled', 'invalid_output'] module-attribute

RunLimits(max_steps: int = 20, max_tool_calls: int = 100, max_output_chars: int = 16000, max_time: float | None = None) dataclass

Per-run bounds. max_time is checked between blocking calls.

max_steps: int = 20 class-attribute instance-attribute

max_tool_calls: int = 100 class-attribute instance-attribute

max_output_chars: int = 16000 class-attribute instance-attribute

max_time: float | None = None class-attribute instance-attribute

__post_init__() -> None

Source code in pygent/harness.py

def __post_init__(self) -> None:
    for name in ("max_steps", "max_tool_calls", "max_output_chars"):
        value = getattr(self, name)
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError(f"{name} must be a positive integer")
    if self.max_time is not None and (not math.isfinite(self.max_time) or self.max_time <= 0):
        raise ValueError("max_time must be finite and positive")

Tool(name: str, description: str, parameters: Mapping[str, Any], function: Callable[..., Any], requires_approval: bool = False) dataclass

A named callable and its JSON Schema; no implicit global registration.

function receives validated keyword arguments. It must return a string or JSON-serializable value. Set requires_approval for side-effecting actions; these are denied unless a run's approval callback returns exactly True.

name: str instance-attribute

description: str instance-attribute

parameters: Mapping[str, Any] instance-attribute

function: Callable[..., Any] instance-attribute

requires_approval: bool = False class-attribute instance-attribute

__post_init__() -> None

Source code in pygent/harness.py

def __post_init__(self) -> None:
    if not self.name or not callable(self.function):
        raise ValueError("tool needs a nonempty name and a callable")
    schema = deepcopy(dict(self.parameters))
    Draft202012Validator.check_schema(schema)
    if schema.get("type") != "object":
        raise ValueError("tool parameters must have type 'object'")
    object.__setattr__(self, "parameters", schema)

schema() -> dict[str, Any]

Source code in pygent/harness.py

def schema(self) -> dict[str, Any]:
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": deepcopy(dict(self.parameters)),
        },
    }

RunEvent(run_id: str, kind: str, step: int, elapsed: float, tool_name: str | None = None, tool_call_id: str | None = None, outcome: str | None = None) dataclass

Metadata-only event, suitable for an application's telemetry adapter.

run_id: str instance-attribute

kind: str instance-attribute

step: int instance-attribute

elapsed: float instance-attribute

tool_name: str | None = None class-attribute instance-attribute

tool_call_id: str | None = None class-attribute instance-attribute

outcome: str | None = None class-attribute instance-attribute

RunResult(run_id: str, status: Status, output: str | None, messages: list[dict[str, Any]], steps: int, tool_calls: int, elapsed: float, data: Any = None, events: list[RunEvent] = list()) dataclass

Serializable result; only completed denotes successful completion.

run_id: str instance-attribute

status: Status instance-attribute

output: str | None instance-attribute

messages: list[dict[str, Any]] instance-attribute

steps: int instance-attribute

tool_calls: int instance-attribute

elapsed: float instance-attribute

data: Any = None class-attribute instance-attribute

events: list[RunEvent] = field(default_factory=list) class-attribute instance-attribute

to_dict() -> dict[str, Any]

Return an independent JSON-serializable representation.

Source code in pygent/harness.py

def to_dict(self) -> dict[str, Any]:
    """Return an independent JSON-serializable representation."""
    return asdict(self)

Harness(model: Model, *, model_name: str, tools: Sequence[Tool] = (), system_prompt: str = 'Complete the task using the provided tools. Reply when finished.')

Run a chat model with explicit tools, limits and optional approvals.

No runtime is created automatically. Pass bound Runtime methods as tools when filesystem or shell access is intended. Exceptions from models, approval callbacks or event sinks propagate; tools' ordinary exceptions become tool error messages. Tools are never automatically retried.

Source code in pygent/harness.py

def __init__(
    self,
    model: Model,
    *,
    model_name: str,
    tools: Sequence[Tool] = (),
    system_prompt: str = "Complete the task using the provided tools. Reply when finished.",
) -> None:
    if len({tool.name for tool in tools}) != len(tools):
        raise ValueError("tool names must be unique")
    self.model = model
    self.model_name = model_name
    self.system_prompt = system_prompt
    # Copy schema metadata, retaining only the caller's actual function objects.
    self._tools = {
        tool.name: Tool(
            tool.name, tool.description, tool.parameters, tool.function, tool.requires_approval
        )
        for tool in tools
    }

model = model instance-attribute

model_name = model_name instance-attribute

system_prompt = system_prompt instance-attribute

run(prompt: str, *, limits: RunLimits | None = None, history: Sequence[dict[str, Any]] = (), approve: Callable[[str, dict[str, Any]], bool] | None = None, cancel: Event | None = None, on_event: Callable[[RunEvent], None] | None = None, output_schema: Mapping[str, Any] | None = None) -> RunResult

Execute until a final answer or a bound is reached.

history is a trusted complete transcript, e.g. result.messages. A whole tool batch is rejected if it would exceed the tool-call budget. Cancelled/skipped calls receive matching tool responses so the transcript can be continued. Event metadata excludes prompts, arguments and outputs.

Source code in pygent/harness.py

def run(
    self,
    prompt: str,
    *,
    limits: RunLimits | None = None,
    history: Sequence[dict[str, Any]] = (),
    approve: Callable[[str, dict[str, Any]], bool] | None = None,
    cancel: Event | None = None,
    on_event: Callable[[RunEvent], None] | None = None,
    output_schema: Mapping[str, Any] | None = None,
) -> RunResult:
    """Execute until a final answer or a bound is reached.

    ``history`` is a trusted complete transcript, e.g. ``result.messages``.
    A whole tool batch is rejected if it would exceed the tool-call budget.
    Cancelled/skipped calls receive matching tool responses so the transcript
    can be continued. Event metadata excludes prompts, arguments and outputs.
    """
    limits = limits or RunLimits()
    validator = None
    if output_schema is not None:
        Draft202012Validator.check_schema(output_schema)
        validator = Draft202012Validator(deepcopy(dict(output_schema)))
    run_id, start = uuid.uuid4().hex, time.monotonic()
    messages = deepcopy(list(history))
    if not messages:
        messages.append({"role": "system", "content": self.system_prompt})
    messages.append({"role": "user", "content": prompt})
    steps = calls_used = 0
    events: list[RunEvent] = []
    last: Message | None = None

    def emit(kind: str, **kwargs: Any) -> None:
        event = RunEvent(run_id, kind, steps, time.monotonic() - start, **kwargs)
        events.append(event)
        if on_event:
            on_event(event)

    def stopped() -> Status | None:
        if cancel is not None and cancel.is_set():
            return "cancelled"
        if limits.max_time is not None and time.monotonic() - start >= limits.max_time:
            return "timeout"
        return None

    def finish(status: Status, data: Any = None) -> RunResult:
        emit("run_finished", outcome=status)
        return RunResult(
            run_id,
            status,
            last.content if last else None,
            messages,
            steps,
            calls_used,
            time.monotonic() - start,
            data,
            events,
        )

    emit("run_started")
    for _ in range(limits.max_steps):
        reason = stopped()
        if reason:
            return finish(reason)
        steps += 1
        emit("model_started")
        last = parse_message(
            self.model.chat(
                deepcopy(messages),
                self.model_name,
                [tool.schema() for tool in self._tools.values()],
            )
        )
        if last.role != "assistant":
            raise ValueError("model must return an assistant message")
        calls = last.tool_calls or []
        if any(not call.id for call in calls) or len({c.id for c in calls}) != len(calls):
            raise ValueError("tool call IDs must be nonempty and unique within a batch")
        message: dict[str, Any] = {"role": last.role, "content": last.content}
        if calls:
            message["tool_calls"] = [asdict(call) for call in calls]
        messages.append(message)
        emit("model_finished")
        reason = stopped()
        if not calls:
            if reason:
                return finish(reason)
            if validator is not None:
                try:
                    data = json.loads(last.content or "", parse_constant=_reject_constant)
                    validator.validate(data)
                except (ValueError, ValidationError):
                    return finish("invalid_output")
                return finish("completed", data)
            return finish("completed")
        if calls_used + len(calls) > limits.max_tool_calls:
            reason = reason or "max_tool_calls"
        for call in calls:
            reason = reason or stopped()
            name = call.function.name
            outcome = "skipped"
            if reason:
                output = f"[error] tool not executed: {reason}"
            else:
                calls_used += 1
                tool = self._tools.get(name)
                outcome = "error"
                try:
                    if tool is None:
                        raise ValueError(f"unknown tool: {name}")
                    args = json.loads(
                        call.function.arguments or "{}", parse_constant=_reject_constant
                    )
                    Draft202012Validator(tool.parameters).validate(args)
                except (ValueError, TypeError, ValidationError) as exc:
                    output = f"[error] invalid tool call: {exc}"
                else:
                    if tool.requires_approval and (
                        approve is None or approve(name, deepcopy(args)) is not True
                    ):
                        output, outcome = "[error] tool execution denied", "denied"
                    elif stopped():
                        reason = stopped()
                        output = f"[error] tool not executed: {reason}"
                        outcome = "skipped"
                    else:
                        emit("tool_started", tool_name=name, tool_call_id=call.id)
                        try:
                            value = tool.function(**args)
                            output = (
                                value
                                if isinstance(value, str)
                                else json.dumps(value, allow_nan=False)
                            )
                            outcome = "success"
                        except Exception as exc:
                            output = f"[error] tool failed: {type(exc).__name__}: {exc}"
            if len(output) > limits.max_output_chars:
                output = output[: limits.max_output_chars] + "\n[output truncated]"
            messages.append({"role": "tool", "tool_call_id": call.id, "content": output})
            emit("tool_finished", tool_name=name, tool_call_id=call.id, outcome=outcome)
        reason = reason or stopped()
        if reason:
            return finish(reason)
    return finish("max_steps")

Offline testing

pygent.testing

Deterministic models for tests, examples and offline harness evaluation.

ScriptedModel(responses: Sequence[Message | dict[str, Any]])

Return each supplied response once; fail if the script is exhausted.

Create one instance per concurrent run. Recorded requests are independent copies, so tests can assert tool schemas and transcript protocol details.

Source code in pygent/testing.py

def __init__(self, responses: Sequence[Message | dict[str, Any]]) -> None:
    self._responses = deepcopy(list(responses))
    self.requests: list[dict[str, Any]] = []

requests: list[dict[str, Any]] = [] instance-attribute

chat(messages: list[dict[str, Any]], model: str, tools: Any) -> Message

Source code in pygent/testing.py

def chat(self, messages: list[dict[str, Any]], model: str, tools: Any) -> Message:
    index = len(self.requests)
    if index >= len(self._responses):
        raise RuntimeError("ScriptedModel responses exhausted")
    self.requests.append(deepcopy({"messages": messages, "model": model, "tools": tools}))
    return parse_message(deepcopy(self._responses[index]))