Agent

stirrup.core.agent

AGENT_MAX_TURNS module-attribute

AGENT_MAX_TURNS = 30

CONTEXT_SUMMARIZATION_CUTOFF module-attribute

CONTEXT_SUMMARIZATION_CUTOFF = 0.7

FINISH_TOOL_NAME module-attribute

FINISH_TOOL_NAME: Literal['finish'] = 'finish'

TURNS_REMAINING_WARNING_THRESHOLD module-attribute

TURNS_REMAINING_WARNING_THRESHOLD = 20

MESSAGE_SUMMARIZER module-attribute

MESSAGE_SUMMARIZER = read_text(encoding='utf-8')

MESSAGE_SUMMARIZER_BRIDGE_TEMPLATE module-attribute

MESSAGE_SUMMARIZER_BRIDGE_TEMPLATE = read_text(
    encoding="utf-8"
)

DEFAULT_TOOLS module-attribute

DEFAULT_TOOLS: list[Tool[Any, Any] | ToolProvider] = [
    LocalCodeExecToolProvider(),
    WebToolProvider(),
]

SIMPLE_FINISH_TOOL module-attribute

SIMPLE_FINISH_TOOL: Tool[
    FinishParams, ToolUseCountMetadata
] = Tool[FinishParams, ToolUseCountMetadata](
    name=FINISH_TOOL_NAME,
    description="Signal task completion with a reason. Use when the task is finished or cannot proceed further. Note that you will need a separate turn to finish.",
    parameters=FinishParams,
    executor=_validating_finish_executor,
)

_PARENT_DEPTH module-attribute

_PARENT_DEPTH: ContextVar[int] = ContextVar(
    "parent_depth", default=0
)

logger module-attribute

logger = getLogger(__name__)

_SESSION_STATE module-attribute

_SESSION_STATE: ContextVar[SessionState] = ContextVar(
    "session_state"
)

__all__ module-attribute

__all__ = ['Agent', 'SubAgentParams']

LOGGER module-attribute

LOGGER = getLogger(__name__)

DEFAULT_SUB_AGENT_DESCRIPTION module-attribute

DEFAULT_SUB_AGENT_DESCRIPTION = "A sub agent that can be used to handle a contained, specific task."

AGENT_NAME_PATTERN module-attribute

AGENT_NAME_PATTERN = compile('^[a-zA-Z0-9_-]{1,128}$')

ChatMessage

ChatMessage = Annotated[
    SystemMessage
    | UserMessage
    | AssistantMessage
    | ToolMessage,
    Field(discriminator=role),
]

Discriminated union of all message types, automatically parsed based on role field.

CacheManager

CacheManager(
    cache_base_dir: Path | None = None,
    clear_on_success: bool = True,
)

Manages cache operations for agent sessions.

Handles saving/loading cache state and execution environment files.

Initialize CacheManager.

Parameters:

Name	Type	Description	Default
cache_base_dir	Path | None	Base directory for cache storage. Defaults to ~/.cache/stirrup/	None
clear_on_success	bool	If True (default), automatically clear the cache when the agent completes successfully. Set to False to preserve caches for inspection or manual management.	True

Source code in src/stirrup/core/cache.py

def __init__(
    self,
    cache_base_dir: Path | None = None,
    clear_on_success: bool = True,
) -> None:
    """Initialize CacheManager.

    Args:
        cache_base_dir: Base directory for cache storage.
                       Defaults to ~/.cache/stirrup/
        clear_on_success: If True (default), automatically clear the cache when
                         the agent completes successfully. Set to False to preserve
                         caches for inspection or manual management.
    """
    self._cache_base_dir = cache_base_dir or DEFAULT_CACHE_DIR
    self.clear_on_success = clear_on_success

save_state

save_state(
    task_hash: str,
    state: CacheState,
    exec_env_dir: Path | None = None,
) -> None

Save cache state and optionally archive execution environment files.

Uses atomic writes to prevent corrupted cache files if interrupted mid-write.

Parameters:

Name	Type	Description	Default
task_hash	str	Unique identifier for this task/cache.	required
state	CacheState	CacheState to persist.	required
exec_env_dir	Path | None	Optional path to execution environment temp directory. If provided, all files will be copied to cache.	None

Source code in src/stirrup/core/cache.py

def save_state(
    self,
    task_hash: str,
    state: CacheState,
    exec_env_dir: Path | None = None,
) -> None:
    """Save cache state and optionally archive execution environment files.

    Uses atomic writes to prevent corrupted cache files if interrupted mid-write.

    Args:
        task_hash: Unique identifier for this task/cache.
        state: CacheState to persist.
        exec_env_dir: Optional path to execution environment temp directory.
                     If provided, all files will be copied to cache.
    """
    cache_dir = self._get_cache_dir(task_hash)
    cache_dir.mkdir(parents=True, exist_ok=True)

    # Save state JSON using atomic write (write to temp file, then rename)
    state_file = self._get_state_file(task_hash)
    temp_file = state_file.with_suffix(".json.tmp")

    try:
        state_data = state.to_dict()
        logger.debug("Serialized cache state: turn=%d, msgs=%d", state.turn, len(state.msgs))

        with open(temp_file, "w", encoding="utf-8") as f:
            json.dump(state_data, f, indent=2, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())  # Ensure data is written to disk

        logger.debug("Wrote temp file: %s", temp_file)

        # Atomic rename (on POSIX systems)
        temp_file.replace(state_file)
        logger.info("Saved cache state to %s (turn %d)", state_file, state.turn)
    except Exception as e:
        logger.exception("Failed to save cache state: %s", e)
        # Try direct write as fallback
        try:
            logger.warning("Attempting direct write as fallback")
            with open(state_file, "w", encoding="utf-8") as f:
                json.dump(state_data, f, indent=2, ensure_ascii=False)
                f.flush()
                os.fsync(f.fileno())
            logger.info("Fallback write succeeded to %s", state_file)
        except Exception as e2:
            logger.exception("Fallback write also failed: %s", e2)
        # Clean up temp file if it exists
        if temp_file.exists():
            temp_file.unlink()
        raise

    # Copy execution environment files if provided
    if exec_env_dir and exec_env_dir.exists():
        files_dir = self._get_files_dir(task_hash)
        if files_dir.exists():
            shutil.rmtree(files_dir)  # Clear existing files
        shutil.copytree(exec_env_dir, files_dir, dirs_exist_ok=True)
        logger.info("Saved execution environment files to %s", files_dir)

load_state

load_state(task_hash: str) -> CacheState | None

Load cached state for a task hash.

Parameters:

Name	Type	Description	Default
task_hash	str	Unique identifier for the task/cache.	required

Returns:

Type	Description
CacheState | None	CacheState if cache exists, None otherwise.

Source code in src/stirrup/core/cache.py

def load_state(self, task_hash: str) -> CacheState | None:
    """Load cached state for a task hash.

    Args:
        task_hash: Unique identifier for the task/cache.

    Returns:
        CacheState if cache exists, None otherwise.
    """
    state_file = self._get_state_file(task_hash)
    if not state_file.exists():
        logger.debug("No cache found for task %s", task_hash)
        return None

    try:
        with open(state_file, encoding="utf-8") as f:
            data = json.load(f)
        state = CacheState.from_dict(data)
        logger.info("Loaded cache state from %s (turn %d)", state_file, state.turn)
        return state
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        logger.warning("Failed to load cache for task %s: %s", task_hash, e)
        return None

restore_files

restore_files(task_hash: str, dest_dir: Path) -> bool

Restore cached files to the destination directory.

Parameters:

Name	Type	Description	Default
task_hash	str	Unique identifier for the task/cache.	required
dest_dir	Path	Destination directory (typically the new exec env temp dir).	required

Returns:

Type	Description
bool	True if files were restored, False if no files cache exists.

Source code in src/stirrup/core/cache.py

def restore_files(self, task_hash: str, dest_dir: Path) -> bool:
    """Restore cached files to the destination directory.

    Args:
        task_hash: Unique identifier for the task/cache.
        dest_dir: Destination directory (typically the new exec env temp dir).

    Returns:
        True if files were restored, False if no files cache exists.
    """
    files_dir = self._get_files_dir(task_hash)
    if not files_dir.exists():
        logger.debug("No cached files for task %s", task_hash)
        return False

    # Copy all files from cache to destination
    for item in files_dir.iterdir():
        dest_item = dest_dir / item.name
        if item.is_file():
            shutil.copy2(item, dest_item)
        else:
            shutil.copytree(item, dest_item, dirs_exist_ok=True)

    logger.info("Restored cached files from %s to %s", files_dir, dest_dir)
    return True

clear_cache

clear_cache(task_hash: str) -> None

Remove cache for a specific task.

Called after successful completion to clean up.

Parameters:

Name	Type	Description	Default
task_hash	str	Unique identifier for the task/cache.	required

Source code in src/stirrup/core/cache.py

def clear_cache(self, task_hash: str) -> None:
    """Remove cache for a specific task.

    Called after successful completion to clean up.

    Args:
        task_hash: Unique identifier for the task/cache.
    """
    cache_dir = self._get_cache_dir(task_hash)
    if cache_dir.exists():
        shutil.rmtree(cache_dir)
        logger.info("Cleared cache for task %s", task_hash)

list_caches

list_caches() -> list[str]

List all available cache hashes.

Returns:

Type	Description
list[str]	List of task hashes with existing caches.

Source code in src/stirrup/core/cache.py

def list_caches(self) -> list[str]:
    """List all available cache hashes.

    Returns:
        List of task hashes with existing caches.
    """
    if not self._cache_base_dir.exists():
        return []

    return [d.name for d in self._cache_base_dir.iterdir() if d.is_dir() and (d / "state.json").exists()]

get_cache_info

get_cache_info(task_hash: str) -> dict | None

Get metadata about a cache without fully loading it.

Parameters:

Name	Type	Description	Default
task_hash	str	Unique identifier for the task/cache.	required

Returns:

Type	Description
dict | None	Dictionary with cache info (turn, timestamp, agent_name) or None.

Source code in src/stirrup/core/cache.py

def get_cache_info(self, task_hash: str) -> dict | None:
    """Get metadata about a cache without fully loading it.

    Args:
        task_hash: Unique identifier for the task/cache.

    Returns:
        Dictionary with cache info (turn, timestamp, agent_name) or None.
    """
    state_file = self._get_state_file(task_hash)
    if not state_file.exists():
        return None

    try:
        with open(state_file, encoding="utf-8") as f:
            data = json.load(f)
        return {
            "task_hash": task_hash,
            "turn": data.get("turn", 0),
            "timestamp": data.get("timestamp", ""),
            "agent_name": data.get("agent_name", ""),
            "has_files": self._get_files_dir(task_hash).exists(),
        }
    except (json.JSONDecodeError, KeyError):
        return None

CacheState dataclass

CacheState(
    msgs: list[ChatMessage],
    full_msg_history: list[list[ChatMessage]],
    turn: int,
    run_metadata: dict[str, list[Any]],
    task_hash: str,
    timestamp: str = (lambda: isoformat())(),
    agent_name: str = "",
)

Serializable state for resuming an agent run.

Captures all necessary state to resume execution from a specific turn.

msgs instance-attribute

msgs: list[ChatMessage]

Current conversation messages in the active run loop.

full_msg_history instance-attribute

full_msg_history: list[list[ChatMessage]]

Groups of messages (separated when context summarization occurs).

turn instance-attribute

turn: int

Current turn number (0-indexed) - resume will start from this turn.

run_metadata instance-attribute

run_metadata: dict[str, list[Any]]

Accumulated tool metadata from the run.

task_hash instance-attribute

task_hash: str

Hash of the original init_msgs for verification on resume.

timestamp class-attribute instance-attribute

timestamp: str = field(default_factory=lambda: isoformat())

ISO timestamp when cache was created.

agent_name class-attribute instance-attribute

agent_name: str = ''

Name of the agent that created this cache.

to_dict

to_dict() -> dict

Convert to JSON-serializable dictionary.

Source code in src/stirrup/core/cache.py

def to_dict(self) -> dict:
    """Convert to JSON-serializable dictionary."""
    return {
        "msgs": serialize_messages(self.msgs),
        "full_msg_history": [serialize_messages(group) for group in self.full_msg_history],
        "turn": self.turn,
        "run_metadata": _serialize_run_metadata(self.run_metadata),
        "task_hash": self.task_hash,
        "timestamp": self.timestamp,
        "agent_name": self.agent_name,
    }

from_dict classmethod

from_dict(data: dict) -> CacheState

Create CacheState from JSON dictionary.

Source code in src/stirrup/core/cache.py

@classmethod
def from_dict(cls, data: dict) -> "CacheState":
    """Create CacheState from JSON dictionary."""
    return cls(
        msgs=deserialize_messages(data["msgs"]),
        full_msg_history=[deserialize_messages(group) for group in data["full_msg_history"]],
        turn=data["turn"],
        run_metadata=data["run_metadata"],
        task_hash=data["task_hash"],
        timestamp=data.get("timestamp", ""),
        agent_name=data.get("agent_name", ""),
    )

AssistantMessage

Bases: BaseModel

LLM response message with optional tool calls and token usage tracking.

ImageContentBlock

Bases: BinaryContentBlock

Image content supporting PNG, JPEG, WebP, PSD formats with automatic downscaling.

to_base64_url

to_base64_url(
    max_pixels: int | None = RESOLUTION_1MP,
) -> str

Convert image to base64 data URL, optionally resizing to max pixel count.

Source code in src/stirrup/core/models.py

def to_base64_url(self, max_pixels: int | None = RESOLUTION_1MP) -> str:
    """Convert image to base64 data URL, optionally resizing to max pixel count."""
    img: Image.Image = Image.open(BytesIO(self.data))
    if max_pixels is not None and img.width * img.height > max_pixels:
        tw, th = downscale_image(img.width, img.height, max_pixels)
        img.thumbnail((tw, th), Image.Resampling.LANCZOS)
    if img.mode != "RGB":
        img = img.convert("RGB")
    buf = BytesIO()
    img.save(buf, format="PNG")
    return f"data:image/png;base64,{b64encode(buf.getvalue()).decode()}"

LLMClient

Bases: Protocol

Protocol defining the interface for LLM client implementations.

Any LLM client must implement this protocol to work with the Agent class. Provides text generation with tool support and model capability inspection.

SubAgentMetadata

Bases: BaseModel

Metadata from sub-agent execution including token usage, message history, and child run metadata.

Implements Addable protocol to support aggregation across multiple subagent calls.

__add__

__add__(other: SubAgentMetadata) -> SubAgentMetadata

Combine metadata from multiple subagent calls.

Source code in src/stirrup/core/models.py

def __add__(self, other: "SubAgentMetadata") -> "SubAgentMetadata":
    """Combine metadata from multiple subagent calls."""
    # Concatenate message histories
    combined_history = self.message_history + other.message_history
    # Merge run metadata (concatenate lists per key)
    combined_meta: dict[str, list[Any]] = dict(self.run_metadata)
    for key, metadata_list in other.run_metadata.items():
        if key in combined_meta:
            combined_meta[key] = combined_meta[key] + metadata_list
        else:
            combined_meta[key] = list(metadata_list)
    return SubAgentMetadata(
        message_history=combined_history,
        run_metadata=combined_meta,
    )

SystemMessage

Bases: BaseModel

System-level instructions and context for the LLM.

TokenUsage

Bases: BaseModel

Token counts for LLM usage (input, output, reasoning tokens).

total property

total: int

Total token count across input, output, and reasoning.

__add__

__add__(other: TokenUsage) -> TokenUsage

Add two TokenUsage objects together, summing each field independently.

Source code in src/stirrup/core/models.py

def __add__(self, other: "TokenUsage") -> "TokenUsage":
    """Add two TokenUsage objects together, summing each field independently."""
    return TokenUsage(
        input=self.input + other.input,
        output=self.output + other.output,
        reasoning=self.reasoning + other.reasoning,
    )

Tool

Bases: BaseModel

Tool definition with name, description, parameter schema, and executor function.

Generic over

P: Parameter model type (Pydantic BaseModel subclass, or EmptyParams for parameterless tools) M: Metadata type (should implement Addable for aggregation; use None for tools without metadata)

Tools are simple, stateless callables. For tools requiring lifecycle management (setup/teardown, resource pooling), use a ToolProvider instead.

Example with parameters

class CalcParams(BaseModel):
    expression: str

calc_tool = Tool[CalcParams, None](
    name="calc",
    description="Evaluate math",
    parameters=CalcParams,
    executor=lambda p: ToolResult(content=str(eval(p.expression))),
)

Example without parameters (uses EmptyParams by default):

time_tool = Tool[EmptyParams, None](
    name="time",
    description="Get current time",
    executor=lambda _: ToolResult(content=datetime.now().isoformat()),
)

ToolCall

Bases: BaseModel

Represents a tool invocation request from the LLM.

Attributes:

Name	Type	Description
name	str	Name of the tool to invoke
arguments	str	JSON string containing tool parameters
tool_call_id	str | None	Unique identifier for tracking this tool call and its result

ToolMessage

Bases: BaseModel

Tool execution result returned to the LLM.

Attributes:

Name	Type	Description
role	Literal['tool']	Always "tool"
content	Content	The tool result content
tool_call_id	str | None	ID linking this result to the corresponding tool call
name	str | None	Name of the tool that was called
args_was_valid	bool	Whether the tool arguments were valid
success	bool	Whether the tool executed successfully (used by finish tool to control termination)

ToolProvider

Bases: ABC

Abstract base class for tool providers with lifecycle management.

ToolProviders manage resources (HTTP clients, sandboxes, server connections) and return Tool instances when entering their async context. They implement the async context manager protocol.

Use ToolProvider for: - Tools requiring setup/teardown (connections, temp directories) - Tools that return multiple Tool instances (e.g., MCP servers) - Tools with shared state across calls (e.g., HTTP client pooling)

Example

class MyToolProvider(ToolProvider): async def aenter(self) -> Tool | list[Tool]: # Setup resources and return tool(s) return self._create_tool()

# __aexit__ is optional - default is no-op

Agent automatically manages ToolProvider lifecycle via its session() context.

__aenter__ abstractmethod async

__aenter__() -> Tool | list[Tool]

Enter async context: setup resources and return tool(s).

Returns:

Type	Description
Tool | list[Tool]	A single Tool instance, or a list of Tool instances for providers
Tool | list[Tool]	that expose multiple tools (e.g., MCP servers).

Source code in src/stirrup/core/models.py

@abstractmethod
async def __aenter__(self) -> "Tool | list[Tool]":
    """Enter async context: setup resources and return tool(s).

    Returns:
        A single Tool instance, or a list of Tool instances for providers
        that expose multiple tools (e.g., MCP servers).
    """
    ...

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit async context: cleanup resources. Default: no-op.

Source code in src/stirrup/core/models.py

async def __aexit__(  # noqa: B027
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit async context: cleanup resources. Default: no-op."""

ToolResult

Bases: BaseModel

Result from a tool executor with optional metadata.

Generic over metadata type M. M should implement Addable protocol for aggregation support, but this is not enforced at the class level due to Pydantic schema generation limitations.

Attributes:

Name	Type	Description
content	Content	The result content (string, list of content blocks, or images)
success	bool	Whether the tool call was successful. For finish tools, controls if agent terminates.
metadata	M | None	Optional metadata (e.g., usage stats) that implements Addable for aggregation

UserMessage

Bases: BaseModel

User input message to the LLM.

SkillMetadata dataclass

SkillMetadata(name: str, description: str, path: str)

Metadata extracted from a skill's SKILL.md frontmatter.

CodeExecToolProvider

CodeExecToolProvider(
    *, allowed_commands: list[str] | None = None
)

Bases: ToolProvider, ABC

Abstract base class for code execution tool providers.

CodeExecToolProvider is a ToolProvider that manages code execution environments (sandboxes, containers, local temp directories) and returns a code_exec Tool.

Subclasses must implement: - aenter(): Initialize environment and return the code_exec tool - aexit(): Cleanup the execution environment - run_command(): Execute a command and return raw result - read_file_bytes(): Read file content as bytes from the environment - write_file_bytes(): Write bytes to a file in the environment

Default implementations are provided for: - save_output_files(): Save files to local dir or another exec env (uses primitives) - upload_files(): Upload files from local or another exec env (uses primitives)

All code execution providers support an optional allowlist of command patterns. If provided, only commands matching at least one pattern are allowed. If None, all commands are allowed.

Usage with Agent

from stirrup.clients.chat_completions_client import ChatCompletionsClient

client = ChatCompletionsClient(model="gpt-5") agent = Agent( client=client, name="assistant", tools=[LocalCodeExecToolProvider(), CALCULATOR_TOOL], )

Initialize execution environment with optional command allowlist.

Parameters:

Name	Type	Description	Default
allowed_commands	list[str] | None	Optional list of regex patterns. If provided, only commands matching at least one pattern are allowed. If None, all commands are allowed.	None

Source code in src/stirrup/tools/code_backends/base.py

def __init__(self, *, allowed_commands: list[str] | None = None) -> None:
    """Initialize execution environment with optional command allowlist.

    Args:
        allowed_commands: Optional list of regex patterns. If provided, only
                         commands matching at least one pattern are allowed.
                         If None, all commands are allowed.

    """
    self._allowed_commands = allowed_commands
    self._compiled_allowed: list[re.Pattern[str]] | None = None
    if allowed_commands is not None:
        self._compiled_allowed = [re.compile(p) for p in allowed_commands]

temp_dir property

temp_dir: Path | None

Return the temporary directory for this execution environment, if any.

__aenter__ abstractmethod async

__aenter__() -> Tool[
    CodeExecutionParams, ToolUseCountMetadata
]

Enter async context: set up environment and return code_exec tool.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def __aenter__(self) -> Tool[CodeExecutionParams, ToolUseCountMetadata]:
    """Enter async context: set up environment and return code_exec tool."""
    ...

__aexit__ abstractmethod async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Exit async context: cleanup the execution environment.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Exit async context: cleanup the execution environment."""
    ...

run_command abstractmethod async

run_command(
    cmd: str, *, timeout: int = SHELL_TIMEOUT
) -> CommandResult

Execute a shell command and return raw CommandResult.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def run_command(self, cmd: str, *, timeout: int = SHELL_TIMEOUT) -> CommandResult:
    """Execute a shell command and return raw CommandResult."""
    ...

read_file_bytes abstractmethod async

read_file_bytes(path: str) -> bytes

Read file content as bytes from this execution environment.

Parameters:

Name	Type	Description	Default
path	str	File path within this execution environment (relative or absolute within the env's working directory).	required

Returns:

Type	Description
bytes	File contents as bytes.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def read_file_bytes(self, path: str) -> bytes:
    """Read file content as bytes from this execution environment.

    Args:
        path: File path within this execution environment (relative or absolute
              within the env's working directory).

    Returns:
        File contents as bytes.

    Raises:
        FileNotFoundError: If file does not exist.
        RuntimeError: If execution environment not started.

    """
    ...

write_file_bytes abstractmethod async

write_file_bytes(path: str, content: bytes) -> None

Write bytes to a file in this execution environment.

Parameters:

Name	Type	Description	Default
path	str	Destination path within this execution environment.	required
content	bytes	File contents to write.	required

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def write_file_bytes(self, path: str, content: bytes) -> None:
    """Write bytes to a file in this execution environment.

    Args:
        path: Destination path within this execution environment.
        content: File contents to write.

    Raises:
        RuntimeError: If execution environment not started.

    """
    ...

file_exists abstractmethod async

file_exists(path: str) -> bool

Check if a file exists in this execution environment.

Parameters:

Name	Type	Description	Default
path	str	File path within this execution environment (relative or absolute within the env's working directory).	required

Returns:

Type	Description
bool	True if the file exists, False otherwise.

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def file_exists(self, path: str) -> bool:
    """Check if a file exists in this execution environment.

    Args:
        path: File path within this execution environment (relative or absolute
              within the env's working directory).

    Returns:
        True if the file exists, False otherwise.

    Raises:
        RuntimeError: If execution environment not started.

    """
    ...

is_directory abstractmethod async

is_directory(path: str) -> bool

Check if a path is a directory in this execution environment.

Parameters:

Name	Type	Description	Default
path	str	Path within this execution environment.	required

Returns:

Type	Description
bool	True if the path exists and is a directory, False otherwise.

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def is_directory(self, path: str) -> bool:
    """Check if a path is a directory in this execution environment.

    Args:
        path: Path within this execution environment.

    Returns:
        True if the path exists and is a directory, False otherwise.

    Raises:
        RuntimeError: If execution environment not started.

    """
    ...

list_files abstractmethod async

list_files(path: str) -> list[str]

List all files recursively in a directory within this execution environment.

Parameters:

Name	Type	Description	Default
path	str	Directory path within this execution environment.	required

Returns:

Type	Description
list[str]	List of file paths (relative to the given path) for all files in the directory.
list[str]	Returns an empty list if the path is a file or doesn't exist.

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def list_files(self, path: str) -> list[str]:
    """List all files recursively in a directory within this execution environment.

    Args:
        path: Directory path within this execution environment.

    Returns:
        List of file paths (relative to the given path) for all files in the directory.
        Returns an empty list if the path is a file or doesn't exist.

    Raises:
        RuntimeError: If execution environment not started.

    """
    ...

save_output_files async

save_output_files(
    paths: list[str],
    output_dir: Path | str,
    dest_env: CodeExecToolProvider | None = None,
) -> SaveOutputFilesResult

Save files from this execution environment to a destination.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths in this execution environment to save.	required
output_dir	Path | str	Directory path to save files to.	required
dest_env	CodeExecToolProvider | None	If provided, output_dir is interpreted as a path within dest_env (cross-environment transfer). If None, output_dir is a local filesystem path.	None

Returns:

Type	Description
SaveOutputFilesResult	SaveOutputFilesResult containing lists of saved files and any failures.

Source code in src/stirrup/tools/code_backends/base.py

async def save_output_files(
    self,
    paths: list[str],
    output_dir: Path | str,
    dest_env: "CodeExecToolProvider | None" = None,
) -> SaveOutputFilesResult:
    """Save files from this execution environment to a destination.

    Args:
        paths: List of file paths in this execution environment to save.
        output_dir: Directory path to save files to.
        dest_env: If provided, output_dir is interpreted as a path within dest_env
                  (cross-environment transfer). If None, output_dir is a local
                  filesystem path.

    Returns:
        SaveOutputFilesResult containing lists of saved files and any failures.

    """
    result = SaveOutputFilesResult()
    output_dir_str = str(output_dir)

    for source_path in paths:
        try:
            content = await self.read_file_bytes(source_path)
            filename = Path(source_path).name
            dest_path = f"{output_dir_str}/{filename}"

            if dest_env:
                # Transfer to another exec env (cross-environment)
                logger.debug(
                    "CROSS-ENV TRANSFER: %s (%d bytes) -> %s (dest_env: %s)",
                    source_path,
                    len(content),
                    dest_path,
                    type(dest_env).__name__,
                )
                await dest_env.write_file_bytes(dest_path, content)
                result.saved.append(SavedFile(source_path, Path(dest_path), len(content)))
            else:
                # Save to local filesystem
                output_path = Path(output_dir) / filename
                output_path.parent.mkdir(parents=True, exist_ok=True)
                logger.debug(
                    "SAVE TO LOCAL: %s (%d bytes) -> %s",
                    source_path,
                    len(content),
                    output_path,
                )
                output_path.write_bytes(content)
                result.saved.append(SavedFile(source_path, output_path, len(content)))
        except Exception as e:
            logger.debug("TRANSFER FAILED: %s -> %s: %s", source_path, output_dir_str, e)
            result.failed[source_path] = str(e)

    return result

upload_files async

upload_files(
    *paths: Path | str,
    source_env: CodeExecToolProvider | None = None,
    dest_dir: str | None = None,
) -> UploadFilesResult

Upload files to this execution environment.

Parameters:

Name	Type	Description	Default
*paths	Path | str	File or directory paths to upload. If source_env is None, these are local filesystem paths. If source_env is provided, these are paths within source_env (cross-environment transfer).	()
source_env	CodeExecToolProvider | None	If provided, paths are within source_env. If None, paths are local filesystem paths.	None
dest_dir	str | None	Destination directory in this environment. If None, uses the environment's working directory.	None

Returns:

Type	Description
UploadFilesResult	UploadFilesResult containing lists of uploaded files and any failures.

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

async def upload_files(
    self,
    *paths: Path | str,
    source_env: "CodeExecToolProvider | None" = None,
    dest_dir: str | None = None,
) -> UploadFilesResult:
    """Upload files to this execution environment.

    Args:
        *paths: File or directory paths to upload. If source_env is None, these
                are local filesystem paths. If source_env is provided, these are
                paths within source_env (cross-environment transfer).
        source_env: If provided, paths are within source_env. If None, paths are
                    local filesystem paths.
        dest_dir: Destination directory in this environment.
                  If None, uses the environment's working directory.

    Returns:
        UploadFilesResult containing lists of uploaded files and any failures.

    Raises:
        RuntimeError: If execution environment not started.

    """
    result = UploadFilesResult()
    dest_dir_str = dest_dir or ""

    for path in paths:
        path_str = str(path)
        try:
            if source_env:
                # Cross-environment transfer: read from source_env
                # Check if it's a directory first
                if await source_env.is_directory(path_str):
                    # Handle directory recursively
                    # Preserve directory name when dest_dir not specified
                    dir_name = Path(path_str).name
                    files = await source_env.list_files(path_str)
                    for rel_file_path in files:
                        src_file_path = f"{path_str}/{rel_file_path}"
                        # If dest_dir specified, put files directly there
                        # Otherwise, preserve the source directory name
                        if dest_dir_str:
                            dest_path = f"{dest_dir_str}/{rel_file_path}"
                        else:
                            dest_path = f"{dir_name}/{rel_file_path}"
                        content = await source_env.read_file_bytes(src_file_path)
                        logger.debug(
                            "UPLOAD CROSS-ENV (dir): %s (%d bytes) from %s -> %s",
                            src_file_path,
                            len(content),
                            type(source_env).__name__,
                            dest_path,
                        )
                        await self.write_file_bytes(dest_path, content)
                        result.uploaded.append(UploadedFile(Path(src_file_path), dest_path, len(content)))
                else:
                    # Single file transfer
                    content = await source_env.read_file_bytes(path_str)
                    filename = Path(path_str).name
                    dest_path = f"{dest_dir_str}/{filename}" if dest_dir_str else filename
                    logger.debug(
                        "UPLOAD CROSS-ENV: %s (%d bytes) from %s -> %s",
                        path_str,
                        len(content),
                        type(source_env).__name__,
                        dest_path,
                    )
                    await self.write_file_bytes(dest_path, content)
                    result.uploaded.append(UploadedFile(Path(path_str), dest_path, len(content)))
            else:
                # Local filesystem upload - must be handled by subclass
                # This is a fallback that reads from local fs and writes to env
                local_path = Path(path)
                if local_path.is_dir():
                    # Handle directory recursively
                    for file_path in local_path.rglob("*"):
                        if file_path.is_file():
                            rel_path = file_path.relative_to(local_path)
                            dest_path = f"{dest_dir_str}/{rel_path}" if dest_dir_str else str(rel_path)
                            content = file_path.read_bytes()
                            logger.debug(
                                "UPLOAD FROM LOCAL: %s (%d bytes) -> %s",
                                file_path,
                                len(content),
                                dest_path,
                            )
                            await self.write_file_bytes(dest_path, content)
                            result.uploaded.append(UploadedFile(file_path, dest_path, len(content)))
                else:
                    filename = local_path.name
                    dest_path = f"{dest_dir_str}/{filename}" if dest_dir_str else filename
                    content = local_path.read_bytes()
                    logger.debug(
                        "UPLOAD FROM LOCAL: %s (%d bytes) -> %s",
                        local_path,
                        len(content),
                        dest_path,
                    )
                    await self.write_file_bytes(dest_path, content)
                    result.uploaded.append(UploadedFile(local_path, dest_path, len(content)))
        except Exception as e:
            logger.debug("UPLOAD FAILED: %s -> %s: %s", path_str, dest_dir_str, e)
            result.failed[path_str] = str(e)

    return result

get_code_exec_tool

get_code_exec_tool(
    *,
    name: str = "code_exec",
    description: str | None = None,
) -> Tool[CodeExecutionParams, ToolUseCountMetadata]

Create a code execution tool for this environment.

Parameters:

Name	Type	Description	Default
name	str	Tool name	'code_exec'
description	str | None	Tool description	None

Returns:

Type	Description
Tool[CodeExecutionParams, ToolUseCountMetadata]	Tool[CodeExecutionParams] that executes commands in this environment

Source code in src/stirrup/tools/code_backends/base.py

def get_code_exec_tool(
    self,
    *,
    name: str = "code_exec",
    description: str | None = None,
) -> Tool[CodeExecutionParams, ToolUseCountMetadata]:
    """Create a code execution tool for this environment.

    Args:
        name: Tool name
        description: Tool description

    Returns:
        Tool[CodeExecutionParams] that executes commands in this environment

    """
    env = self

    async def executor(params: CodeExecutionParams) -> ToolResult[ToolUseCountMetadata]:
        result = await env.run_command(params.cmd)
        return format_result(result)

    return Tool[CodeExecutionParams, ToolUseCountMetadata](
        name=name,
        description=description
        or "Execute a shell command in the execution environment. Returns exit code, stdout, and stderr as XML.",
        parameters=CodeExecutionParams,
        executor=executor,  # ty: ignore[invalid-argument-type]
    )

get_view_image_tool

get_view_image_tool(
    *,
    name: str = "view_image",
    description: str | None = None,
) -> Tool[ViewImageParams, ToolUseCountMetadata]

Create a view_image tool for this environment.

Parameters:

Name	Type	Description	Default
name	str	Tool name	'view_image'
description	str | None	Tool description	None

Returns:

Type	Description
Tool[ViewImageParams, ToolUseCountMetadata]	Tool[ViewImageParams, ToolUseCountMetadata] that views images in this environment

Source code in src/stirrup/tools/code_backends/base.py

def get_view_image_tool(
    self,
    *,
    name: str = "view_image",
    description: str | None = None,
) -> Tool[ViewImageParams, ToolUseCountMetadata]:
    """Create a view_image tool for this environment.

    Args:
        name: Tool name
        description: Tool description

    Returns:
        Tool[ViewImageParams, ToolUseCountMetadata] that views images in this environment

    """
    env = self

    async def executor(params: ViewImageParams) -> ToolResult[ToolUseCountMetadata]:
        try:
            image = await env.view_image(params.path)
            return ToolResult(
                content=["Viewing image at path: " + params.path, image],
                metadata=ToolUseCountMetadata(),
            )
        except FileNotFoundError:
            return ToolResult(
                content=f"Image `{params.path}` not found.",
                success=False,
                metadata=ToolUseCountMetadata(),
            )
        except ValueError as e:
            return ToolResult(
                content=str(e),
                success=False,
                metadata=ToolUseCountMetadata(),
            )

    return Tool[ViewImageParams, ToolUseCountMetadata](
        name=name,
        description=description or "View an image file from the execution environment's filesystem.",
        parameters=ViewImageParams,
        executor=executor,  # ty: ignore[invalid-argument-type]
    )

view_image abstractmethod async

view_image(path: str) -> ImageContentBlock

Read and return an image file from the execution environment.

Parameters:

Name	Type	Description	Default
path	str	Path to image file in the execution environment (relative or absolute).	required

Returns:

Type	Description
ImageContentBlock	ImageContentBlock containing the image data.

Raises:

Type	Description
RuntimeError	If execution environment not started.
FileNotFoundError	If file does not exist.
ValueError	If path is outside the execution environment, is a directory, or the file is not a valid image.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def view_image(self, path: str) -> ImageContentBlock:
    """Read and return an image file from the execution environment.

    Args:
        path: Path to image file in the execution environment (relative or absolute).

    Returns:
        ImageContentBlock containing the image data.

    Raises:
        RuntimeError: If execution environment not started.
        FileNotFoundError: If file does not exist.
        ValueError: If path is outside the execution environment, is a directory,
                    or the file is not a valid image.

    """
    ...

LocalCodeExecToolProvider

LocalCodeExecToolProvider(
    *,
    allowed_commands: list[str] | None = None,
    temp_base_dir: Path | str | None = None,
    description: str | None = None,
)

Bases: CodeExecToolProvider

Local code execution tool provider using an isolated temp directory.

Commands are executed with the temp directory as the working directory. An optional allowlist can restrict which commands are permitted.

Usage with Agent

from stirrup.clients.chat_completions_client import ChatCompletionsClient

client = ChatCompletionsClient(model="gpt-5") agent = Agent( client=client, name="assistant", tools=[LocalCodeExecToolProvider(), CALCULATOR_TOOL], )

async with agent.session(output_dir="./output") as session: await session.run("Run some Python code")

Standalone usage

provider = LocalCodeExecToolProvider()

async with provider as tool: # tool is a Tool instance for code execution result = await provider.run_command("python script.py") await provider.save_output_files(["output.txt"], "/path/to/output")

Initialize LocalCodeExecToolProvider configuration.

Parameters:

Name	Type	Description	Default
allowed_commands	list[str] | None	Optional list of regex patterns. If provided, only commands matching at least one pattern are allowed. If None, all commands are allowed.	None
temp_base_dir	Path | str | None	Optional base directory for creating the execution environment temp directory. If None, uses the system default temp directory.	None
description	str | None	Optional description of the tool. If None, uses the default description.	None

Source code in src/stirrup/tools/code_backends/local.py

def __init__(
    self,
    *,
    allowed_commands: list[str] | None = None,
    temp_base_dir: Path | str | None = None,
    description: str | None = None,
) -> None:
    """Initialize LocalCodeExecToolProvider configuration.

    Args:
        allowed_commands: Optional list of regex patterns. If provided, only
                         commands matching at least one pattern are allowed.
                         If None, all commands are allowed.
        temp_base_dir: Optional base directory for creating the execution environment
                      temp directory. If None, uses the system default temp directory.
        description: Optional description of the tool. If None, uses the default description.

    """
    super().__init__(allowed_commands=allowed_commands)
    self._temp_dir: Path | None = None
    self._temp_base_dir: Path | None = Path(temp_base_dir) if temp_base_dir else None
    self._description = (
        description
        or "Execute a shell command in the execution environment. Returns exit code, stdout, and stderr as XML. Use `uv` to manage packages."
    )

temp_dir property

temp_dir: Path | None

Return the temp directory path, or None if not started.

__aenter__ async

__aenter__() -> Tool[
    CodeExecutionParams, ToolUseCountMetadata
]

Create temp directory and return the code_exec tool.

Source code in src/stirrup/tools/code_backends/local.py

async def __aenter__(self) -> "Tool[CodeExecutionParams, ToolUseCountMetadata]":
    """Create temp directory and return the code_exec tool."""
    if self._temp_base_dir:
        self._temp_base_dir.mkdir(parents=True, exist_ok=True)
    self._temp_dir = Path(tempfile.mkdtemp(prefix="local_exec_env_", dir=self._temp_base_dir))
    logger.debug("Created local execution environment temp directory: %s", self._temp_dir)
    return self.get_code_exec_tool(description=self._description)

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Cleanup the local execution environment.

Source code in src/stirrup/tools/code_backends/local.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Cleanup the local execution environment."""
    if self._temp_dir and self._temp_dir.exists():
        try:
            shutil.rmtree(self._temp_dir)
        except Exception as exc:
            logger.warning("Failed to cleanup temp directory %s: %s", self._temp_dir, exc)
    self._temp_dir = None

read_file_bytes async

read_file_bytes(path: str) -> bytes

Read file content as bytes from the temp directory.

Parameters:

Name	Type	Description	Default
path	str	File path (relative or absolute within the temp dir).	required

Returns:

Type	Description
bytes	File contents as bytes.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.
FileNotFoundError	If file does not exist.

Source code in src/stirrup/tools/code_backends/local.py

async def read_file_bytes(self, path: str) -> bytes:
    """Read file content as bytes from the temp directory.

    Args:
        path: File path (relative or absolute within the temp dir).

    Returns:
        File contents as bytes.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.
        FileNotFoundError: If file does not exist.

    """
    resolved = self._resolve_and_validate_path(path)
    if not resolved.exists():
        raise FileNotFoundError(f"File not found: {path}")
    return resolved.read_bytes()

write_file_bytes async

write_file_bytes(path: str, content: bytes) -> None

Write bytes to a file in the temp directory.

Parameters:

Name	Type	Description	Default
path	str	Destination path (relative or absolute within the temp dir).	required
content	bytes	File contents to write.	required

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def write_file_bytes(self, path: str, content: bytes) -> None:
    """Write bytes to a file in the temp directory.

    Args:
        path: Destination path (relative or absolute within the temp dir).
        content: File contents to write.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    resolved.parent.mkdir(parents=True, exist_ok=True)
    resolved.write_bytes(content)

file_exists async

file_exists(path: str) -> bool

Check if a file exists in the temp directory.

Parameters:

Name	Type	Description	Default
path	str	File path (relative or absolute within the temp dir).	required

Returns:

Type	Description
bool	True if the file exists, False otherwise.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def file_exists(self, path: str) -> bool:
    """Check if a file exists in the temp directory.

    Args:
        path: File path (relative or absolute within the temp dir).

    Returns:
        True if the file exists, False otherwise.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    return resolved.exists() and resolved.is_file()

is_directory async

is_directory(path: str) -> bool

Check if a path is a directory in the temp directory.

Parameters:

Name	Type	Description	Default
path	str	Path (relative or absolute within the temp dir).	required

Returns:

Type	Description
bool	True if the path exists and is a directory, False otherwise.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def is_directory(self, path: str) -> bool:
    """Check if a path is a directory in the temp directory.

    Args:
        path: Path (relative or absolute within the temp dir).

    Returns:
        True if the path exists and is a directory, False otherwise.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    return resolved.exists() and resolved.is_dir()

list_files async

list_files(path: str) -> list[str]

List all files recursively in a directory within the temp directory.

Parameters:

Name	Type	Description	Default
path	str	Directory path (relative or absolute within the temp dir).	required

Returns:

Type	Description
list[str]	List of file paths (relative to the given path) for all files in the directory.
list[str]	Returns an empty list if the path is a file or doesn't exist.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def list_files(self, path: str) -> list[str]:
    """List all files recursively in a directory within the temp directory.

    Args:
        path: Directory path (relative or absolute within the temp dir).

    Returns:
        List of file paths (relative to the given path) for all files in the directory.
        Returns an empty list if the path is a file or doesn't exist.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    if not resolved.exists() or not resolved.is_dir():
        return []

    files = []
    for file_path in resolved.rglob("*"):
        if file_path.is_file():
            rel_path = file_path.relative_to(resolved)
            files.append(str(rel_path))
    return files

run_command async

run_command(
    cmd: str, *, timeout: int = SHELL_TIMEOUT
) -> CommandResult

Execute command in the temp directory.

Parameters:

Name	Type	Description	Default
cmd	str	Shell command to execute (bash syntax).	required
timeout	int	Maximum time in seconds to wait for command completion.	SHELL_TIMEOUT

Returns:

Type	Description
CommandResult	CommandResult with exit_code, stdout, stderr, and optional error info.

Source code in src/stirrup/tools/code_backends/local.py

async def run_command(self, cmd: str, *, timeout: int = SHELL_TIMEOUT) -> CommandResult:
    """Execute command in the temp directory.

    Args:
        cmd: Shell command to execute (bash syntax).
        timeout: Maximum time in seconds to wait for command completion.

    Returns:
        CommandResult with exit_code, stdout, stderr, and optional error info.

    """
    if self._temp_dir is None:
        raise RuntimeError(
            "ExecutionEnvironment not started. Ensure current Agent is equipped with a CodeExecToolProvider."
        )

    # Check allowlist
    if not self._check_allowed(cmd):
        return CommandResult(
            exit_code=1,
            stdout="",
            stderr=f"Command not allowed: '{cmd}' does not match any allowed patterns",
            error_kind="command_not_allowed",
            advice="Only commands matching the allowlist patterns are permitted.",
        )

    # Check for absolute paths (local environment is not sandboxed)
    absolute_path_error = self._check_absolute_paths(cmd)
    if absolute_path_error:
        return absolute_path_error

    process = None
    try:
        with anyio.fail_after(timeout):
            # Use shell=True by wrapping in a shell command
            process = await anyio.open_process(
                ["bash", "-c", cmd],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                cwd=self._temp_dir,
            )

            # Read all output from streams concurrently
            stdout_chunks: list[bytes] = []
            stderr_chunks: list[bytes] = []

            async def read_stdout() -> None:
                if process.stdout:
                    stdout_chunks.extend([chunk async for chunk in process.stdout])

            async def read_stderr() -> None:
                if process.stderr:
                    stderr_chunks.extend([chunk async for chunk in process.stderr])

            async with anyio.create_task_group() as tg:
                tg.start_soon(read_stdout)
                tg.start_soon(read_stderr)

            await process.wait()

            return CommandResult(
                exit_code=process.returncode or 0,
                stdout=b"".join(stdout_chunks).decode("utf-8", errors="replace"),
                stderr=b"".join(stderr_chunks).decode("utf-8", errors="replace"),
            )

    except TimeoutError:
        if process:
            process.kill()
        return CommandResult(
            exit_code=1,
            stdout="",
            stderr=f"Command timed out after {timeout} seconds",
            error_kind="timeout",
        )
    except Exception as exc:
        return CommandResult(
            exit_code=1,
            stdout="",
            stderr=str(exc),
            error_kind="execution_error",
        )

save_output_files async

save_output_files(
    paths: list[str],
    output_dir: Path | str,
    dest_env: CodeExecToolProvider | None = None,
) -> SaveOutputFilesResult

Move files from the temp directory to a destination.

When dest_env is None (local filesystem), files are MOVED (not copied) - originals are deleted from the execution environment. Existing files in output_dir are silently overwritten.

When dest_env is provided (cross-environment transfer), files are copied using the base class implementation via read/write primitives.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths in the execution environment (relative or absolute). Relative paths are resolved against the execution environment temp directory.	required
output_dir	Path | str	Directory path to save files to.	required
dest_env	CodeExecToolProvider | None	If provided, output_dir is interpreted as a path within dest_env (cross-environment transfer). If None, output_dir is a local filesystem path.	None

Returns:

Type	Description
SaveOutputFilesResult	SaveOutputFilesResult containing lists of saved files and any failures.

Source code in src/stirrup/tools/code_backends/local.py

async def save_output_files(
    self,
    paths: list[str],
    output_dir: Path | str,
    dest_env: "CodeExecToolProvider | None" = None,
) -> SaveOutputFilesResult:
    """Move files from the temp directory to a destination.

    When dest_env is None (local filesystem), files are MOVED (not copied) -
    originals are deleted from the execution environment.
    Existing files in output_dir are silently overwritten.

    When dest_env is provided (cross-environment transfer), files are copied
    using the base class implementation via read/write primitives.

    Args:
        paths: List of file paths in the execution environment (relative or absolute).
               Relative paths are resolved against the execution environment temp directory.
        output_dir: Directory path to save files to.
        dest_env: If provided, output_dir is interpreted as a path within dest_env
                  (cross-environment transfer). If None, output_dir is a local
                  filesystem path.

    Returns:
        SaveOutputFilesResult containing lists of saved files and any failures.

    """
    if self._temp_dir is None:
        raise RuntimeError(
            "ExecutionEnvironment not started. Ensure current Agent is equipped with a CodeExecToolProvider."
        )

    # If dest_env is provided, use the base class implementation (cross-env transfer)
    if dest_env is not None:
        return await super().save_output_files(paths, output_dir, dest_env)

    # Local filesystem - use optimized move operation
    output_dir_path = Path(output_dir)
    output_dir_path.mkdir(parents=True, exist_ok=True)

    result = SaveOutputFilesResult()

    for source_path_str in paths:
        try:
            source_path = Path(source_path_str)
            if not source_path.is_absolute():
                source_path = self._temp_dir / source_path

            # Security: ensure path is within temp directory
            try:
                source_path.resolve().relative_to(self._temp_dir.resolve())
            except ValueError:
                result.failed[source_path_str] = "Path is outside execution environment directory"
                logger.warning("Attempted to access path outside execution environment: %s", source_path_str)
                continue

            if not source_path.exists():
                result.failed[source_path_str] = "File does not exist"
                logger.warning("Execution environment file does not exist: %s", source_path_str)
                continue

            if not source_path.is_file():
                result.failed[source_path_str] = "Path is not a file"
                logger.warning("Execution environment path is not a file: %s", source_path_str)
                continue

            file_size = source_path.stat().st_size
            dest_path = output_dir_path / source_path.name

            # Move file (overwrites if exists)
            shutil.move(str(source_path), str(dest_path))
            logger.debug("Moved file: %s -> %s", source_path, dest_path)

            result.saved.append(
                SavedFile(
                    source_path=source_path_str,
                    output_path=dest_path,
                    size=file_size,
                ),
            )

        except Exception as exc:
            result.failed[source_path_str] = str(exc)
            logger.exception("Failed to move file: %s", source_path_str)

    return result

upload_files async

upload_files(
    *paths: Path | str,
    source_env: CodeExecToolProvider | None = None,
    dest_dir: str | None = None,
) -> UploadFilesResult

Upload files to the execution environment.

When source_env is None (local filesystem), files are COPIED (not moved) - originals remain on the local filesystem. Directories are uploaded recursively, preserving their structure.

When source_env is provided (cross-environment transfer), files are copied using the base class implementation via read/write primitives.

Parameters:

Name	Type	Description	Default
*paths	Path | str	File or directory paths to upload. If source_env is None, these are local filesystem paths. If source_env is provided, these are paths within source_env.	()
source_env	CodeExecToolProvider | None	If provided, paths are within source_env. If None, paths are local filesystem paths.	None
dest_dir	str | None	Destination subdirectory within the temp directory. If None, files are placed directly in the temp directory.	None

Returns:

Type	Description
UploadFilesResult	UploadFilesResult containing lists of uploaded files and any failures.

Source code in src/stirrup/tools/code_backends/local.py

async def upload_files(
    self,
    *paths: Path | str,
    source_env: "CodeExecToolProvider | None" = None,
    dest_dir: str | None = None,
) -> UploadFilesResult:
    """Upload files to the execution environment.

    When source_env is None (local filesystem), files are COPIED (not moved) -
    originals remain on the local filesystem.
    Directories are uploaded recursively, preserving their structure.

    When source_env is provided (cross-environment transfer), files are copied
    using the base class implementation via read/write primitives.

    Args:
        *paths: File or directory paths to upload. If source_env is None, these
                are local filesystem paths. If source_env is provided, these are
                paths within source_env.
        source_env: If provided, paths are within source_env. If None, paths are
                    local filesystem paths.
        dest_dir: Destination subdirectory within the temp directory.
                  If None, files are placed directly in the temp directory.

    Returns:
        UploadFilesResult containing lists of uploaded files and any failures.

    """
    if self._temp_dir is None:
        raise RuntimeError(
            "ExecutionEnvironment not started. Ensure current Agent is equipped with a CodeExecToolProvider."
        )

    # If source_env is provided, use the base class implementation (cross-env transfer)
    if source_env is not None:
        return await super().upload_files(*paths, source_env=source_env, dest_dir=dest_dir)

    # Local filesystem - use optimized copy operation
    dest_base = self._temp_dir / dest_dir if dest_dir else self._temp_dir
    dest_base.mkdir(parents=True, exist_ok=True)

    result = UploadFilesResult()

    for source in paths:
        source = Path(source).resolve()

        if not source.exists():
            result.failed[str(source)] = "File or directory does not exist"
            logger.warning("Upload source does not exist: %s", source)
            continue

        try:
            if source.is_file():
                dest = dest_base / source.name
                shutil.copy2(source, dest)
                result.uploaded.append(
                    UploadedFile(
                        source_path=source,
                        dest_path=str(dest.relative_to(self._temp_dir)),
                        size=source.stat().st_size,
                    ),
                )
                logger.debug("Uploaded file: %s -> %s", source, dest)

            elif source.is_dir():
                # If dest_dir was explicitly provided, copy contents directly to dest_base
                # Otherwise, create a subdirectory with the source's name
                if dest_dir:
                    dest = dest_base
                    # Copy contents of source directory into dest_base
                    for item in source.iterdir():
                        item_dest = dest / item.name
                        if item.is_file():
                            shutil.copy2(item, item_dest)
                        else:
                            shutil.copytree(item, item_dest, dirs_exist_ok=True)
                else:
                    dest = dest_base / source.name
                    shutil.copytree(source, dest, dirs_exist_ok=True)
                # Track all individual files uploaded
                for file_path in source.rglob("*"):
                    if file_path.is_file():
                        relative = file_path.relative_to(source)
                        dest_file = dest / relative
                        result.uploaded.append(
                            UploadedFile(
                                source_path=file_path,
                                dest_path=str(dest_file.relative_to(self._temp_dir)),
                                size=file_path.stat().st_size,
                            ),
                        )
                logger.debug("Uploaded directory: %s -> %s", source, dest)

        except Exception as exc:
            result.failed[str(source)] = str(exc)
            logger.exception("Failed to upload: %s", source)

    return result

view_image async

view_image(path: str) -> ImageContentBlock

Read and return an image file from the local execution environment.

Parameters:

Name	Type	Description	Default
path	str	Path to image file (relative to temp directory, or absolute within it).	required

Returns:

Type	Description
ImageContentBlock	ImageContentBlock containing the image data.

Raises:

Type	Description
RuntimeError	If execution environment not started.
FileNotFoundError	If file does not exist.
ValueError	If path is outside temp directory, is a directory, or not a valid image.

Source code in src/stirrup/tools/code_backends/local.py

async def view_image(self, path: str) -> ImageContentBlock:
    """Read and return an image file from the local execution environment.

    Args:
        path: Path to image file (relative to temp directory, or absolute within it).

    Returns:
        ImageContentBlock containing the image data.

    Raises:
        RuntimeError: If execution environment not started.
        FileNotFoundError: If file does not exist.
        ValueError: If path is outside temp directory, is a directory, or not a valid image.

    """
    file_bytes = await self.read_file_bytes(path)
    return ImageContentBlock(data=file_bytes)

AgentLogger

AgentLogger(
    *, show_spinner: bool = True, level: int = INFO
)

Bases: AgentLoggerBase

Rich console logger for agent workflows.

Implements AgentLoggerBase with rich formatting, spinners, and visual hierarchy. Each agent (including sub-agents) should have its own logger instance.

Usage

from stirrup.clients.chat_completions_client import ChatCompletionsClient

Agent creates logger internally by default

client = ChatCompletionsClient(model="gpt-4") agent = Agent(client=client, name="assistant")

Or pass a pre-configured logger

logger = AgentLogger(show_spinner=False) agent = Agent(client=client, name="assistant", logger=logger)

Agent sets these properties before calling enter:

logger.name, logger.model, logger.max_turns, logger.depth

Agent sets these before calling exit:

logger.finish_params, logger.run_metadata, logger.output_dir

Initialize the agent logger.

Parameters:

Name	Type	Description	Default
show_spinner	bool	Whether to show a spinner while agent runs (only for depth=0)	True
level	int	Logging level (default: INFO)	INFO

Source code in src/stirrup/utils/logging.py

def __init__(
    self,
    *,
    show_spinner: bool = True,
    level: int = logging.INFO,
) -> None:
    """Initialize the agent logger.

    Args:
        show_spinner: Whether to show a spinner while agent runs (only for depth=0)
        level: Logging level (default: INFO)
    """
    # Properties set by Agent before __enter__
    self.name: str = "agent"
    self.model: str | None = None
    self.max_turns: int | None = None
    self.depth: int = 0

    # State set by Agent before __exit__
    self.finish_params: BaseModel | None = None
    self.run_metadata: dict[str, list[Any]] | None = None
    self.output_dir: str | None = None

    # Configuration
    self._show_spinner = show_spinner
    self._level = level

    # Spinner state (only used when depth == 0 and show_spinner is True)
    self._current_step = 0
    self._tool_calls = 0
    self._input_tokens = 0
    self._output_tokens = 0
    self._live: Live | None = None

    # Configure rich logging on first logger creation
    self._configure_logging()

__enter__

__enter__() -> Self

Enter logging context. Logs agent start and starts spinner if depth=0.

Source code in src/stirrup/utils/logging.py

def __enter__(self) -> Self:
    """Enter logging context. Logs agent start and starts spinner if depth=0."""
    # Log agent start (rule + system prompt display)
    indent_spaces = self.depth * SUBAGENT_INDENT_SPACES

    # Build title with optional model info
    model_str = f" ({self.model})" if self.model else ""
    if self.depth == 0:
        title = f"▶ {self.name}{model_str}"
        console.rule(f"[bold cyan]{title}[/]", style="cyan")
    else:
        title = f"▶ {self.name}: Level {self.depth}{model_str}"
        rule = Rule(f"[bold cyan]{title}[/]", style="cyan")
        self._print_indented(rule, indent_spaces)
    console.print()

    # Start spinner only for top-level agent
    if self.depth == 0 and self._show_spinner:
        self._live = Live(self._make_spinner(), console=console, refresh_per_second=10)
        self._live.start()

    return self

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Exit logging context. Stops spinner and logs completion stats.

Source code in src/stirrup/utils/logging.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Exit logging context. Stops spinner and logs completion stats."""
    # Stop spinner first
    if self._live:
        self._live.stop()
        self._live = None

    error = str(exc_val) if exc_type is not None else None
    self._log_finish(error=error)

on_step

on_step(
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None

Report step progress and stats during agent execution.

Source code in src/stirrup/utils/logging.py

def on_step(
    self,
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None:
    """Report step progress and stats during agent execution."""
    self._current_step = step
    self._tool_calls = tool_calls
    self._input_tokens = input_tokens
    self._output_tokens = output_tokens
    if self._live:
        self._live.update(self._make_spinner())

pause_live

pause_live() -> None

Pause the live spinner display.

Call this before prompting for user input to prevent the spinner from interfering with the input prompt.

Source code in src/stirrup/utils/logging.py

def pause_live(self) -> None:
    """Pause the live spinner display.

    Call this before prompting for user input to prevent the spinner
    from interfering with the input prompt.
    """
    if self._live is not None:
        self._live.stop()

resume_live

resume_live() -> None

Resume the live spinner display.

Call this after user input is complete to restart the spinner.

Source code in src/stirrup/utils/logging.py

def resume_live(self) -> None:
    """Resume the live spinner display.

    Call this after user input is complete to restart the spinner.
    """
    if self._live is not None:
        self._live.start()

set_level

set_level(level: int) -> None

Set the logging level.

Source code in src/stirrup/utils/logging.py

def set_level(self, level: int) -> None:
    """Set the logging level."""
    self._level = level
    # Also update root logger level
    logging.getLogger().setLevel(level)

is_enabled_for

is_enabled_for(level: int) -> bool

Check if a given log level is enabled.

Source code in src/stirrup/utils/logging.py

def is_enabled_for(self, level: int) -> bool:
    """Check if a given log level is enabled."""
    return level >= self._level

debug

debug(message: str, *args: object) -> None

Log a debug message (dim style).

Source code in src/stirrup/utils/logging.py

def debug(self, message: str, *args: object) -> None:
    """Log a debug message (dim style)."""
    if self._level <= logging.DEBUG:
        formatted = message % args if args else message
        console.print(f"[dim]{formatted}[/]")

info

info(message: str, *args: object) -> None

Log an info message.

Source code in src/stirrup/utils/logging.py

def info(self, message: str, *args: object) -> None:
    """Log an info message."""
    if self._level <= logging.INFO:
        formatted = message % args if args else message
        console.print(formatted)

warning

warning(message: str, *args: object) -> None

Log a warning message (yellow style).

Source code in src/stirrup/utils/logging.py

def warning(self, message: str, *args: object) -> None:
    """Log a warning message (yellow style)."""
    if self._level <= logging.WARNING:
        formatted = message % args if args else message
        console.print(f"[yellow]⚠ {formatted}[/]")

error

error(message: str, *args: object) -> None

Log an error message (red style).

Source code in src/stirrup/utils/logging.py

def error(self, message: str, *args: object) -> None:
    """Log an error message (red style)."""
    if self._level <= logging.ERROR:
        formatted = message % args if args else message
        console.print(f"[red]✗ {formatted}[/]")

critical

critical(message: str, *args: object) -> None

Log a critical message (bold red style).

Source code in src/stirrup/utils/logging.py

def critical(self, message: str, *args: object) -> None:
    """Log a critical message (bold red style)."""
    if self._level <= logging.CRITICAL:
        formatted = message % args if args else message
        console.print(f"[bold red]✗ CRITICAL: {formatted}[/]")

exception

exception(message: str, *args: object) -> None

Log an error message with exception traceback (red style with traceback).

Source code in src/stirrup/utils/logging.py

def exception(self, message: str, *args: object) -> None:
    """Log an error message with exception traceback (red style with traceback)."""
    if self._level <= logging.ERROR:
        formatted = message % args if args else message
        console.print(f"[red]✗ {formatted}[/]")
        console.print_exception()

assistant_message

assistant_message(
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None

Log an assistant message with content and tool calls in a panel.

Parameters:

Name	Type	Description	Default
turn	int	Current turn number (1-indexed)	required
max_turns	int	Maximum number of turns	required
assistant_message	AssistantMessage	The assistant's response message	required

Source code in src/stirrup/utils/logging.py

def assistant_message(
    self,
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None:
    """Log an assistant message with content and tool calls in a panel.

    Args:
        turn: Current turn number (1-indexed)
        max_turns: Maximum number of turns
        assistant_message: The assistant's response message
    """
    if self._level > logging.INFO:
        return

    # Build panel content
    content = Text()

    # Add assistant content if present
    if assistant_message.content:
        text = assistant_message.content
        if isinstance(text, list):
            text = "\n".join(str(block) for block in text)
        # Truncate long content
        if len(text) > 500:
            text = text[:500] + "..."
        content.append(text, style="white")

    # Add tool calls if present
    if assistant_message.tool_calls:
        if assistant_message.content:
            content.append("\n\n")
        content.append("Tool Calls:\n", style="bold magenta")
        for tc in assistant_message.tool_calls:
            content.append(f"  🔧 {tc.name}", style="magenta")
            if tc.arguments and tc.arguments.strip():
                args_parsed = json.loads(tc.arguments)
                args_formatted = json.dumps(args_parsed, indent=2, ensure_ascii=False)
                args_preview = args_formatted[:1000] + "..." if len(args_formatted) > 1000 else args_formatted
                content.append(args_preview, style="dim")

    # Create and print panel with agent name in title
    title = f"[bold]AssistantMessage[/bold] │ {self.name} │ Turn {turn}/{max_turns}"
    panel = Panel(content, title=title, title_align="left", border_style="yellow", padding=(0, 1))

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

user_message

user_message(user_message: UserMessage) -> None

Log a user message in a panel.

Parameters:

Name	Type	Description	Default
user_message	UserMessage	The user's message	required

Source code in src/stirrup/utils/logging.py

def user_message(self, user_message: UserMessage) -> None:
    """Log a user message in a panel.

    Args:
        user_message: The user's message
    """
    if self._level > logging.INFO:
        return

    # Build panel content
    content = Text()

    # Add user content
    if user_message.content:
        text = user_message.content
        if isinstance(text, list):
            text = "\n".join(str(block) for block in text)
        # Truncate long content
        if len(text) > 500:
            text = text[:500] + "..."
        content.append(text, style="white")

    # Create and print panel with agent name in title
    title = f"[bold]UserMessage[/bold] │ {self.name}"
    panel = Panel(content, title=title, title_align="left", border_style="blue", padding=(0, 1))

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

task_message

task_message(task: str | list[Any]) -> None

Log the initial task/prompt at the start of a run.

Source code in src/stirrup/utils/logging.py

def task_message(self, task: str | list[Any]) -> None:
    """Log the initial task/prompt at the start of a run."""
    if self._level > logging.INFO:
        return

    # Convert list content to string
    if isinstance(task, list):
        task = "\n".join(str(block) for block in task)

    # Clean up whitespace from multi-line strings
    # Normalize each line by stripping leading/trailing whitespace and rejoining
    lines = [line.strip() for line in task.split("\n")]
    task = " ".join(line for line in lines if line)

    # Use "Sub Agent" prefix for nested agents
    prefix = "Sub Agent" if self.depth > 0 else "Agent"

    if self.depth > 0:
        indent = " " * (self.depth * SUBAGENT_INDENT_SPACES)
        console.print(f"{indent}[bold]{prefix} Task:[/bold]")
        console.print()
        for line in task.split("\n"):
            console.print(f"{indent}{line}")
    else:
        console.print(f"[bold]{prefix} Task:[/bold]")
        console.print()
        console.print(task)

    console.print()  # Add gap after task section

warnings_message

warnings_message(warnings: list[str]) -> None

Display warnings at run start as simple text.

Source code in src/stirrup/utils/logging.py

def warnings_message(self, warnings: list[str]) -> None:
    """Display warnings at run start as simple text."""
    if self._level > logging.INFO or not warnings:
        return

    console.print("[bold orange1]Warnings[/bold orange1]")
    console.print()
    for warning in warnings:
        console.print(f"[orange1]⚠ {warning}[/orange1]")
        console.print()  # Add gap between warnings

tool_result

tool_result(tool_message: ToolMessage) -> None

Log a single tool execution result in a panel with XML syntax highlighting.

Parameters:

Name	Type	Description	Default
tool_message	ToolMessage	The tool execution result	required

Source code in src/stirrup/utils/logging.py

def tool_result(self, tool_message: ToolMessage) -> None:
    """Log a single tool execution result in a panel with XML syntax highlighting.

    Args:
        tool_message: The tool execution result
    """
    if self._level > logging.INFO:
        return

    tool_name = tool_message.name or "unknown"

    # Get result content
    result_text = tool_message.content
    if isinstance(result_text, list):
        result_text = "\n".join(str(block) for block in result_text)

    # Unescape HTML entities (e.g., &lt; -> <, &gt; -> >, &amp; -> &)
    result_text = html.unescape(result_text)

    # Truncate long results (keeps start and end, removes middle)
    result_text = truncate_msg(result_text, 1000)

    # Format as XML with syntax highlighting
    content = Syntax(result_text, "xml", theme="monokai", word_wrap=True)

    # Status indicator in title with agent name
    status = "✓" if tool_message.args_was_valid else "✗"
    status_style = "green" if tool_message.args_was_valid else "red"
    title = f"[{status_style}]{status}[/{status_style}] [bold]ToolResult[/bold] │ {self.name} │ [green]{tool_name}[/green]"

    panel = Panel(content, title=title, title_align="left", border_style="green", padding=(0, 1))

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

context_summarization_start

context_summarization_start(
    pct_used: float, cutoff: float
) -> None

Log context window summarization starting in an orange panel.

Parameters:

Name	Type	Description	Default
pct_used	float	Percentage of context window currently used (0.0-1.0)	required
cutoff	float	The threshold that triggered summarization (0.0-1.0)	required

Source code in src/stirrup/utils/logging.py

def context_summarization_start(self, pct_used: float, cutoff: float) -> None:
    """Log context window summarization starting in an orange panel.

    Args:
        pct_used: Percentage of context window currently used (0.0-1.0)
        cutoff: The threshold that triggered summarization (0.0-1.0)
    """
    # Build panel content
    content = Text()
    content.append("Context window limit reached\n\n", style="bold")
    content.append("Used: ", style="dim")
    content.append(f"{pct_used:.1%}", style="bold orange1")
    content.append("  │  ", style="dim")
    content.append("Threshold: ", style="dim")
    content.append(f"{cutoff:.1%}", style="bold")
    content.append("\n\n", style="dim")
    content.append("Summarizing conversation history...", style="italic")

    panel = Panel(
        content,
        title="[bold orange1]📝 Context Summarization[/]",
        title_align="left",
        border_style="orange1",
        padding=(0, 1),
    )

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

context_summarization_complete

context_summarization_complete(
    summary: str, bridge: str
) -> None

Log the completed context summarization with summary content.

Parameters:

Name	Type	Description	Default
summary	str	The generated summary of the conversation	required
bridge	str	The bridge message that will be used to continue the conversation	required

Source code in src/stirrup/utils/logging.py

def context_summarization_complete(self, summary: str, bridge: str) -> None:
    """Log the completed context summarization with summary content.

    Args:
        summary: The generated summary of the conversation
        bridge: The bridge message that will be used to continue the conversation
    """
    # Truncate long summaries for display
    summary_display = summary
    if len(summary_display) > 800:
        summary_display = summary_display[:800] + "..."

    # Build panel content
    content = Text()
    content.append("Summary:\n", style="bold")
    content.append(summary_display, style="white")

    if self._level > logging.INFO:
        bridge_display = bridge
        if len(bridge_display) > 200:
            bridge_display = bridge_display[:200] + "..."
        content.append("\n\n")
        content.append("Bridge Message:\n", style="bold dim")
        content.append(bridge_display, style="dim italic")

    panel = Panel(
        content,
        title="[bold green]✓ Summary Generated[/]",
        title_align="left",
        border_style="green",
        padding=(0, 1),
    )

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

AgentLoggerBase

Bases: ABC

Abstract base class for agent loggers.

Defines the interface that Agent uses for logging. Implement this to create custom loggers (e.g., for testing, file output, or monitoring services).

Properties are set by Agent after construction: - name, model, max_turns, depth: Agent configuration - finish_params, run_metadata, output_dir: Set before exit for final stats

__enter__ abstractmethod

__enter__() -> Self

Enter logging context. Called when agent session starts.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def __enter__(self) -> Self:
    """Enter logging context. Called when agent session starts."""
    ...

__exit__ abstractmethod

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Exit logging context. Called when agent session ends.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Exit logging context. Called when agent session ends."""
    ...

on_step abstractmethod

on_step(
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None

Report step progress and stats during agent execution.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def on_step(
    self,
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None:
    """Report step progress and stats during agent execution."""
    ...

assistant_message abstractmethod

assistant_message(
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None

Log an assistant message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def assistant_message(
    self,
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None:
    """Log an assistant message."""
    ...

user_message abstractmethod

user_message(user_message: UserMessage) -> None

Log a user message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def user_message(self, user_message: UserMessage) -> None:
    """Log a user message."""
    ...

task_message abstractmethod

task_message(task: str | list[Any]) -> None

Log the initial task/prompt at the start of a run.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def task_message(self, task: str | list[Any]) -> None:
    """Log the initial task/prompt at the start of a run."""
    ...

tool_result abstractmethod

tool_result(tool_message: ToolMessage) -> None

Log a tool execution result.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def tool_result(self, tool_message: ToolMessage) -> None:
    """Log a tool execution result."""
    ...

context_summarization_start abstractmethod

context_summarization_start(
    pct_used: float, cutoff: float
) -> None

Log that context summarization is starting.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def context_summarization_start(self, pct_used: float, cutoff: float) -> None:
    """Log that context summarization is starting."""
    ...

context_summarization_complete abstractmethod

context_summarization_complete(
    summary: str, bridge: str
) -> None

Log completed context summarization.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def context_summarization_complete(self, summary: str, bridge: str) -> None:
    """Log completed context summarization."""
    ...

debug abstractmethod

debug(message: str, *args: object) -> None

Log a debug message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def debug(self, message: str, *args: object) -> None:
    """Log a debug message."""
    ...

info abstractmethod

info(message: str, *args: object) -> None

Log an info message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def info(self, message: str, *args: object) -> None:
    """Log an info message."""
    ...

warning abstractmethod

warning(message: str, *args: object) -> None

Log a warning message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def warning(self, message: str, *args: object) -> None:
    """Log a warning message."""
    ...

error abstractmethod

error(message: str, *args: object) -> None

Log an error message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def error(self, message: str, *args: object) -> None:
    """Log an error message."""
    ...

pause_live

pause_live() -> None

Pause live display (e.g., spinner) before user interaction.

Source code in src/stirrup/utils/logging.py

def pause_live(self) -> None:  # noqa: B027
    """Pause live display (e.g., spinner) before user interaction."""

resume_live

resume_live() -> None

Resume live display after user interaction.

Source code in src/stirrup/utils/logging.py

def resume_live(self) -> None:  # noqa: B027
    """Resume live display after user interaction."""

SessionState dataclass

SessionState(
    exit_stack: AsyncExitStack,
    exec_env: CodeExecToolProvider | None = None,
    output_dir: str | None = None,
    parent_exec_env: CodeExecToolProvider | None = None,
    depth: int = 0,
    uploaded_file_paths: list[str] = list(),
    skills_metadata: list[SkillMetadata] = list(),
    logger: AgentLoggerBase | None = None,
)

Per-session state for resource lifecycle management.

Kept minimal - only contains resources that need async lifecycle management (exit_stack, exec_env) and session-specific configuration (output_dir).

Tool availability is managed via Agent._active_tools (instance-scoped), and run results are stored on the agent instance temporarily.

For subagent file transfer: - parent_exec_env: Reference to the parent's exec env (for cross-env transfers) - depth: Agent depth (0 = root, >0 = subagent) - output_dir: For root agent, this is a local filesystem path. For subagents, this is a path within the parent's exec env.

SubAgentParams

Bases: BaseModel

Parameters for sub-agent tool invocation.

Agent

Agent(
    client: LLMClient,
    name: str,
    *,
    max_turns: int = AGENT_MAX_TURNS,
    system_prompt: str | None = None,
    tools: list[Tool | ToolProvider] | None = None,
    finish_tool: Tool[FinishParams, FinishMeta]
    | None = None,
    context_summarization_cutoff: float = CONTEXT_SUMMARIZATION_CUTOFF,
    turns_remaining_warning_threshold: int = TURNS_REMAINING_WARNING_THRESHOLD,
    run_sync_in_thread: bool = True,
    text_only_tool_responses: bool = True,
    logger: AgentLoggerBase | None = None,
)

Agent that executes tool-using loops with automatic context management.

Runs up to max_turns iterations of: LLM generation → tool execution → message accumulation. When conversation history exceeds context window limits, older messages are automatically condensed into a summary to preserve working memory.

The Agent can be used as an async context manager via .session() for automatic tool lifecycle management, logging, and file saving:

from stirrup.clients.chat_completions_client import ChatCompletionsClient

# Create client and agent
client = ChatCompletionsClient(model="gpt-5")
agent = Agent(client=client, name="assistant")

async with agent.session(output_dir="./output") as session:
    finish_params, history, metadata = await session.run("Your task here")

Initialize the agent with an LLM client and configuration.

Parameters:

Name	Type	Description	Default
client	LLMClient	LLM client for generating responses. Use ChatCompletionsClient for OpenAI/OpenAI-compatible APIs, or LiteLLMClient for other providers.	required
name	str	Name of the agent (used for logging purposes)	required
max_turns	int	Maximum number of turns before stopping	AGENT_MAX_TURNS
system_prompt	str | None	System prompt to prepend to all runs (when using string prompts)	None
tools	list[Tool | ToolProvider] | None	List of Tools and/or ToolProviders available to the agent. If None, uses DEFAULT_TOOLS. ToolProviders are automatically set up and torn down by Agent.session(). Use [*DEFAULT_TOOLS, extra_tool] to extend defaults.	None
finish_tool	Tool[FinishParams, FinishMeta] | None	Tool used to signal task completion. Defaults to SIMPLE_FINISH_TOOL.	None
context_summarization_cutoff	float	Fraction of context window (0-1) at which to trigger summarization	CONTEXT_SUMMARIZATION_CUTOFF
run_sync_in_thread	bool	Execute synchronous tool executors in a separate thread	True
text_only_tool_responses	bool	Extract images from tool responses as separate user messages	True
logger	AgentLoggerBase | None	Optional logger instance. If None, creates AgentLogger() internally.	None

Source code in src/stirrup/core/agent.py

def __init__(
    self,
    client: LLMClient,
    name: str,
    *,
    max_turns: int = AGENT_MAX_TURNS,
    system_prompt: str | None = None,
    tools: list[Tool | ToolProvider] | None = None,
    finish_tool: Tool[FinishParams, FinishMeta] | None = None,
    # Agent options
    context_summarization_cutoff: float = CONTEXT_SUMMARIZATION_CUTOFF,
    turns_remaining_warning_threshold: int = TURNS_REMAINING_WARNING_THRESHOLD,
    run_sync_in_thread: bool = True,
    text_only_tool_responses: bool = True,
    # Logging
    logger: AgentLoggerBase | None = None,
) -> None:
    """Initialize the agent with an LLM client and configuration.

    Args:
        client: LLM client for generating responses. Use ChatCompletionsClient for
                OpenAI/OpenAI-compatible APIs, or LiteLLMClient for other providers.
        name: Name of the agent (used for logging purposes)
        max_turns: Maximum number of turns before stopping
        system_prompt: System prompt to prepend to all runs (when using string prompts)
        tools: List of Tools and/or ToolProviders available to the agent.
               If None, uses DEFAULT_TOOLS. ToolProviders are automatically
               set up and torn down by Agent.session().
               Use [*DEFAULT_TOOLS, extra_tool] to extend defaults.
        finish_tool: Tool used to signal task completion. Defaults to SIMPLE_FINISH_TOOL.
        context_summarization_cutoff: Fraction of context window (0-1) at which to trigger summarization
        run_sync_in_thread: Execute synchronous tool executors in a separate thread
        text_only_tool_responses: Extract images from tool responses as separate user messages
        logger: Optional logger instance. If None, creates AgentLogger() internally.

    """
    # Validate agent name
    if not AGENT_NAME_PATTERN.match(name):
        raise ValueError(
            f"Invalid agent name '{name}'. "
            "Agent names must match pattern '^[a-zA-Z0-9_-]{1,128}$' "
            "(alphanumeric, underscores, hyphens only, 1-128 characters)."
        )

    self._client: LLMClient = client
    self._name = name
    self._max_turns = max_turns
    self._system_prompt = system_prompt
    self._tools = tools if tools is not None else DEFAULT_TOOLS
    self._finish_tool: Tool = finish_tool if finish_tool is not None else SIMPLE_FINISH_TOOL
    self._context_summarization_cutoff = context_summarization_cutoff
    self._turns_remaining_warning_threshold = turns_remaining_warning_threshold
    self._run_sync_in_thread = run_sync_in_thread
    self._text_only_tool_responses = text_only_tool_responses

    # Logger (can be passed in or created here)
    self._logger: AgentLoggerBase = logger if logger is not None else AgentLogger()

    # Session configuration (set during session(), used in __aenter__)
    self._pending_output_dir: Path | None = None
    self._pending_input_files: str | Path | list[str | Path] | None = None
    self._pending_skills_dir: Path | None = None
    self._resume: bool = False
    self._clear_cache_on_success: bool = True
    self._cache_on_interrupt: bool = True

    # Instance-scoped state (populated during __aenter__, isolated per agent instance)
    self._active_tools: dict[str, Tool] = {}
    self._last_finish_params: Any = None  # FinishParams type parameter
    self._last_run_metadata: dict[str, list[Any]] = {}
    self._transferred_paths: list[str] = []  # Paths transferred to parent (for subagents)

    # Cache state for resumption (set during run(), used in __aexit__ for caching on interrupt)
    self._current_task_hash: str | None = None
    self._current_run_state: CacheState | None = None

name property

name: str

The name of this agent.

client property

client: LLMClient

The LLM client used by this agent.

tools property

tools: dict[str, Tool]

Currently active tools (available after entering session context).

finish_tool property

finish_tool: Tool

The finish tool used to signal task completion.

logger property

logger: AgentLoggerBase

The logger instance used by this agent.

session

session(
    output_dir: Path | str | None = None,
    input_files: str
    | Path
    | list[str | Path]
    | None = None,
    skills_dir: Path | str | None = None,
    resume: bool = False,
    clear_cache_on_success: bool = True,
    cache_on_interrupt: bool = True,
) -> Self

Configure a session and return self for use as async context manager.

Parameters:

Name	Type	Description	Default
output_dir	Path | str | None	Directory to save output files from finish_params.paths	None
input_files	str | Path | list[str | Path] | None	Files to upload to the execution environment at session start. Accepts a single path or list of paths. Supports: - File paths (str or Path) - Directory paths (uploaded recursively) - Glob patterns (e.g., "data/.csv", "/.py") Raises ValueError if no CodeExecToolProvider is configured or if a glob pattern matches no files.	None
skills_dir	Path | str | None	Directory containing skill definitions to load and make available to the agent. Skills are uploaded to the execution environment and their metadata is included in the system prompt.	None
resume	bool	If True, attempt to resume from cached state if available. The cache is identified by hashing the init_msgs passed to run(). Cached state includes message history, current turn, and execution environment files from a previous interrupted run.	False
clear_cache_on_success	bool	If True (default), automatically clear the cache when the agent completes successfully. Set to False to preserve caches for inspection or debugging.	True
cache_on_interrupt	bool	If True (default), set up a SIGINT handler to cache state on Ctrl+C. Set to False when running agents in threads or subprocesses where signal handlers cannot be registered from non-main threads.	True

Returns:

Type	Description
Self	Self, for use with async with agent.session(...) as session:

Example

async with agent.session(output_dir="./output", input_files="data/*.csv") as session: result = await session.run("Analyze the CSV files")

Note

Multiple concurrent sessions from the same Agent instance are supported. Each session maintains isolated state via ContextVar.

Source code in src/stirrup/core/agent.py

def session(
    self,
    output_dir: Path | str | None = None,
    input_files: str | Path | list[str | Path] | None = None,
    skills_dir: Path | str | None = None,
    resume: bool = False,
    clear_cache_on_success: bool = True,
    cache_on_interrupt: bool = True,
) -> Self:
    """Configure a session and return self for use as async context manager.

    Args:
        output_dir: Directory to save output files from finish_params.paths
        input_files: Files to upload to the execution environment at session start.
                    Accepts a single path or list of paths. Supports:
                    - File paths (str or Path)
                    - Directory paths (uploaded recursively)
                    - Glob patterns (e.g., "data/*.csv", "**/*.py")
                    Raises ValueError if no CodeExecToolProvider is configured
                    or if a glob pattern matches no files.
        skills_dir: Directory containing skill definitions to load and make available
                   to the agent. Skills are uploaded to the execution environment
                   and their metadata is included in the system prompt.
        resume: If True, attempt to resume from cached state if available.
               The cache is identified by hashing the init_msgs passed to run().
               Cached state includes message history, current turn, and execution
               environment files from a previous interrupted run.
        clear_cache_on_success: If True (default), automatically clear the cache
                               when the agent completes successfully. Set to False
                               to preserve caches for inspection or debugging.
        cache_on_interrupt: If True (default), set up a SIGINT handler to cache
                           state on Ctrl+C. Set to False when running agents in
                           threads or subprocesses where signal handlers cannot
                           be registered from non-main threads.

    Returns:
        Self, for use with `async with agent.session(...) as session:`

    Example:
        async with agent.session(output_dir="./output", input_files="data/*.csv") as session:
            result = await session.run("Analyze the CSV files")

    Note:
        Multiple concurrent sessions from the same Agent instance are supported.
        Each session maintains isolated state via ContextVar.

    """
    self._pending_output_dir = Path(output_dir) if output_dir else None
    self._pending_input_files = input_files
    self._pending_skills_dir = Path(skills_dir) if skills_dir else None
    self._resume = resume
    self._clear_cache_on_success = clear_cache_on_success
    self._cache_on_interrupt = cache_on_interrupt
    return self

__aenter__ async

__aenter__() -> Self

Enter session context: set up tools, logging, and resources.

Creates a new SessionState and stores it in the _SESSION_STATE ContextVar, allowing concurrent sessions from the same Agent instance.

Source code in src/stirrup/core/agent.py

async def __aenter__(self) -> Self:
    """Enter session context: set up tools, logging, and resources.

    Creates a new SessionState and stores it in the _SESSION_STATE ContextVar,
    allowing concurrent sessions from the same Agent instance.
    """
    exit_stack = AsyncExitStack()
    await exit_stack.__aenter__()

    # Get parent state if exists (for subagent file transfer)
    parent_state = _SESSION_STATE.get(None)

    current_depth = _PARENT_DEPTH.get()

    # Create session state and store in ContextVar
    state = SessionState(
        exit_stack=exit_stack,
        output_dir=str(self._pending_output_dir) if self._pending_output_dir else None,
        parent_exec_env=parent_state.exec_env if parent_state else None,
        depth=current_depth,
        logger=self._logger,
    )
    _SESSION_STATE.set(state)

    try:
        # === TWO-PASS TOOL INITIALIZATION ===
        # First pass initializes CodeExecToolProvider so that dependent tools
        # (like ViewImageToolProvider) can access state.exec_env in second pass.
        active_tools: list[Tool] = []

        # First pass: Initialize CodeExecToolProvider (at most one allowed)
        code_exec_providers = [t for t in self._tools if isinstance(t, CodeExecToolProvider)]
        if len(code_exec_providers) > 1:
            raise ValueError(
                f"Agent can only have one CodeExecToolProvider, found {len(code_exec_providers)}: "
                f"{[type(p).__name__ for p in code_exec_providers]}"
            )

        if code_exec_providers:
            provider = code_exec_providers[0]
            result = await exit_stack.enter_async_context(provider)
            if isinstance(result, list):
                active_tools.extend(result)
            else:
                active_tools.append(result)
            state.exec_env = provider

        # Second pass: Initialize remaining ToolProviders and static Tools
        for tool in self._tools:
            if isinstance(tool, CodeExecToolProvider):
                continue  # Already processed in first pass

            if isinstance(tool, ToolProvider):
                # ToolProvider: enter context and get returned tool(s)
                result = await exit_stack.enter_async_context(tool)
                # Handle both single Tool and list[Tool] returns (e.g., MCPToolProvider)
                if isinstance(result, list):
                    active_tools.extend(result)
                else:
                    active_tools.append(result)
            else:
                # Static Tool, use directly
                active_tools.append(tool)

        # Build active tools dict with finish tool (stored on instance, not session)
        self._active_tools = {FINISH_TOOL_NAME: self._finish_tool}
        self._active_tools.update({t.name: t for t in active_tools})

        # Validate subagent code exec requirements (only at root level)
        if current_depth == 0:
            self._validate_subagent_code_exec_requirements()

        # Upload input files to exec_env if specified
        if self._pending_input_files:
            if not state.exec_env:
                raise ValueError("input_files specified but no CodeExecToolProvider configured")

            logger.debug(
                "[%s __aenter__] Uploading input files: %s, depth=%d, parent_exec_env=%s, parent_exec_env._temp_dir=%s",
                self._name,
                self._pending_input_files,
                state.depth,
                type(state.parent_exec_env).__name__ if state.parent_exec_env else None,
                getattr(state.parent_exec_env, "_temp_dir", "N/A") if state.parent_exec_env else None,
            )

            if state.depth > 0 and state.parent_exec_env:
                # SUBAGENT: Read files from parent's exec env, write to subagent's exec env
                # input_files are paths within the parent's environment
                result = await state.exec_env.upload_files(
                    *self._pending_input_files,
                    source_env=state.parent_exec_env,
                )
            else:
                # ROOT AGENT: Read files from local filesystem
                resolved = self._resolve_input_files(self._pending_input_files)
                result = await state.exec_env.upload_files(*resolved)

            logger.debug(
                "[%s __aenter__] Upload result: uploaded=%s, failed=%s", self._name, result.uploaded, result.failed
            )

            # Store uploaded paths for system prompt
            state.uploaded_file_paths = [uf.dest_path for uf in result.uploaded]

            if result.failed:
                raise RuntimeError(f"Failed to upload files: {result.failed}")
        self._pending_input_files = None  # Clear pending state

        # Upload skills directory if it exists and load metadata
        if self._pending_skills_dir:
            skills_path = self._pending_skills_dir
            if skills_path.exists() and skills_path.is_dir():
                if state.exec_env:
                    logger.debug("[%s __aenter__] Uploading skills directory: %s", self._name, skills_path)
                    await state.exec_env.upload_files(skills_path, dest_dir="skills")
                # Load skills metadata (even if no exec_env, for system prompt)
                state.skills_metadata = load_skills_metadata(skills_path)
                logger.debug("[%s __aenter__] Loaded %d skills", self._name, len(state.skills_metadata))
            self._pending_skills_dir = None  # Clear pending state
        elif parent_state and parent_state.skills_metadata:
            # Sub-agent: inherit skills from parent
            state.skills_metadata = parent_state.skills_metadata
            logger.debug("[%s __aenter__] Inherited %d skills from parent", self._name, len(state.skills_metadata))
            # Transfer skills directory from parent's exec_env to sub-agent's exec_env
            if state.exec_env and parent_state.exec_env:
                await state.exec_env.upload_files("skills", source_env=parent_state.exec_env)

        # Configure and enter logger context
        self._logger.name = self._name
        self._logger.model = self._client.model_slug
        self._logger.max_turns = self._max_turns
        # depth is already set (0 for main agent, passed in for sub-agents)
        self._logger.__enter__()

        # Set up signal handler for graceful caching on interrupt (root agent only)
        if current_depth == 0 and self._cache_on_interrupt:
            self._original_sigint = signal.getsignal(signal.SIGINT)
            signal.signal(signal.SIGINT, self._handle_interrupt)

        return self

    except Exception:
        await exit_stack.__aexit__(None, None, None)
        raise

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit session context: save files, cleanup resources.

File handling is depth-aware: - Root agent (depth=0): Saves files to local filesystem output_dir - Subagent (depth>0): Transfers files to parent's exec env at output_dir path

Source code in src/stirrup/core/agent.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit session context: save files, cleanup resources.

    File handling is depth-aware:
    - Root agent (depth=0): Saves files to local filesystem output_dir
    - Subagent (depth>0): Transfers files to parent's exec env at output_dir path
    """
    state = _SESSION_STATE.get()

    try:
        # Cache state on non-success exit (only at root level)
        should_cache = (
            state.depth == 0
            and (exc_type is not None or self._last_finish_params is None)
            and self._current_task_hash is not None
            and self._current_run_state is not None
        )

        logger.debug(
            "[%s __aexit__] Cache decision: should_cache=%s, depth=%d, exc_type=%s, "
            "finish_params=%s, task_hash=%s, run_state=%s",
            self._name,
            should_cache,
            state.depth,
            exc_type,
            self._last_finish_params is not None,
            self._current_task_hash,
            self._current_run_state is not None,
        )

        if should_cache:
            cache_manager = CacheManager(clear_on_success=self._clear_cache_on_success)

            exec_env_dir = state.exec_env.temp_dir if state.exec_env else None

            # Explicit checks to keep type checker happy - should_cache condition guarantees these
            if self._current_task_hash is None or self._current_run_state is None:
                raise ValueError("Cache state is unexpectedly None after should_cache check")

            # Temporarily block SIGINT during cache save to prevent interruption
            original_handler = signal.getsignal(signal.SIGINT)
            signal.signal(signal.SIGINT, signal.SIG_IGN)
            try:
                cache_manager.save_state(
                    self._current_task_hash,
                    self._current_run_state,
                    exec_env_dir,
                )
            finally:
                signal.signal(signal.SIGINT, original_handler)
            self._logger.info(f"Cached state for task {self._current_task_hash}")
        # Save files from finish_params.paths based on depth
        if state.output_dir and self._last_finish_params and state.exec_env:
            paths = getattr(self._last_finish_params, "paths", None)
            if paths:
                if state.depth == 0:
                    # ROOT AGENT: Save to local filesystem
                    output_path = Path(state.output_dir)
                    output_path.mkdir(parents=True, exist_ok=True)
                    logger.debug(
                        "[%s] ROOT AGENT (depth=0): Saving %d file(s) to local filesystem: %s -> %s",
                        self._name,
                        len(paths),
                        paths,
                        output_path,
                    )
                    result = await state.exec_env.save_output_files(paths, output_path, dest_env=None)
                    logger.debug(
                        "[%s] ROOT AGENT: Saved %d file(s), failed %d",
                        self._name,
                        len(result.saved),
                        len(result.failed),
                    )
                else:
                    # SUBAGENT: Transfer to parent's exec env
                    if state.parent_exec_env:
                        logger.debug(
                            "[%s] SUBAGENT (depth=%d): Transferring %d file(s) to parent exec env: %s -> %s",
                            self._name,
                            state.depth,
                            len(paths),
                            paths,
                            state.output_dir,
                        )
                        result = await state.exec_env.save_output_files(
                            paths, state.output_dir, dest_env=state.parent_exec_env
                        )
                        # Store transferred paths for returning to parent
                        self._transferred_paths = [str(sf.output_path) for sf in result.saved]
                        logger.debug(
                            "[%s] SUBAGENT: Transferred %d file(s) to parent, failed %d. Paths: %s",
                            self._name,
                            len(result.saved),
                            len(result.failed),
                            self._transferred_paths,
                        )
                        if result.failed:
                            logger.warning("Failed to transfer some files to parent env: %s", result.failed)
                    else:
                        logger.warning(
                            "Subagent at depth %d has exec_env but no parent_exec_env. "
                            "Files will not be transferred.",
                            state.depth,
                        )
    finally:
        # Restore original signal handler (root agent only)
        if hasattr(self, "_original_sigint"):
            signal.signal(signal.SIGINT, self._original_sigint)
            del self._original_sigint

        # Exit logger context
        self._logger.finish_params = self._last_finish_params
        self._logger.run_metadata = self._last_run_metadata
        self._logger.output_dir = str(state.output_dir) if state.output_dir else None
        self._logger.__exit__(exc_type, exc_val, exc_tb)

        # Cleanup all async resources
        await state.exit_stack.__aexit__(exc_type, exc_val, exc_tb)

run_tool async

run_tool(
    tool_call: ToolCall, run_metadata: dict[str, list[Any]]
) -> ToolMessage

Execute a single tool call with error handling for invalid JSON/arguments.

Returns a ToolMessage containing either the tool output or an error description. Metadata from the tool result is stored in the provided run_metadata dict.

Source code in src/stirrup/core/agent.py

async def run_tool(self, tool_call: ToolCall, run_metadata: dict[str, list[Any]]) -> ToolMessage:
    """Execute a single tool call with error handling for invalid JSON/arguments.

    Returns a ToolMessage containing either the tool output or an error description.
    Metadata from the tool result is stored in the provided run_metadata dict.
    """
    tool = self._active_tools.get(tool_call.name)
    result: ToolResult
    args_valid = True

    # Ensure tool is tracked in metadata dict (even if no metadata returned)
    if tool_call.name not in run_metadata:
        run_metadata[tool_call.name] = []

    if tool:
        try:
            # Normalize empty arguments to valid empty JSON object
            args = tool_call.arguments if tool_call.arguments and tool_call.arguments.strip() else "{}"
            params = tool.parameters.model_validate_json(args)

            # Set parent depth for sub-agent tools to read
            prev_depth = _PARENT_DEPTH.set(self._logger.depth)
            try:
                if inspect.iscoroutinefunction(tool.executor):
                    result = await tool.executor(params)  # ty: ignore[invalid-await]
                elif self._run_sync_in_thread:
                    # ty: ignore - type checker doesn't understand iscoroutinefunction narrowing
                    result = await anyio.to_thread.run_sync(tool.executor, params)  # ty: ignore[unresolved-attribute]
                else:
                    # ty: ignore - iscoroutinefunction check above ensures this is sync
                    result = tool.executor(params)  # ty: ignore[invalid-assignment]
            finally:
                _PARENT_DEPTH.reset(prev_depth)

            # Store metadata if present
            if result.metadata is not None:
                run_metadata[tool_call.name].append(result.metadata)
        except ValidationError:
            LOGGER.debug(
                "LLMClient tried to use the tool %s but the tool arguments are not valid: %r",
                tool_call.name,
                tool_call.arguments,
            )
            result = ToolResult(content="Tool arguments are not valid", success=False)
            args_valid = False
    else:
        LOGGER.debug(f"LLMClient tried to use the tool {tool_call.name} which is not in the tools list")
        result = ToolResult(content=f"{tool_call.name} is not a valid tool", success=False)

    return ToolMessage(
        content=result.content,
        tool_call_id=tool_call.tool_call_id,
        name=tool_call.name,
        args_was_valid=args_valid,
        success=result.success,
    )

step async

step(
    messages: list[ChatMessage],
    run_metadata: dict[str, list[Any]],
    turn: int = 0,
    max_turns: int = 0,
) -> tuple[
    AssistantMessage, list[ToolMessage], FinishParams | None
]

Execute one agent step: generate assistant message and run any requested tool calls.

Parameters:

Name	Type	Description	Default
messages	list[ChatMessage]	Current conversation messages	required
run_metadata	dict[str, list[Any]]	Metadata storage for tool results	required
turn	int	Current turn number (1-indexed) for logging	0
max_turns	int	Maximum turns for logging	0

Returns the assistant message, tool execution results, and finish tool call (if present).

Source code in src/stirrup/core/agent.py

async def step(
    self,
    messages: list[ChatMessage],
    run_metadata: dict[str, list[Any]],
    turn: int = 0,
    max_turns: int = 0,
) -> tuple[AssistantMessage, list[ToolMessage], FinishParams | None]:
    """Execute one agent step: generate assistant message and run any requested tool calls.

    Args:
        messages: Current conversation messages
        run_metadata: Metadata storage for tool results
        turn: Current turn number (1-indexed) for logging
        max_turns: Maximum turns for logging

    Returns the assistant message, tool execution results, and finish tool call (if present).

    """
    assistant_message = await self._client.generate(messages, self._active_tools)

    # Log assistant message immediately
    if turn > 0:
        self._logger.assistant_message(turn, max_turns, assistant_message)

    finish_params: FinishParams | None = None
    tool_messages: list[ToolMessage] = []
    if assistant_message.tool_calls:
        tool_messages = []
        for tool_call in assistant_message.tool_calls:
            tool_message = await self.run_tool(tool_call, run_metadata)
            tool_messages.append(tool_message)

            if tool_message.success and tool_message.name == FINISH_TOOL_NAME:
                finish_params = self._finish_tool.parameters.model_validate_json(tool_call.arguments)

            # Log tool result immediately
            self._logger.tool_result(tool_message)

    return assistant_message, tool_messages, finish_params

summarize_messages async

summarize_messages(
    messages: list[ChatMessage],
) -> list[ChatMessage]

Condense message history using LLM to stay within context window.

Source code in src/stirrup/core/agent.py

async def summarize_messages(self, messages: list[ChatMessage]) -> list[ChatMessage]:
    """Condense message history using LLM to stay within context window."""
    task_context: list[ChatMessage] = list(takewhile(lambda m: not isinstance(m, AssistantMessage), messages))

    summary_prompt = [*messages, UserMessage(content=MESSAGE_SUMMARIZER)]

    # We need to pass the tools to the client so that it has context of tools used in the conversation
    summary = await self._client.generate(summary_prompt, self._active_tools)

    summary_bridge_prompt = MESSAGE_SUMMARIZER_BRIDGE_TEMPLATE.format(summary=summary.content)
    summary_bridge = UserMessage(content=summary_bridge_prompt)
    acknowledgement_msg = UserMessage(content="Got it, thanks!")

    # Log the completed summary
    summary_content = summary.content if isinstance(summary.content, str) else str(summary.content)
    self._logger.context_summarization_complete(summary_content, summary_bridge_prompt)

    return [*task_context, summary_bridge, acknowledgement_msg]

run async

run(
    init_msgs: str | list[ChatMessage],
    *,
    depth: int | None = None,
) -> tuple[
    FinishParams | None,
    list[list[ChatMessage]],
    dict[str, Any],
]

Execute the agent loop until finish tool is called or max_turns reached.

A base system prompt is automatically prepended to all runs, including: - Agent purpose and max_turns info - List of input files (if provided via session()) - User's custom system_prompt (if configured in init)

Parameters:

Name	Type	Description	Default
init_msgs	str | list[ChatMessage]	Either a string prompt (converted to UserMessage) or a list of ChatMessage to extend the conversation after the system prompt.	required
depth	int | None	Logging depth for sub-agent runs. If provided, updates logger.depth for this run.	None

Returns:

Type	Description
FinishParams | None	Tuple of (finish params, message history, run metadata).
list[list[ChatMessage]]	finish params is None if max_turns reached.
dict[str, Any]	run metadata maps tool/agent names to lists of metadata returned by each call.

Example

Simple string prompt

await agent.run("Analyze this data and create a report")

Multiple messages

await agent.run([ UserMessage(content="First, read the data"), AssistantMessage(content="I've read the data file..."), UserMessage(content="Now analyze it"), ])

Source code in src/stirrup/core/agent.py

async def run(
    self,
    init_msgs: str | list[ChatMessage],
    *,
    depth: int | None = None,
) -> tuple[FinishParams | None, list[list[ChatMessage]], dict[str, Any]]:
    """Execute the agent loop until finish tool is called or max_turns reached.

    A base system prompt is automatically prepended to all runs, including:
    - Agent purpose and max_turns info
    - List of input files (if provided via session())
    - User's custom system_prompt (if configured in __init__)

    Args:
        init_msgs: Either a string prompt (converted to UserMessage) or a list of
                  ChatMessage to extend the conversation after the system prompt.
        depth: Logging depth for sub-agent runs. If provided, updates logger.depth for this run.

    Returns:
        Tuple of (finish params, message history, run metadata).
        finish params is None if max_turns reached.
        run metadata maps tool/agent names to lists of metadata returned by each call.

    Example:
        # Simple string prompt
        await agent.run("Analyze this data and create a report")

        # Multiple messages
        await agent.run([
            UserMessage(content="First, read the data"),
            AssistantMessage(content="I've read the data file..."),
            UserMessage(content="Now analyze it"),
        ])

    """

    # Compute task hash for caching/resume
    task_hash = compute_task_hash(init_msgs)
    self._current_task_hash = task_hash

    # Initialize cache manager
    cache_manager = CacheManager(clear_on_success=self._clear_cache_on_success)
    start_turn = 0
    resumed = False

    # Try to resume from cache if requested
    if self._resume:
        state = _SESSION_STATE.get()
        cached = cache_manager.load_state(task_hash)
        if cached:
            # Restore files to exec env
            if state.exec_env and state.exec_env.temp_dir:
                cache_manager.restore_files(task_hash, state.exec_env.temp_dir)

            # Restore state
            msgs = cached.msgs
            full_msg_history = cached.full_msg_history
            run_metadata = cached.run_metadata
            start_turn = cached.turn
            resumed = True
            self._logger.info(f"Resuming from cached state at turn {start_turn}")
        else:
            self._logger.info(f"No cache found for task {task_hash}, starting fresh")

    if not resumed:
        msgs: list[ChatMessage] = []

        # Build the complete system prompt (base + input files + user instructions)
        full_system_prompt = self._build_system_prompt()
        msgs.append(SystemMessage(content=full_system_prompt))

        if isinstance(init_msgs, str):
            msgs.append(UserMessage(content=init_msgs))
        else:
            msgs.extend(init_msgs)

        # Local metadata storage - isolated per run() invocation for thread safety
        run_metadata: dict[str, list[Any]] = {}

        full_msg_history: list[list[ChatMessage]] = []

    # Set logger depth if provided (for sub-agent runs)
    if depth is not None:
        self._logger.depth = depth

    # Log the task at run start (only if not resuming)
    if not resumed:
        self._logger.task_message(msgs[-1].content)

    # Show warnings (top-level only, if logger supports it)
    if self._logger.depth == 0 and isinstance(self._logger, AgentLogger):
        run_warnings = self._collect_warnings()
        if run_warnings:
            self._logger.warnings_message(run_warnings)

    # Use logger callback if available and not overridden
    step_callback = self._logger.on_step

    full_msg_history: list[list[ChatMessage]] = []

    # Cumulative stats for spinner
    total_tool_calls = 0
    total_input_tokens = 0
    total_output_tokens = 0

    for i in range(start_turn, self._max_turns):
        # Capture current state for potential caching (before any async work)
        self._current_run_state = CacheState(
            msgs=list(msgs),
            full_msg_history=[list(group) for group in full_msg_history],
            turn=i,
            run_metadata=dict(run_metadata),
            task_hash=task_hash,
            agent_name=self._name,
        )
        if self._max_turns - i <= self._turns_remaining_warning_threshold and i != 0:
            num_turns_remaining_msg = _num_turns_remaining_msg(self._max_turns - i)
            msgs.append(num_turns_remaining_msg)
            self._logger.user_message(num_turns_remaining_msg)

        # Pass turn info to step() for real-time logging
        assistant_message, tool_messages, finish_params = await self.step(
            msgs,
            run_metadata,
            turn=i + 1,
            max_turns=self._max_turns,
        )

        # Update cumulative stats
        total_tool_calls += len(tool_messages)
        total_input_tokens += assistant_message.token_usage.input
        total_output_tokens += assistant_message.token_usage.output

        # Call progress callback after step completes
        if step_callback:
            step_callback(i + 1, total_tool_calls, total_input_tokens, total_output_tokens)

        user_messages: list[UserMessage] = []
        if self._text_only_tool_responses:
            tool_messages, user_messages = _handle_text_only_tool_responses(tool_messages)

        # Log user messages (e.g., image content extracted from tool responses)
        for user_msg in user_messages:
            self._logger.user_message(user_msg)

        msgs.extend([assistant_message, *tool_messages, *user_messages])

        if finish_params:
            break

        pct_context_used = assistant_message.token_usage.total / self._client.max_tokens
        if pct_context_used >= self._context_summarization_cutoff and i + 1 != self._max_turns:
            self._logger.context_summarization_start(pct_context_used, self._context_summarization_cutoff)
            full_msg_history.append(msgs)
            msgs = await self.summarize_messages(msgs)
    else:
        LOGGER.error(
            f"Maximum number of turns reached: {self._max_turns}. The agent was not able to finish the task. Consider increasing the max_turns parameter.",
        )

    full_msg_history.append(msgs)

    # Add agent's own token usage to run_metadata under "token_usage" key
    run_metadata["token_usage"] = _get_total_token_usage(full_msg_history)

    # Store for __aexit__ to access (on instance for this agent)
    self._last_finish_params = finish_params
    self._last_run_metadata = run_metadata

    # Clear cache on successful completion (finish_params is set)
    if finish_params is not None and cache_manager.clear_on_success:
        cache_manager.clear_cache(task_hash)
        self._current_task_hash = None
        self._current_run_state = None

    return finish_params, full_msg_history, run_metadata

to_tool

to_tool(
    *,
    description: str = DEFAULT_SUB_AGENT_DESCRIPTION,
    system_prompt: str | None = None,
) -> Tool[SubAgentParams, SubAgentMetadata]

Convert this Agent to a Tool for use as a sub-agent.

Parameters:

Name	Type	Description	Default
description	str	Tool description shown to the parent agent	DEFAULT_SUB_AGENT_DESCRIPTION
system_prompt	str | None	Optional system prompt to prepend when running	None

Returns:

Type	Description
Tool[SubAgentParams, SubAgentMetadata]	Tool that executes this agent when called, returning SubAgentMetadata
Tool[SubAgentParams, SubAgentMetadata]	containing token usage, message history, and any metadata from tools
Tool[SubAgentParams, SubAgentMetadata]	the sub-agent used.

Source code in src/stirrup/core/agent.py

def to_tool(
    self,
    *,
    description: str = DEFAULT_SUB_AGENT_DESCRIPTION,
    system_prompt: str | None = None,
) -> Tool[SubAgentParams, SubAgentMetadata]:
    """Convert this Agent to a Tool for use as a sub-agent.

    Args:
        description: Tool description shown to the parent agent
        system_prompt: Optional system prompt to prepend when running

    Returns:
        Tool that executes this agent when called, returning SubAgentMetadata
        containing token usage, message history, and any metadata from tools
        the sub-agent used.

    """
    agent = self  # Capture self for closure

    async def sub_agent_executor(params: SubAgentParams) -> ToolResult[SubAgentMetadata]:
        """Execute the sub-agent with the given task.

        Sub-agents enter their own full session to ensure:
        1. Tool isolation - each agent only sees its own tools (fixes recursive sub-agent bug)
        2. Proper ToolProvider lifecycle - sub-agent's ToolProviders are initialized
        3. Correct logging - logger context is entered for proper output formatting
        """
        # Get parent's depth and calculate subagent depth
        parent_depth = _PARENT_DEPTH.get()
        sub_agent_depth = parent_depth + 1

        # Save parent's session state so we can restore it after subagent completes
        # This ensures sibling subagents see the parent's state, not a previous sibling's stale state
        parent_session_state = _SESSION_STATE.get(None)
        logger.debug(
            "[%s] PRE-SESSION: _SESSION_STATE=%s, exec_env=%s, exec_env._temp_dir=%s",
            agent.name,
            id(parent_session_state) if parent_session_state else None,
            type(parent_session_state.exec_env).__name__
            if parent_session_state and parent_session_state.exec_env
            else None,
            getattr(parent_session_state.exec_env, "_temp_dir", "N/A")
            if parent_session_state and parent_session_state.exec_env
            else None,
        )

        # Set _PARENT_DEPTH to subagent's depth BEFORE entering session
        # so that __aenter__ reads the correct depth for SessionState.depth
        prev_depth = _PARENT_DEPTH.set(sub_agent_depth)
        try:
            init_msgs: list[ChatMessage] = []
            if system_prompt:
                init_msgs.append(SystemMessage(content=system_prompt))
            init_msgs.append(UserMessage(content=params.task))

            # Sub-agent enters its own full session for tool isolation and proper lifecycle
            # output_dir is a path within the parent's exec env (not local filesystem)
            # Files are transferred to parent's env at __aexit__ via save_output_files(dest_env=parent)
            async with agent.session(
                output_dir=".",  # Path in parent's exec env
                input_files=list(params.input_files) if params.input_files else None,  # ty: ignore[invalid-argument-type]
            ) as agent_session:
                # Override logger depth for proper indentation in console output
                agent_session._logger.depth = sub_agent_depth  # noqa: SLF001

                finish_params, msg_history, run_metadata = await agent_session.run(init_msgs)

                # Extract the last assistant message with actual content (not just tool calls)
                last_assistant_msg: AssistantMessage | None = None
                for msg_group in reversed(msg_history):
                    for msg in reversed(msg_group):
                        if isinstance(msg, AssistantMessage) and msg.content:
                            last_assistant_msg = msg
                            break
                    if last_assistant_msg:
                        break

                # Build content from the assistant message and/or finish params
                content_parts: list[str] = []

                if last_assistant_msg and last_assistant_msg.content:
                    content = last_assistant_msg.content
                    if isinstance(content, list):
                        content = "\n".join(str(block) for block in content)
                    content_parts.append(content)

                # Include finish params if available (they often contain the actual result)
                if finish_params is not None:
                    finish_dict = finish_params.model_dump()
                    if finish_dict:
                        content_parts.append(f"Finish params: {finish_dict}")

                # Report files transferred to parent's exec env (set in __aexit__)
                transferred_paths = agent_session._transferred_paths  # noqa: SLF001
                if transferred_paths:
                    content_parts.append(f"Files available in your environment: {transferred_paths}")

                if not content_parts:
                    result_content = "<sub_agent_result>\n<error>No assistant message or finish params found</error>\n</sub_agent_result>"
                else:
                    content = "\n".join(content_parts)
                    result_content = (
                        f"<sub_agent_result>"
                        f"\n<response>{content}</response>"
                        f"\n<finished>{finish_params is not None}</finished>"
                        f"\n</sub_agent_result>"
                    )

                # Create subagent metadata with token usage, message history, and run metadata
                sub_metadata = SubAgentMetadata(
                    message_history=msg_history,
                    run_metadata=run_metadata,
                )

                return ToolResult(content=result_content, metadata=sub_metadata)

        except Exception as e:
            # On error, return empty metadata
            error_metadata = SubAgentMetadata(
                message_history=[],
                run_metadata={},
            )
            return ToolResult(
                content=f"<sub_agent_result>\n<error>{e!s}</error>\n</sub_agent_result>",
                success=False,
                metadata=error_metadata,
            )
        finally:
            # DEBUG: Log SESSION_STATE after subagent session
            post_session_state = _SESSION_STATE.get(None)
            logger.debug(
                "[%s] POST-SESSION: _SESSION_STATE=%s, exec_env=%s, exec_env._temp_dir=%s",
                agent.name,
                id(post_session_state) if post_session_state else None,
                type(post_session_state.exec_env).__name__
                if post_session_state and post_session_state.exec_env
                else None,
                getattr(post_session_state.exec_env, "_temp_dir", "N/A")
                if post_session_state and post_session_state.exec_env
                else None,
            )

            # Restore parent's depth
            _PARENT_DEPTH.reset(prev_depth)
            # Restore parent's session state so next sibling subagent sees it
            if parent_session_state is not None:
                _SESSION_STATE.set(parent_session_state)

    return Tool[SubAgentParams, SubAgentMetadata](
        name=self._name,
        description=description,
        parameters=SubAgentParams,
        executor=sub_agent_executor,  # ty: ignore[invalid-argument-type]
    )

compute_task_hash

compute_task_hash(
    init_msgs: str | list[ChatMessage],
) -> str

Compute deterministic hash from initial messages for cache identification.

Parameters:

Name	Type	Description	Default
init_msgs	str | list[ChatMessage]	Either a string prompt or list of ChatMessage objects.	required

Returns:

Type	Description
str	First 12 characters of SHA256 hash (hex) for readability.

Source code in src/stirrup/core/cache.py

def compute_task_hash(init_msgs: str | list[ChatMessage]) -> str:
    """Compute deterministic hash from initial messages for cache identification.

    Args:
        init_msgs: Either a string prompt or list of ChatMessage objects.

    Returns:
        First 12 characters of SHA256 hash (hex) for readability.
    """
    if isinstance(init_msgs, str):
        content = init_msgs
    else:
        # Serialize messages to JSON for hashing
        content = json.dumps(
            [serialize_message(msg) for msg in init_msgs],
            sort_keys=True,
            ensure_ascii=True,
        )

    hash_bytes = hashlib.sha256(content.encode("utf-8")).hexdigest()
    return hash_bytes[:12]

format_skills_section

format_skills_section(skills: list[SkillMetadata]) -> str

Format skills metadata as a system prompt section.

Parameters:

Name	Type	Description	Default
skills	list[SkillMetadata]	List of skill metadata to include	required

Returns:

Type	Description
str	Formatted string for inclusion in system prompt.
str	Returns empty string if no skills provided.

Source code in src/stirrup/skills/skills.py

def format_skills_section(skills: list[SkillMetadata]) -> str:
    """Format skills metadata as a system prompt section.

    Args:
        skills: List of skill metadata to include

    Returns:
        Formatted string for inclusion in system prompt.
        Returns empty string if no skills provided.

    """
    if not skills:
        return ""

    lines = [
        "## Available Skills",
        "",
        "You have access to the following skills located in the `skills/` directory. "
        "Each skill contains a SKILL.md file with detailed instructions and potentially bundled scripts.",
        "",
        "To use a skill:",
        "1. Read the full instructions: `cat <skill_path>/SKILL.md`",
        "2. Follow the instructions and use any bundled resources as described",
        "",
    ]
    lines.extend([f"- **{skill.name}**: {skill.description} (`{skill.path}/SKILL.md`)" for skill in skills])

    return "\n".join(lines)

load_skills_metadata

load_skills_metadata(
    skills_dir: Path,
) -> list[SkillMetadata]

Scan skills directory for SKILL.md files and extract metadata.

Parameters:

Name	Type	Description	Default
skills_dir	Path	Path to the skills directory	required

Returns:

Type	Description
list[SkillMetadata]	List of SkillMetadata for each valid skill found.
list[SkillMetadata]	Returns empty list if skills_dir doesn't exist or has no skills.

Source code in src/stirrup/skills/skills.py

def load_skills_metadata(skills_dir: Path) -> list[SkillMetadata]:
    """Scan skills directory for SKILL.md files and extract metadata.

    Args:
        skills_dir: Path to the skills directory

    Returns:
        List of SkillMetadata for each valid skill found.
        Returns empty list if skills_dir doesn't exist or has no skills.

    """
    if not skills_dir.exists():
        logger.debug("Skills directory does not exist: %s", skills_dir)
        return []

    if not skills_dir.is_dir():
        logger.warning("Skills path is not a directory: %s", skills_dir)
        return []

    skills: list[SkillMetadata] = []

    for skill_path in skills_dir.iterdir():
        if not skill_path.is_dir():
            continue

        skill_md = skill_path / "SKILL.md"
        if not skill_md.exists():
            logger.debug("Skill directory missing SKILL.md: %s", skill_path)
            continue

        try:
            content = skill_md.read_text(encoding="utf-8")
            metadata = parse_frontmatter(content)

            name = metadata.get("name")
            description = metadata.get("description")

            if not name or not description:
                logger.warning(
                    "Skill %s missing required frontmatter (name, description)",
                    skill_path.name,
                )
                continue

            skills.append(
                SkillMetadata(
                    name=name,
                    description=description,
                    path=f"skills/{skill_path.name}",
                )
            )
            logger.debug("Loaded skill: %s", name)

        except Exception as e:
            logger.warning("Failed to load skill %s: %s", skill_path.name, e)

    return skills

_num_turns_remaining_msg

_num_turns_remaining_msg(
    number_of_turns_remaining: int,
) -> UserMessage

Create a user message warning the agent about remaining turns before max_turns is reached.

Source code in src/stirrup/core/agent.py

def _num_turns_remaining_msg(number_of_turns_remaining: int) -> UserMessage:
    """Create a user message warning the agent about remaining turns before max_turns is reached."""
    if number_of_turns_remaining == 1:
        return UserMessage(content="This is the last turn. Please finish the task by calling the finish tool.")
    return UserMessage(
        content=f"You have {number_of_turns_remaining} turns remaining to complete the task. Please continue. Remember you will need a separate turn to finish the task.",
    )

_handle_text_only_tool_responses

_handle_text_only_tool_responses(
    tool_messages: list[ToolMessage],
) -> tuple[list[ToolMessage], list[UserMessage]]

Extract image blocks from tool messages and convert them to user messages for text-only models.

Source code in src/stirrup/core/agent.py

def _handle_text_only_tool_responses(tool_messages: list[ToolMessage]) -> tuple[list[ToolMessage], list[UserMessage]]:
    """Extract image blocks from tool messages and convert them to user messages for text-only models."""
    user_messages: list[UserMessage] = []
    for tm in tool_messages:
        if isinstance(tm.content, list):
            for idx, block in enumerate(tm.content):
                if isinstance(block, ImageContentBlock):
                    user_messages.append(
                        UserMessage(content=[f"Here is the image for tool call {tm.tool_call_id}", block]),
                    )
                    tm.content[idx] = f"Done! The User will provide the image for tool call {tm.tool_call_id}"
                elif isinstance(block, str):
                    continue
                else:
                    raise NotImplementedError(f"Unsupported content block: {type(block)}")

    return tool_messages, user_messages

_get_total_token_usage

_get_total_token_usage(
    messages: list[list[ChatMessage]],
) -> list[TokenUsage]

Returns a list of TokenUsage objects aggregated from all AssistantMessage instances across the provided grouped message history.

Parameters:

Name	Type	Description	Default
messages	list[list[ChatMessage]]	A list where each item is a list of ChatMessage objects representing a segment or turn group of the conversation history.	required

Returns:

Type	Description
list[TokenUsage]	List of TokenUsage corresponding to each AssistantMessage in the flattened conversation history.

Source code in src/stirrup/core/agent.py

def _get_total_token_usage(messages: list[list[ChatMessage]]) -> list[TokenUsage]:
    """
    Returns a list of TokenUsage objects aggregated from all AssistantMessage
    instances across the provided grouped message history.

    Args:
        messages: A list where each item is a list of ChatMessage objects representing a segment
                  or turn group of the conversation history.

    Returns:
        List of TokenUsage corresponding to each AssistantMessage in the flattened conversation history.
    """
    return [msg.token_usage for msg in chain.from_iterable(messages) if isinstance(msg, AssistantMessage)]