Skip to content

Python API

Shoal is primarily a CLI, but a few modules already act as stable integration seams for contributors and automation work. This page renders the current docstrings directly from the source tree.

Configuration helpers

config

Configuration loading and XDG path helpers.

ConfigLoadError

ConfigLoadError(path: Path | str, detail: str)

Bases: Exception

User-friendly error for malformed or invalid config files.

soul_text cached

soul_text() -> str

Return SOUL.md content, or empty string if not found.

config_dir

config_dir() -> Path

Return Shoal config directory.

Reads XDG_CONFIG_HOME env var, falling back to ~/.config/shoal.

data_dir

data_dir() -> Path

Return Shoal persistent data directory (sessions, robo state).

Reads XDG_DATA_HOME env var, falling back to ~/.local/share/shoal.

state_dir

state_dir() -> Path

Return Shoal transient state directory (PIDs, logs).

Reads XDG_STATE_HOME env var, falling back to ~/.local/state/shoal.

ensure_dirs

ensure_dirs() -> None

Create all required data and state directories.

scaffold_defaults

scaffold_defaults() -> list[str]

Copy bundled example configs into the user's config dir.

Only writes files that do not already exist — never overwrites. Returns a list of relative paths that were created.

load_config cached

load_config() -> ShoalConfig

Load and cache the main config.toml.

load_tool_config

load_tool_config(name: str) -> ToolConfig

Load a tool config, flattening [tool] + [detection] + [mcp] sections.

load_robo_profile

load_robo_profile(name: str) -> RoboProfileConfig

Load a robo profile TOML.

available_tools

available_tools() -> list[str]

List available tool names from config/tools/*.toml.

templates_dir

templates_dir() -> Path

Return ~/.config/shoal/templates.

fins_dir

fins_dir() -> Path

Return ~/.config/shoal/fins — installed fins registry.

project_templates_dir

project_templates_dir() -> Path | None

Return <git-root>/.shoal/templates if inside a git repo and the dir exists.

available_templates

available_templates() -> list[str]

List available template names from local and global template dirs.

template_source

template_source(name: str) -> str

Return 'local' if the template exists in project-local dir, else 'global'.

load_mcp_registry

load_mcp_registry() -> dict[str, str]

Load MCP server registry: user file merged over built-in defaults.

Reads ~/.config/shoal/mcp-servers.toml. Each top-level key is a server name whose value is a table with a command key. Built-in defaults are used as a fallback for servers not overridden by the user.

Returns:

Type Description
dict[str, str]

Mapping of server name → command string.

load_mcp_registry_full

load_mcp_registry_full() -> dict[str, dict[str, str]]

Load the full MCP server registry with all fields per entry.

Seeds with built-in defaults, then merges user overrides. Returns raw dicts so callers can read transport and other fields.

mixins_dir

mixins_dir() -> Path

Return ~/.config/shoal/templates/mixins.

available_mixins

available_mixins() -> list[str]

List available mixin names from local and global mixin dirs.

resolve_template

resolve_template(
    name: str, _chain: set[str] | None = None
) -> SessionTemplateConfig

Load and fully resolve a template: extends -> mixins -> final.

Raises ValueError on inheritance cycles or unknown mixins.

load_mixin

load_mixin(name: str) -> TemplateMixinConfig

Load a template mixin TOML from local or global mixins dir.

load_template

load_template(name: str) -> SessionTemplateConfig

Load a session template TOML with full inheritance resolution.

refresh_tools

refresh_tools() -> list[str]

Re-copy bundled tool profiles to the user's config dir, overwriting existing files.

Unlike scaffold_defaults(), this always overwrites. Use it when bundled tool profiles have been updated and the user wants to pull in the latest version.

Returns:

Type Description
list[str]

Sorted list of relative filenames that were refreshed (e.g. ["omp.toml", "pi.toml"]).

list[str]

Returns an empty list and logs a warning if the bundled source dir is missing.

load_project_config

load_project_config(git_root: str) -> ProjectConfig | None

Load project-level config from <git_root>/.shoal.toml.

Returns None if the file does not exist.

discover_skills

discover_skills(
    git_root: str | None = None,
) -> list[SkillConfig]

Discover skills from project-local and global paths.

Search order: 1. <git_root>/.shoal/skills/*/SKILL.md (project-local) 2. ~/.config/shoal/skills/*/SKILL.md (global)

Returns parsed SkillConfig for each valid skill found. Duplicates (same name) resolved by local-wins.

load_workspace_config

load_workspace_config(
    git_root: str,
) -> WorkspaceConfig | None

Load workspace manifest from <git_root>/.shoal/workspace.toml.

Returns None if the file does not exist. Raises ConfigLoadError on parse or validation errors so the caller can surface them.

load_project_hooks

load_project_hooks() -> list[ProjectHookEntry]

Load project-local lifecycle hooks from .shoal/hooks.toml.

Returns an empty list if the file does not exist or the git root cannot be determined. Validation errors are logged and that entry is skipped.

Session state

state

Session state CRUD — all state stored in SQLite.

generate_id

generate_id(length: int = 8) -> str

Generate a short unique session ID from [a-z0-9].

build_nvim_socket_path

build_nvim_socket_path(
    tmux_session_id: str, tmux_window_id: str
) -> str

Build Neovim socket path from tmux IDs.

Uses XDG_RUNTIME_DIR if set, falling back to /tmp.

resolve_nvim_socket async

resolve_nvim_socket(session: SessionState) -> str | None

Resolve and persist a session's Neovim socket from current tmux IDs.

create_session async

create_session(
    name: str,
    tool: str,
    git_root: str,
    worktree: str = "",
    branch: str = "",
    parent_id: str = "",
    tags: list[str] | None = None,
    template_name: str = "",
    model: str = "",
) -> SessionState

Create a new session state in DB and return the session.

Raises:

Type Description
ValueError

If session name validation fails or tmux name collision detected.

get_session async

get_session(session_id: str) -> SessionState | None

Read a session from DB, or None if not found.

get_sessions async

get_sessions(
    session_ids: Iterable[str],
) -> dict[str, SessionState]

Read multiple sessions keyed by session ID.

update_session async

update_session(
    session_id: str, **fields: Any
) -> SessionState | None

Update specific fields on a session in DB.

Raises:

Type Description
ValueError

If name field validation fails.

delete_session async

delete_session(session_id: str) -> bool

Delete a session from DB.

list_sessions async

list_sessions() -> list[SessionState]

Return all sessions.

find_by_name async

find_by_name(name: str) -> str | None

Find a session ID by name.

find_sessions_by_names async

find_sessions_by_names(
    names: Iterable[str],
) -> dict[str, SessionState]

Find multiple sessions keyed by session name.

touch_session async

touch_session(session_id: str) -> None

Update last_activity timestamp.

add_mcp_to_session async

add_mcp_to_session(session_id: str, mcp_name: str) -> None

Add an MCP server to a session's list.

remove_mcp_from_session async

remove_mcp_from_session(
    session_id: str, mcp_name: str
) -> None

Remove an MCP server from a session's list.

add_tag async

add_tag(session_id: str, tag: str) -> None

Add a tag to a session (no-op if already present).

remove_tag async

remove_tag(session_id: str, tag: str) -> None

Remove a tag from a session (no-op if not present).

resolve_session async

resolve_session(name_or_id: str) -> str | None

Resolve a name or ID to a session ID. Returns None if not found.

resolve_sessions async

resolve_sessions(
    names_or_ids: Iterable[str],
) -> dict[str, str | None]

Resolve many session names or IDs while preserving single-item semantics.

load_sessions async

load_sessions(
    names_or_ids: Iterable[str],
) -> dict[str, SessionState | None]

Load many sessions keyed by the original reference string.

resolve_session_interactive

resolve_session_interactive(
    name_or_id: str | None = None,
) -> str

Resolve session with fzf fallback. Raises SystemExit on failure.

filter_sessions_by_path

filter_sessions_by_path(
    sessions: Sequence[SessionState], path: str
) -> list[SessionState]

Filter sessions whose git root or worktree matches a repo path.

A session matches when its path equals the resolved filter path, or its worktree falls under it (starts with resolved + "/").

MCP orchestration server

mcp_shoal_server

Shoal MCP server — exposes session orchestration as MCP tools.

Runs as a stdio process, spawned per connection by the MCP pool. AI agents (especially robo supervisors) use these tools to manage sessions natively via the MCP protocol.

Requires the mcp optional dependency: pip install shoal[mcp]

list_sessions_tool async

list_sessions_tool(
    path: str | None = None,
) -> list[dict[str, Any]]

List all active Shoal sessions, optionally filtered to a repo path.

session_status_tool async

session_status_tool(
    session: str | list[str] | None = None,
) -> dict[str, object]

Get session status counts or per-session status.

session_info_tool async

session_info_tool(session: str) -> dict[str, object]

Get full details for a session.

send_keys_tool async

send_keys_tool(
    session: str | list[str],
    keys: str,
    enter: bool | None = None,
) -> dict[str, object]

Send keys to a session.

capture_pane_tool async

capture_pane_tool(
    session: str | list[str], lines: int = 20
) -> dict[str, object]

Capture recent terminal output from a session's pane.

read_history_tool async

read_history_tool(
    session: str, limit: int = 50
) -> list[dict[str, object]]

Read status transition history for a session.

session_snapshot_tool async

session_snapshot_tool(
    sessions: list[str],
    fields: list[SnapshotField] | None = None,
    pane_lines: int = 50,
    max_parallelism: int = 8,
) -> dict[str, object]

Capture a read-optimized snapshot for multiple sessions.

batch_execute_tool async

batch_execute_tool(
    ops: list[dict[str, object]],
    continue_on_error: bool = True,
    max_parallelism: int = 8,
) -> dict[str, object]

Execute a heterogeneous batch of Shoal session operations.

create_session_tool async

create_session_tool(
    name: str,
    path: str = ".",
    tool: str | None = None,
    worktree: str | None = None,
    branch: bool = False,
    template: str | None = None,
    mcp_servers: list[str] | None = None,
    prompt: str | None = None,
    model: str | None = None,
) -> dict[str, Any]

Create a new agent session.

Parameters:

Name Type Description Default
name str

Session name (required).

required
path str

Project directory (defaults to current directory).

'.'
tool str | None

AI tool to use (omp, claude, codex, gemini, pi). Defaults to config.

None
worktree str | None

Create a git worktree with this name.

None
branch bool

Create a new branch for the worktree.

False
template str | None

Session template name to apply.

None
mcp_servers list[str] | None

MCP servers to provision (e.g. ["memory", "github"]).

None
prompt str | None

Initial prompt to send to the agent after startup. Enter is pressed automatically.

None
model str | None

Model to use for the session (e.g., "z-ai/glm-5"). Overrides tool default.

None

fork_session_tool async

fork_session_tool(
    source: str,
    name: str,
    prompt: str | None = None,
    template: str | None = None,
    mcp_servers: list[str] | None = None,
) -> dict[str, Any]

Fork a session into a new worker.

Parameters:

Name Type Description Default
source str

Name or ID of the session to fork from.

required
name str

Name for the new worker session.

required
prompt str | None

Initial prompt to send to the worker after startup.

None
template str | None

Template name to apply to the worker session.

None
mcp_servers list[str] | None

MCP servers to provision for the worker.

None

spawn_team_tool async

spawn_team_tool(
    source: str,
    workers: list[dict[str, Any]],
    template: str | None = None,
    mcp_servers: list[str] | None = None,
) -> dict[str, Any]

Spawn multiple worker sessions.

Parameters:

Name Type Description Default
source str

Name or ID of the coordinator session to fork from.

required
workers list[dict[str, Any]]

List of worker specs. Each dict must have 'name' (str) and optionally 'prompt' (str) to send after startup.

required
template str | None

Template to apply to all workers.

None
mcp_servers list[str] | None

MCP servers to provision for all workers.

None

wait_for_team_tool async

wait_for_team_tool(
    correlation_id: str,
    session_names: list[str],
    timeout_seconds: int = 300,
    poll_interval_seconds: int = 10,
) -> dict[str, Any]

Wait for a team of workers to reach terminal states.

Parameters:

Name Type Description Default
correlation_id str

The correlation ID returned by spawn_team.

required
session_names list[str]

Names of the worker sessions to wait for.

required
timeout_seconds int

Maximum seconds to wait (default 300).

300
poll_interval_seconds int

Seconds between status polls (default 10).

10

kill_session_tool async

kill_session_tool(
    session: str,
    remove_worktree: bool = False,
    force: bool = False,
) -> dict[str, object]

Kill a session.

append_journal_tool async

append_journal_tool(
    session: str, entry: str, source: str = "mcp"
) -> dict[str, object]

Append a journal entry for a session.

read_journal_tool async

read_journal_tool(
    session: str, limit: int = 10
) -> list[dict[str, object]]

Read recent journal entries for a session.

session_summary_tool async

session_summary_tool(session: str) -> dict[str, object]

Return the latest Dreamer summary for a session.

send_session_message_tool async

send_session_message_tool(
    to: str,
    topic: str,
    payload: str,
    from_session: str = "",
    kind: str = "event",
    correlation_id: str | None = None,
    reply_to_message_id: int | None = None,
    priority: int = 3,
    requires_ack: bool = False,
    metadata_json: str | None = None,
) -> dict[str, object]

Post a typed message to another session.

receive_session_messages_tool async

receive_session_messages_tool(
    session: str,
    topic: str | None = None,
    kind: str | None = None,
    correlation_id: str | None = None,
    unconsumed_only: bool = True,
    limit: int = 50,
    after_id: int | None = None,
) -> list[dict[str, object]]

Fetch messages for a session.

mark_session_message_consumed_tool async

mark_session_message_consumed_tool(
    message_id: int,
) -> dict[str, object]

Mark a message as consumed.

mark_session_message_acked_tool async

mark_session_message_acked_tool(
    message_id: int,
) -> dict[str, object]

Acknowledge a message.

request_session_action_tool async

request_session_action_tool(
    requester_session: str,
    action_type: str,
    payload_json: str,
    target_session: str | None = None,
    target_role: str | None = None,
    correlation_id: str | None = None,
    metadata_json: str | None = None,
) -> dict[str, object]

Submit an action request.

list_pending_session_actions_tool async

list_pending_session_actions_tool(
    target_session: str | None = None,
    target_role: str | None = None,
    correlation_id: str | None = None,
    limit: int = 50,
) -> list[dict[str, object]]

List pending action requests.

approve_session_action_tool async

approve_session_action_tool(
    action_id: int,
    resolved_by: str,
    reason: str | None = None,
) -> dict[str, object]

Approve an action.

deny_session_action_tool async

deny_session_action_tool(
    action_id: int,
    resolved_by: str,
    reason: str | None = None,
) -> dict[str, object]

Deny an action.

watch_session_messages_tool async

watch_session_messages_tool(
    session: str,
    topic: str | None = None,
    kind: str | None = None,
    correlation_id: str | None = None,
    after_id: int | None = None,
    timeout_seconds: float = 30.0,
) -> list[dict[str, object]]

Block until a matching message arrives or timeout.

get_workflow_messages_tool async

get_workflow_messages_tool(
    correlation_id: str,
    kind: str | None = None,
    limit: int = 50,
    after_id: int | None = None,
) -> list[dict[str, object]]

Return all messages for a workflow correlation ID.

watch_session_actions_tool async

watch_session_actions_tool(
    target_session: str | None = None,
    target_role: str | None = None,
    correlation_id: str | None = None,
    timeout_seconds: float = 30.0,
) -> list[dict[str, object]]

Block until a pending action request appears or timeout.

get_failure_context_tool async

get_failure_context_tool(
    session: str, consume: bool = False
) -> dict[str, object]

Return pending failure context for a session.

wait_for_completion_tool async

wait_for_completion_tool(
    session: str, timeout_seconds: int = 300
) -> dict[str, object]

Poll until session is marked complete or timeout elapses.

branch_status_tool async

branch_status_tool(session: str) -> dict[str, object]

Return git branch status for the session's worktree.

merge_branch_tool async

merge_branch_tool(
    session: str, target: str, strategy: str = "merge"
) -> dict[str, object]

Merge the session's current branch into target.

mark_complete_tool async

mark_complete_tool(
    session: str, summary: str = ""
) -> dict[str, object]

Mark a session as complete with an optional summary.

read_worktree_file_tool async

read_worktree_file_tool(
    session: str, path: str, max_lines: int = 200
) -> dict[str, object]

Read a file from a session's worktree.

list_worktree_files_tool async

list_worktree_files_tool(
    session: str, glob_pattern: str = "*"
) -> dict[str, object]

List files in a session's worktree.

main

main() -> None

Run the Shoal MCP server.

Supports --http [PORT] for streamable-http transport (default: stdio). HTTP mode is used for benchmarking and remote session support.

Batch models

batch

Shared models for Shoal batch execution and session snapshots.

BatchError

Bases: BaseModel

Structured per-item error for batch execution.

BatchItemResult

Bases: BaseModel

Per-item batch execution result envelope.

BatchExecutionRequest

Bases: BaseModel

Mixed-operation batch request.

BatchExecutionResponse

Bases: BaseModel

Mixed-operation batch response.

SessionSnapshotRequest

Bases: BaseModel

Aggregate multi-session read request optimized for supervisors.

SessionSnapshotItem

Bases: BaseModel

Per-session snapshot result envelope.

SessionSnapshotResponse

Bases: BaseModel

Supervisor-friendly multi-session snapshot response.

Batch execution service

batch

Shared application-level batching for Shoal MCP and HTTP APIs.

BatchOperationFailure dataclass

BatchOperationFailure(code: str, message: str)

Bases: Exception

Expected per-item failure captured in the batch envelope.

SessionCache dataclass

SessionCache(
    resolved_by_ref: dict[str, str | None] = dict(),
    sessions_by_id: dict[str, SessionState | None] = dict(),
    refs_by_session: dict[str, set[str]] = dict(),
)

Bulk-aware session resolution cache with invalidation for destructive ops.

execute_batch async

execute_batch(
    request: BatchExecutionRequest,
) -> BatchExecutionResponse

Execute heterogeneous session operations with shared resolution and caching.

session_snapshot async

session_snapshot(
    request: SessionSnapshotRequest,
) -> SessionSnapshotResponse

Capture a supervisor-friendly read snapshot across many sessions.

HTTP server

server

FastAPI server for shoal — exposes session management over HTTP.

SessionCreate

Bases: BaseModel

validate_name classmethod

validate_name(v: str | None) -> str | None

Validate session name if provided.

McpResponse

Bases: BaseModel

MCP server info.

McpCreate

Bases: BaseModel

MCP server creation request.

validate_mcp_name classmethod

validate_mcp_name(v: str) -> str

Validate MCP server name.

SendKeysRequest

Bases: BaseModel

Request to send keys to a session.

RenameRequest

Bases: BaseModel

Request to rename a session.

validate_name classmethod

validate_name(v: str) -> str

Validate new session name.

RequestIdMiddleware

Bases: BaseHTTPMiddleware

Inject a request ID into each request context and response headers.

poll_status_changes async

poll_status_changes() -> None

Background task to broadcast status changes.

health async

health() -> dict[str, object]

Deep health check — DB, watcher, tmux reachability.

batch_execute_api async

batch_execute_api(
    data: BatchExecutionRequest,
) -> BatchExecutionResponse

Execute a heterogeneous application-level batch.

session_snapshot_api async

session_snapshot_api(
    data: SessionSnapshotRequest,
) -> SessionSnapshotResponse

Capture selected fields across multiple sessions in one read-optimized call.

rename_session_api async

rename_session_api(
    session_id: str, body: RenameRequest
) -> SessionResponse

Rename a session.

list_mcp_servers async

list_mcp_servers() -> list[McpResponse]

List all MCP servers in the pool.

list_known_servers async

list_known_servers() -> list[dict[str, str]]

List known MCP server commands (from registry + built-in defaults).

get_mcp_server async

get_mcp_server(name: str) -> McpResponse

Get details of a specific MCP server.

start_mcp_server_api async

start_mcp_server_api(data: McpCreate) -> McpResponse

Start an MCP server in the pool.

stop_mcp_server_api async

stop_mcp_server_api(name: str) -> None

Stop an MCP server and clean up sessions.

attach_mcp_to_session async

attach_mcp_to_session(
    session_id: str, mcp_name: str
) -> dict[str, str | bool]

Attach an MCP server to a session.

detach_mcp_from_session async

detach_mcp_from_session(
    session_id: str, mcp_name: str
) -> None

Detach an MCP server from a session.