Agent

stirrup.core.agent

AGENT_MAX_TURNS `module-attribute`

AGENT_MAX_TURNS = 30

CONTEXT_SUMMARIZATION_CUTOFF `module-attribute`

CONTEXT_SUMMARIZATION_CUTOFF = 0.7

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=DEFAULT_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', 'SessionAgent', '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: msgs=%d", 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", state_file)
    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", state_file)
        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]],
    task_hash: str,
    timestamp: str = (lambda: isoformat())(),
    agent_name: str = "",
    run_metadata_by_turn: dict[
        str, dict[str, list[Any]]
    ] = dict(),
)

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

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.

run_metadata_by_turn `class-attribute` `instance-attribute`

run_metadata_by_turn: dict[str, dict[str, list[Any]]] = (
    field(default_factory=dict)
)

Accumulated tool metadata keyed by assistant message id.

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],
        "run_metadata_by_turn": _serialize_run_metadata_by_turn(self.run_metadata_by_turn),
        "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"]],
        task_hash=data["task_hash"],
        timestamp=data.get("timestamp", ""),
        agent_name=data.get("agent_name", ""),
        run_metadata_by_turn=data["run_metadata_by_turn"],
    )

ContextOverflowError

Bases: Exception

Raised when LLM context window is exceeded (max_tokens or length finish_reason).

AssistantMessage

Bases: BaseModel

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

e2e_otps `property`

e2e_otps: float | None

End-to-end output tokens per second.

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, keep last for non-list internal keys)
    combined_meta: dict[str, Any] = dict(self.run_metadata)
    for key, value in other.run_metadata.items():
        if key in combined_meta and isinstance(combined_meta[key], list) and isinstance(value, list):
            combined_meta[key] = combined_meta[key] + value
        else:
            combined_meta[key] = value
    return SubAgentMetadata(
        message_history=combined_history,
        run_metadata=combined_meta,
    )

SummaryMessage

Bases: UserMessage

Summary message to the LLM.

SystemMessage

Bases: BaseModel

System-level instructions and context for the LLM.

TokenUsage

Bases: BaseModel

Token counts for LLM usage.

Token terminology: output = reasoning + answer.

output `property`

output: int

Total output tokens (reasoning + answer).

total `property`

total: int

Total token count across input, answer, 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,
        answer=self.answer + other.answer,
        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)

tool_duration `property`

tool_duration: float | None

Tool execution duration in seconds.

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

ToolUseCountMetadata

Bases: BaseModel

Generic metadata tracking tool usage count.

Implements Addable protocol for aggregation. Use this for tools that only need to track how many times they were called.

Subclasses can override add with their own type thanks to Self typing.

TurnWarningMessage

Bases: UserMessage

Warning message injected when the agent is close to max_turns.

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,
    shell_timeout: int = SHELL_TIMEOUT,
)

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`
`shell_timeout`	`int`	Per-command wall-clock timeout (seconds) applied to every `code_exec` invocation from the LLM. Defaults to `SHELL_TIMEOUT`. Callers should set this to match their application's expected long-running-command budget rather than rely on the stirrup default.	`SHELL_TIMEOUT`

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

def __init__(
    self,
    *,
    allowed_commands: list[str] | None = None,
    shell_timeout: int = SHELL_TIMEOUT,
) -> 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.
        shell_timeout: Per-command wall-clock timeout (seconds) applied to
                       every ``code_exec`` invocation from the LLM. Defaults
                       to ``SHELL_TIMEOUT``. Callers should set this to match
                       their application's expected long-running-command
                       budget rather than rely on the stirrup default.

    """
    self._allowed_commands = allowed_commands
    self._shell_timeout = shell_timeout
    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 | None = None
) -> CommandResult

Execute a shell command and return raw CommandResult.

Parameters:

Name	Type	Description	Default
`cmd`	`str`	Shell command to execute (bash syntax).	required
`timeout`	`int \| None`	Per-call wall-clock timeout (seconds). If None, falls back to `self._shell_timeout` configured on the provider, so direct callers and the LLM `code_exec` tool share the same default.	`None`

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

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

    Args:
        cmd: Shell command to execute (bash syntax).
        timeout: Per-call wall-clock timeout (seconds). If None, falls back
            to ``self._shell_timeout`` configured on the provider, so direct
            callers and the LLM ``code_exec`` tool share the same default.
    """
    ...

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]:
        # timeout=None defers to env._shell_timeout, set in CodeExecToolProvider.__init__.
        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,
    shell_timeout: int = SHELL_TIMEOUT,
)

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`
`shell_timeout`	`int`	Per-command wall-clock timeout (seconds) for every `code_exec` invocation, and the default for direct `run_command` calls that pass `timeout=None`. Defaults to `SHELL_TIMEOUT`.	`SHELL_TIMEOUT`

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,
    shell_timeout: int = SHELL_TIMEOUT,
) -> 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.
        shell_timeout: Per-command wall-clock timeout (seconds) for every
            ``code_exec`` invocation, and the default for direct
            ``run_command`` calls that pass ``timeout=None``. Defaults to
            ``SHELL_TIMEOUT``.

    """
    super().__init__(allowed_commands=allowed_commands, shell_timeout=shell_timeout)
    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 | None = None
) -> CommandResult

Execute command in the temp directory.

Parameters:

Name	Type	Description	Default
`cmd`	`str`	Shell command to execute (bash syntax).	required
`timeout`	`int \| None`	Maximum time in seconds to wait for command completion. If None, falls back to `self._shell_timeout` (set in `__init__`), so direct callers share the same budget as the LLM `code_exec` tool.	`None`

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 | None = None) -> 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.
            If None, falls back to ``self._shell_timeout`` (set in
            ``__init__``), so direct callers share the same budget as the
            LLM ``code_exec`` tool.

    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."
        )
    if timeout is None:
        timeout = self._shell_timeout

    # 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):
            # start_new_session=True puts bash at the head of its own session,
            # so pgid == child pid and we can killpg the whole tree on timeout.
            # Without it, process.kill() only terminates root bash, leaving
            # grandchildren reparented to init (same bug shape as the docker
            # orphan leak that motivated this PR).
            process = await anyio.open_process(
                ["bash", "-c", cmd],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                cwd=self._temp_dir,
                start_new_session=True,
            )

            # 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:
            with contextlib.suppress(ProcessLookupError):
                os.killpg(process.pid, signal.SIGKILL)
            with anyio.move_on_after(5):
                await process.wait()
        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)

SimpleFinishParams

Bases: BaseModel

Explanation for why the task is complete or cannot proceed.

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, "text", 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,
    exec_env_owned: bool = True,
    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. - exec_env_owned: Whether this session owns the exec_env and should clean it up. When share_parent_exec_env=True, the subagent borrows the parent's exec_env and exec_env_owned=False to prevent cleanup on subagent exit.

SubAgentParams

Bases: BaseModel

Parameters for sub-agent tool invocation.

Agent

Agent(
    client: LLMClient,
    name: str,
    *,
    max_turns: int = ...,
    system_prompt: str | None = ...,
    tools: list[Tool | ToolProvider] | None = ...,
    finish_tool: None = None,
    context_summarization_cutoff: float = ...,
    turns_remaining_warning_threshold: int = ...,
    run_sync_in_thread: bool = ...,
    text_only_tool_responses: bool = ...,
    block_successive_assistant_messages: bool = ...,
    recover_from_context_overflow: bool = ...,
    share_parent_exec_env: bool = ...,
    logger: AgentLoggerBase | None = ...,
)

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

Agent(
    client: LLMClient,
    name: str,
    *,
    max_turns: int = ...,
    system_prompt: str | None = ...,
    tools: list[Tool | ToolProvider] | None = ...,
    finish_tool: list[Tool],
    context_summarization_cutoff: float = ...,
    turns_remaining_warning_threshold: int = ...,
    run_sync_in_thread: bool = ...,
    text_only_tool_responses: bool = ...,
    block_successive_assistant_messages: bool = ...,
    recover_from_context_overflow: bool = ...,
    share_parent_exec_env: bool = ...,
    logger: AgentLoggerBase | None = ...,
)

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]
    | list[Tool]
    | 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,
    block_successive_assistant_messages: bool = True,
    recover_from_context_overflow: bool = True,
    share_parent_exec_env: bool = False,
    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] \| list[Tool] \| None`	Tool or list of Tools used to signal task completion. Defaults to SIMPLE_FINISH_TOOL. If a list is provided, a successful call to any listed tool ends the run.	`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`
`block_successive_assistant_messages`	`bool`	If True (default), automatically inject a continue message when assistant responds without tool calls to prevent successive assistant messages.	`True`
`recover_from_context_overflow`	`bool`	If True (default), drop one completed turn and retry when the model reports a context overflow. If the original prompt still overflows, the context error is raised.	`True`
`share_parent_exec_env`	`bool`	When True and used as a subagent, share the parent's code execution environment instead of creating a new one. This provides better performance (no file copying) and allows the subagent to see all files in the parent's environment. Only effective when the agent is used as a subagent via to_tool().	`False`
`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] | list[Tool] | 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,
    block_successive_assistant_messages: bool = True,
    recover_from_context_overflow: bool = True,
    # Subagent options
    share_parent_exec_env: bool = False,
    # 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 or list of Tools used to signal task completion.
                     Defaults to SIMPLE_FINISH_TOOL. If a list is provided,
                     a successful call to any listed tool ends the run.
        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
        block_successive_assistant_messages: If True (default), automatically inject a continue
                                           message when assistant responds without tool calls to
                                           prevent successive assistant messages.
        recover_from_context_overflow: If True (default), drop one completed turn and retry when
                                       the model reports a context overflow. If the original
                                       prompt still overflows, the context error is raised.
        share_parent_exec_env: When True and used as a subagent, share the parent's code
                               execution environment instead of creating a new one. This
                               provides better performance (no file copying) and allows
                               the subagent to see all files in the parent's environment.
                               Only effective when the agent is used as a subagent via to_tool().
        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_tools: dict[str, Tool[Any, Any]] = _normalize_finish_tools(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
    self._block_successive_assistant_messages = block_successive_assistant_messages
    self._recover_from_context_overflow = recover_from_context_overflow
    self._share_parent_exec_env = share_parent_exec_env

    # 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

    # Eagerly register static tools + finish tools (available without a session)
    self._active_tools: dict[str, Tool] = {}
    for tool in self._tools:
        if isinstance(tool, Tool):
            if tool.name in self._finish_tools:
                raise ValueError(
                    f"Tool name '{tool.name}' in `tools` collides with a finish tool. "
                    "Tool names must be unique across `tools` and `finish_tool`."
                )
            self._active_tools[tool.name] = tool
    self._active_tools.update(self._finish_tools)
    self._has_tool_providers = any(isinstance(t, ToolProvider) for t in self._tools)

    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[FinishParams, FinishMeta]

The single finish tool used to signal task completion.

Raises ValueError if multiple finish tools are configured; use finish_tools instead.

finish_tools `property`

finish_tools: dict[str, Tool]

All finish tools keyed by name.

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__() -> SessionAgent[FinishParams, FinishMeta]

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

Returns a SessionAgent wrapping this agent's state, providing access to both static tools and ToolProvider-created tools. The SessionAgent shares this agent's dict, so state changes are visible to aexit.

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

async def __aenter__(self) -> "SessionAgent[FinishParams, FinishMeta]":
    """Enter session context: set up tools, logging, and resources.

    Returns a SessionAgent wrapping this agent's state, providing access to
    both static tools and ToolProvider-created tools. The SessionAgent shares
    this agent's __dict__, so state changes are visible to __aexit__.
    """
    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] = []

        # Check if we should share parent's exec_env (subagent with share_parent_exec_env=True)
        should_share_exec_env = (
            self._share_parent_exec_env
            and current_depth > 0
            and parent_state is not None
            and parent_state.exec_env is not None
        )

        if should_share_exec_env:
            # SHARED EXEC ENV: Use parent's exec_env directly, don't create new one
            state.exec_env = parent_state.exec_env  # type: ignore[union-attr]
            state.exec_env_owned = False
            logger.debug(
                "[%s __aenter__] Sharing parent's exec_env: %s (temp_dir=%s)",
                self._name,
                type(state.exec_env).__name__,
                getattr(state.exec_env, "_temp_dir", "N/A"),
            )
            # Skip CodeExecToolProvider initialization but still need to add code exec tool
            # Create the tool from the shared exec_env using get_code_exec_tool()
            # (the exec_env is already entered by parent, so we just create the tool wrapper)
            if state.exec_env is None:
                raise RuntimeError("Expected shared exec_env to be set, but it is None")
            code_exec_tool = state.exec_env.get_code_exec_tool()
            active_tools.append(code_exec_tool)
        else:
            # OWNED EXEC ENV: Initialize our own 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
                state.exec_env_owned = True

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

        # Detect collisions between session-resolved tool names and finish tool names.
        # Static-Tool collisions are caught earlier in __init__; ToolProvider-derived tools
        # only have known names at this point.
        colliding = sorted({t.name for t in active_tools} & self._finish_tools.keys())
        if colliding:
            raise ValueError(
                f"Session-resolved tool name(s) {colliding} collide with finish tool name(s). "
                "Tool names must be unique across `tools` and `finish_tool`."
            )

        # Merge session-initialized tools into _active_tools and ensure finish tools take precedence.
        self._active_tools.update({t.name: t for t in active_tools})
        self._active_tools.update(self._finish_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, exec_env_owned=%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,
                state.exec_env_owned,
            )

            if state.depth > 0 and state.parent_exec_env:
                if not state.exec_env_owned:
                    # SHARED EXEC ENV: Files already accessible - no transfer needed
                    # Just record the paths as "uploaded" for system prompt
                    if isinstance(self._pending_input_files, (str, Path)):
                        state.uploaded_file_paths = [str(self._pending_input_files)]
                    else:
                        state.uploaded_file_paths = [str(p) for p in self._pending_input_files]
                    logger.debug(
                        "[%s __aenter__] Shared exec_env - files already accessible: %s",
                        self._name,
                        state.uploaded_file_paths,
                    )
                else:
                    # SEPARATE EXEC ENV: 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,
                    )
                    logger.debug(
                        "[%s __aenter__] Upload result: uploaded=%s, failed=%s",
                        self._name,
                        result.uploaded,
                        result.failed,
                    )
                    state.uploaded_file_paths = [uf.dest_path for uf in result.uploaded]
                    if result.failed:
                        raise RuntimeError(f"Failed to upload files: {result.failed}")
            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,
                )
                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
            # (only if we have a separate exec_env)
            if state.exec_env and parent_state.exec_env and state.exec_env_owned:
                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 SessionAgent.from_agent(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 self._cache_on_interrupt
            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: Handle file transfer based on exec_env ownership
                    if not state.exec_env_owned:
                        # SHARED EXEC ENV: Files already in parent's env - no transfer needed
                        # Just record the paths for reporting to parent
                        self._transferred_paths = list(paths)
                        logger.debug(
                            "[%s] SUBAGENT (depth=%d, shared_exec_env): Files already in parent env: %s",
                            self._name,
                            state.depth,
                            self._transferred_paths,
                        )
                    elif state.parent_exec_env:
                        # SEPARATE EXEC ENV: Transfer to parent's 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)

        # Reset active tools to static-only (remove session-initialized tools)
        self._active_tools = {}
        for tool in self._tools:
            if isinstance(tool, Tool):
                self._active_tools[tool.name] = tool
        self._active_tools.update(self._finish_tools)

        # 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] = []

    tool_start_time = perf_counter()

    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)

    tool_end_time = perf_counter()

    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,
        tool_start_time=tool_start_time,
        tool_end_time=tool_end_time,
    )

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:
        # If multiple finish tools were called in one turn, refuse all of them without
        # executing — silently picking one would discard the model's other intent and
        # may swap in contradictory params. Force the model to retry with a single call.
        finish_call_names = [tc.name for tc in assistant_message.tool_calls if tc.name in self._finish_tools]
        reject_all_finish_calls = len(finish_call_names) > 1

        tool_messages = []
        for tool_call in assistant_message.tool_calls:
            if reject_all_finish_calls and tool_call.name in self._finish_tools:
                now = perf_counter()
                tool_message = ToolMessage(
                    content=(
                        f"Cannot call finish tool '{tool_call.name}': multiple finish tools "
                        f"({sorted(set(finish_call_names))}) were called in the same turn. "
                        "Only one finish tool may be called per turn — retry with a single "
                        "finish tool call."
                    ),
                    tool_call_id=tool_call.tool_call_id,
                    name=tool_call.name,
                    args_was_valid=True,
                    success=False,
                    tool_start_time=now,
                    tool_end_time=now,
                )
                tool_messages.append(tool_message)
                self._logger.tool_result(tool_message)
                continue

            tool_message = await self.run_tool(tool_call, run_metadata)
            tool_messages.append(tool_message)

            if tool_message.success and tool_message.name in self._finish_tools:
                finish_tool = self._finish_tools[tool_message.name]
                finish_params = 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],
    run_metadata_by_turn: dict[str, dict[str, list[Any]]],
) -> tuple[list[ChatMessage], list[ChatMessage]]

Summarize messages, unwinding completed turns if summarization overflows.

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

async def summarize_messages(
    self,
    messages: list[ChatMessage],
    run_metadata_by_turn: dict[str, dict[str, list[Any]]],
) -> tuple[list[ChatMessage], list[ChatMessage]]:
    """Summarize messages, unwinding completed turns if summarization overflows."""
    current_messages = messages
    while True:
        try:
            task_context: list[ChatMessage] = list(
                takewhile(lambda m: not isinstance(m, (AssistantMessage, SummaryMessage)), current_messages)
            )

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

            # Give the summarizer the active tools so it can interpret prior tool calls/results.
            summary = await self._client.generate(summary_prompt, self._active_tools)

            summary_bridge_prompt = MESSAGE_SUMMARIZER_BRIDGE_TEMPLATE.format(summary=summary.content)
            summary_bridge = SummaryMessage(content=summary_bridge_prompt)
            # Use a user acknowledgement to avoid consecutive assistant messages with strict providers.
            acknowledgement_msg = UserMessage(content="Got it, thanks!")

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

            return current_messages, [*task_context, summary_bridge, acknowledgement_msg]
        except ContextOverflowError:
            # Summarization can overflow too; drop the latest completed turn and retry.
            # _unwind_context_overflow raises if that would cross a progress boundary.
            current_messages, dropped_turn_id = self._unwind_context_overflow(current_messages)
            run_metadata_by_turn.pop(dropped_turn_id, None)

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"),
        ])

    """

    # Guard: fail if ToolProviders are attached but we're not in a session
    if self._has_tool_providers and not isinstance(self, SessionAgent):
        provider_names = [type(t).__name__ for t in self._tools if isinstance(t, ToolProvider)]
        raise RuntimeError(
            f"Agent.run() called without a session, but the agent has ToolProviders "
            f"that require session initialization: {provider_names}. "
            f"Use `async with agent.session(...) as session: await session.run(...)` instead."
        )

    # 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)
    resumed = False
    cached_run_metadata_by_turn: dict[str, dict[str, list[Any]]] | None = None

    # 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
            cached_run_metadata_by_turn = cached.run_metadata_by_turn
            resumed = True
            restored_turn = _get_turn_count(full_msg_history, msgs)
            self._logger.info(f"Resuming from cached state at turn {restored_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)

        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
    run_metadata_by_turn: dict[str, dict[str, list[Any]]] = {}
    if cached_run_metadata_by_turn is not None:
        run_metadata_by_turn.update(cached_run_metadata_by_turn)

    # Cumulative stats for spinner
    total_tool_calls = 0
    total_input_tokens = 0
    total_output_tokens = 0
    finish_params: FinishParams | None = None

    while _get_turn_count(full_msg_history, msgs) < self._max_turns:
        turn = _get_turn_count(full_msg_history, msgs)
        # 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],
            run_metadata_by_turn={turn_id: dict(metadata) for turn_id, metadata in run_metadata_by_turn.items()},
            task_hash=task_hash,
            agent_name=self._name,
        )
        if self._max_turns - turn <= self._turns_remaining_warning_threshold and turn != 0:
            num_turns_remaining_msg = _num_turns_remaining_msg(self._max_turns - turn)
            msgs.append(num_turns_remaining_msg)
            self._logger.user_message(num_turns_remaining_msg)

        while True:
            turn_metadata: dict[str, list[Any]] = {}
            try:
                # Pass turn info to step() for real-time logging
                assistant_message, tool_messages, finish_params = await self.step(
                    msgs,
                    turn_metadata,
                    turn=_get_turn_count(full_msg_history, msgs) + 1,
                    max_turns=self._max_turns,
                )
                break
            except ContextOverflowError:
                msgs, dropped_turn_id = self._unwind_context_overflow(msgs)
                run_metadata_by_turn.pop(dropped_turn_id, None)

        # Update cumulative stats
        run_metadata_by_turn[assistant_message.id] = turn_metadata
        accepted_turn = _get_turn_count(full_msg_history, msgs) + 1
        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(accepted_turn, 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 accepted_turn != self._max_turns:
            self._logger.context_summarization_start(pct_context_used, self._context_summarization_cutoff)
            messages_to_summarize, msgs = await self.summarize_messages(msgs, run_metadata_by_turn)
            full_msg_history.append(messages_to_summarize)

        # Avoid successive assistant messages (only if next turn won't show turns remaining)
        next_turn_will_show_warning = self._max_turns - accepted_turn <= self._turns_remaining_warning_threshold
        if (
            self._block_successive_assistant_messages
            and not tool_messages
            and not user_messages
            and not next_turn_will_show_warning
        ):
            msgs.extend([UserMessage(content="Please continue the task")])

    if finish_params is None:
        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, tool durations, and model speed to run_metadata
    run_metadata = _merge_run_metadata(run_metadata_by_turn)
    run_metadata["token_usage"] = _get_total_token_usage(full_msg_history)
    run_metadata["_tool_durations"] = _get_tool_durations(full_msg_history)  # type: ignore[assignment]
    run_metadata["_model_speed"] = _get_model_speed_stats(full_msg_history, self._client.model_slug)  # type: ignore[assignment]
    # 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]
    )

SessionAgent

SessionAgent(
    client: LLMClient,
    name: str,
    *,
    max_turns: int = ...,
    system_prompt: str | None = ...,
    tools: list[Tool | ToolProvider] | None = ...,
    finish_tool: None = None,
    context_summarization_cutoff: float = ...,
    turns_remaining_warning_threshold: int = ...,
    run_sync_in_thread: bool = ...,
    text_only_tool_responses: bool = ...,
    block_successive_assistant_messages: bool = ...,
    recover_from_context_overflow: bool = ...,
    share_parent_exec_env: bool = ...,
    logger: AgentLoggerBase | None = ...,
)

SessionAgent(
    client: LLMClient,
    name: str,
    *,
    max_turns: int = ...,
    system_prompt: str | None = ...,
    tools: list[Tool | ToolProvider] | None = ...,
    finish_tool: Tool[FinishParams, FinishMeta],
    context_summarization_cutoff: float = ...,
    turns_remaining_warning_threshold: int = ...,
    run_sync_in_thread: bool = ...,
    text_only_tool_responses: bool = ...,
    block_successive_assistant_messages: bool = ...,
    recover_from_context_overflow: bool = ...,
    share_parent_exec_env: bool = ...,
    logger: AgentLoggerBase | None = ...,
)

SessionAgent(
    client: LLMClient,
    name: str,
    *,
    max_turns: int = ...,
    system_prompt: str | None = ...,
    tools: list[Tool | ToolProvider] | None = ...,
    finish_tool: list[Tool],
    context_summarization_cutoff: float = ...,
    turns_remaining_warning_threshold: int = ...,
    run_sync_in_thread: bool = ...,
    text_only_tool_responses: bool = ...,
    block_successive_assistant_messages: bool = ...,
    recover_from_context_overflow: bool = ...,
    share_parent_exec_env: bool = ...,
    logger: AgentLoggerBase | None = ...,
)

SessionAgent(
    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]
    | list[Tool]
    | 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,
    block_successive_assistant_messages: bool = True,
    recover_from_context_overflow: bool = True,
    share_parent_exec_env: bool = False,
    logger: AgentLoggerBase | None = None,
)

Bases: Agent[FinishParams, FinishMeta]

Agent running inside an active session with full tool access.

Returned by async with agent.session(...) as session. A SessionAgent shares its __dict__ with the parent Agent, so it has the same configuration and state. The key difference is that ToolProvider-created tools have been initialized and are available in _active_tools.

Agent.run() raises RuntimeError when ToolProviders are present but the caller is not a SessionAgent. This makes the invalid state (calling run() with uninitialized ToolProviders) a type-level distinction rather than just a runtime flag.

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] | list[Tool] | 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,
    block_successive_assistant_messages: bool = True,
    recover_from_context_overflow: bool = True,
    # Subagent options
    share_parent_exec_env: bool = False,
    # 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 or list of Tools used to signal task completion.
                     Defaults to SIMPLE_FINISH_TOOL. If a list is provided,
                     a successful call to any listed tool ends the run.
        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
        block_successive_assistant_messages: If True (default), automatically inject a continue
                                           message when assistant responds without tool calls to
                                           prevent successive assistant messages.
        recover_from_context_overflow: If True (default), drop one completed turn and retry when
                                       the model reports a context overflow. If the original
                                       prompt still overflows, the context error is raised.
        share_parent_exec_env: When True and used as a subagent, share the parent's code
                               execution environment instead of creating a new one. This
                               provides better performance (no file copying) and allows
                               the subagent to see all files in the parent's environment.
                               Only effective when the agent is used as a subagent via to_tool().
        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_tools: dict[str, Tool[Any, Any]] = _normalize_finish_tools(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
    self._block_successive_assistant_messages = block_successive_assistant_messages
    self._recover_from_context_overflow = recover_from_context_overflow
    self._share_parent_exec_env = share_parent_exec_env

    # 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

    # Eagerly register static tools + finish tools (available without a session)
    self._active_tools: dict[str, Tool] = {}
    for tool in self._tools:
        if isinstance(tool, Tool):
            if tool.name in self._finish_tools:
                raise ValueError(
                    f"Tool name '{tool.name}' in `tools` collides with a finish tool. "
                    "Tool names must be unique across `tools` and `finish_tool`."
                )
            self._active_tools[tool.name] = tool
    self._active_tools.update(self._finish_tools)
    self._has_tool_providers = any(isinstance(t, ToolProvider) for t in self._tools)

    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

from_agent `classmethod`

from_agent(
    agent: Agent[FinishParams, FinishMeta],
) -> SessionAgent[FinishParams, FinishMeta]

Create a SessionAgent sharing the given Agent's __dict__.

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

@classmethod
def from_agent(cls, agent: Agent[FinishParams, FinishMeta]) -> "SessionAgent[FinishParams, FinishMeta]":
    """Create a SessionAgent sharing the given Agent's ``__dict__``."""
    sa: SessionAgent[FinishParams, FinishMeta] = object.__new__(cls)
    sa.__dict__ = agent.__dict__
    return sa

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,
) -> TurnWarningMessage

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

_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

_normalize_finish_tools

_normalize_finish_tools(
    finish_tool: Tool[FinishParams, FinishMeta]
    | list[Tool]
    | None,
) -> dict[str, Tool[Any, Any]]

Normalize a single finish tool or list of finish tools into a name-keyed mapping.

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

def _normalize_finish_tools[FinishParams: BaseModel, FinishMeta](
    finish_tool: Tool[FinishParams, FinishMeta] | list[Tool] | None,
) -> dict[str, Tool[Any, Any]]:
    """Normalize a single finish tool or list of finish tools into a name-keyed mapping."""
    finish_tools: list[Tool]
    if finish_tool is None:
        finish_tools = [SIMPLE_FINISH_TOOL]
    elif isinstance(finish_tool, list):
        if not finish_tool:
            raise ValueError("finish_tool list cannot be empty")
        finish_tools = finish_tool
    else:
        finish_tools = [finish_tool]

    seen_names: set[str] = set()
    duplicate_names: set[str] = set()
    for tool in finish_tools:
        if tool.name in seen_names:
            duplicate_names.add(tool.name)
        seen_names.add(tool.name)
    if duplicate_names:
        raise ValueError(f"Finish tools must have unique names, found duplicates: {sorted(duplicate_names)}")

    return {tool.name: tool for tool in finish_tools}

_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)]

_get_tool_durations

_get_tool_durations(
    messages: list[list[ChatMessage]],
) -> dict[str, list[float]]

Collect tool execution durations grouped by tool name from message history.

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

def _get_tool_durations(messages: list[list[ChatMessage]]) -> dict[str, list[float]]:
    """Collect tool execution durations grouped by tool name from message history."""
    durations: dict[str, list[float]] = {}
    for msg in chain.from_iterable(messages):
        if isinstance(msg, ToolMessage) and msg.name and msg.tool_duration is not None:
            durations.setdefault(msg.name, []).append(msg.tool_duration)
    return durations

_get_turn_count

_get_turn_count(
    full_msg_history: list[list[ChatMessage]],
    messages: list[ChatMessage],
) -> int

Count accepted assistant turns still present in history.

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

def _get_turn_count(full_msg_history: list[list[ChatMessage]], messages: list[ChatMessage]) -> int:
    """Count accepted assistant turns still present in history."""
    all_messages = chain(chain.from_iterable(full_msg_history), messages)
    return sum(1 for msg in all_messages if isinstance(msg, AssistantMessage))

_merge_run_metadata

_merge_run_metadata(
    run_metadata_by_turn: dict[str, dict[str, list[Any]]],
) -> dict[str, list[Any]]

Merge per-turn metadata into the public run_metadata shape.

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

def _merge_run_metadata(run_metadata_by_turn: dict[str, dict[str, list[Any]]]) -> dict[str, list[Any]]:
    """Merge per-turn metadata into the public run_metadata shape."""
    run_metadata: dict[str, list[Any]] = {}
    for turn_metadata in run_metadata_by_turn.values():
        for name, metadata_list in turn_metadata.items():
            run_metadata.setdefault(name, []).extend(metadata_list)
    return run_metadata

_get_model_speed_stats

_get_model_speed_stats(
    messages: list[list[ChatMessage]], model_slug: str
) -> dict[str, float | int | str]

Compute speed stats for this agent's model from AssistantMessages.

Returns a flat dict with model_slug, num_calls, output_tokens, duration, e2e_otps. Returns empty dict if no timed messages found.

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

def _get_model_speed_stats(messages: list[list[ChatMessage]], model_slug: str) -> dict[str, float | int | str]:
    """Compute speed stats for this agent's model from AssistantMessages.

    Returns a flat dict with model_slug, num_calls, output_tokens, duration, e2e_otps.
    Returns empty dict if no timed messages found.
    """
    num_calls = 0
    output_tokens = 0
    duration = 0.0
    for msg in chain.from_iterable(messages):
        if not isinstance(msg, AssistantMessage):
            continue
        if msg.request_start_time is None or msg.request_end_time is None:
            continue
        msg_duration = msg.request_end_time - msg.request_start_time
        if msg_duration <= 0:
            continue
        num_calls += 1
        output_tokens += msg.token_usage.output
        duration += msg_duration
    if num_calls == 0:
        return {}
    return {
        "model_slug": model_slug,
        "num_calls": num_calls,
        "output_tokens": output_tokens,
        "duration": duration,
        "e2e_otps": output_tokens / duration if duration > 0 else 0.0,
    }

Agent

stirrup.core.agent

AGENT_MAX_TURNS module-attribute

CONTEXT_SUMMARIZATION_CUTOFF module-attribute

TURNS_REMAINING_WARNING_THRESHOLD module-attribute

MESSAGE_SUMMARIZER module-attribute

MESSAGE_SUMMARIZER_BRIDGE_TEMPLATE module-attribute

DEFAULT_TOOLS module-attribute

SIMPLE_FINISH_TOOL module-attribute

_PARENT_DEPTH module-attribute

logger module-attribute

_SESSION_STATE module-attribute

__all__ module-attribute

LOGGER module-attribute

DEFAULT_SUB_AGENT_DESCRIPTION module-attribute

AGENT_NAME_PATTERN module-attribute

ChatMessage

CacheManager

save_state

load_state

restore_files

clear_cache

list_caches

get_cache_info

CacheState dataclass

msgs instance-attribute

full_msg_history instance-attribute

task_hash instance-attribute

timestamp class-attribute instance-attribute

agent_name class-attribute instance-attribute

run_metadata_by_turn class-attribute instance-attribute

to_dict

from_dict classmethod

ContextOverflowError

AssistantMessage

e2e_otps property

ImageContentBlock

to_base64_url

LLMClient

SubAgentMetadata

__add__

SummaryMessage

SystemMessage

TokenUsage

output property

total property

__add__

Tool

ToolCall

ToolMessage

tool_duration property

ToolProvider

__aenter__ abstractmethod async

__aexit__ async

ToolResult

ToolUseCountMetadata

TurnWarningMessage

UserMessage

SkillMetadata dataclass

CodeExecToolProvider

temp_dir property

__aenter__ abstractmethod async

__aexit__ abstractmethod async

run_command abstractmethod async

read_file_bytes abstractmethod async

write_file_bytes abstractmethod async

file_exists abstractmethod async

is_directory abstractmethod async

list_files abstractmethod async

save_output_files async

upload_files async

get_code_exec_tool

get_view_image_tool

view_image abstractmethod async

LocalCodeExecToolProvider

temp_dir property

__aenter__ async

__aexit__ async

read_file_bytes async

write_file_bytes async

AGENT_MAX_TURNS `module-attribute`

CONTEXT_SUMMARIZATION_CUTOFF `module-attribute`

TURNS_REMAINING_WARNING_THRESHOLD `module-attribute`

MESSAGE_SUMMARIZER `module-attribute`

MESSAGE_SUMMARIZER_BRIDGE_TEMPLATE `module-attribute`

DEFAULT_TOOLS `module-attribute`

SIMPLE_FINISH_TOOL `module-attribute`

_PARENT_DEPTH `module-attribute`

logger `module-attribute`

_SESSION_STATE `module-attribute`

all `module-attribute`

LOGGER `module-attribute`

DEFAULT_SUB_AGENT_DESCRIPTION `module-attribute`

AGENT_NAME_PATTERN `module-attribute`

CacheState `dataclass`

msgs `instance-attribute`

full_msg_history `instance-attribute`

task_hash `instance-attribute`

timestamp `class-attribute` `instance-attribute`

agent_name `class-attribute` `instance-attribute`

run_metadata_by_turn `class-attribute` `instance-attribute`

from_dict `classmethod`

e2e_otps `property`

add

output `property`

total `property`

add

tool_duration `property`

aenter `abstractmethod` `async`

aexit `async`

SkillMetadata `dataclass`

temp_dir `property`

aenter `abstractmethod` `async`

aexit `abstractmethod` `async`

run_command `abstractmethod` `async`

read_file_bytes `abstractmethod` `async`

write_file_bytes `abstractmethod` `async`

file_exists `abstractmethod` `async`

is_directory `abstractmethod` `async`

list_files `abstractmethod` `async`

save_output_files `async`

upload_files `async`

view_image `abstractmethod` `async`

temp_dir `property`

aenter `async`

aexit `async`

read_file_bytes `async`

write_file_bytes `async`

file_exists `async`

is_directory `async`

list_files `async`

run_command `async`

save_output_files `async`

upload_files `async`

view_image `async`

enter

exit

enter `abstractmethod`

exit `abstractmethod`

on_step `abstractmethod`

assistant_message `abstractmethod`

user_message `abstractmethod`

task_message `abstractmethod`

tool_result `abstractmethod`

context_summarization_start `abstractmethod`

context_summarization_complete `abstractmethod`

debug `abstractmethod`

info `abstractmethod`

warning `abstractmethod`

error `abstractmethod`

SessionState `dataclass`

name `property`

client `property`

tools `property`

finish_tool `property`

finish_tools `property`

logger `property`

aenter `async`

aexit `async`

run_tool `async`