Client

client

OpenShell gRPC client wrapper for Shoreguard.

ShoreGuardClient

ShoreGuardClient(
    endpoint: str,
    *,
    ca_path: Path | None = None,
    cert_path: Path | None = None,
    key_path: Path | None = None,
    timeout: float = 30.0,
)

Unified client for OpenShell gateway operations.

Parameters:

Name	Type	Description	Default
endpoint	str	gRPC endpoint address (host:port).	required
ca_path	Path | None	Path to CA certificate for TLS.	None
cert_path	Path | None	Path to client certificate for mTLS.	None
key_path	Path | None	Path to client private key for mTLS.	None
timeout	float	Default gRPC call timeout in seconds.	30.0

from_credentials classmethod

from_credentials(
    endpoint: str,
    *,
    ca_cert: bytes | None = None,
    client_cert: bytes | None = None,
    client_key: bytes | None = None,
    timeout: float = 30.0,
) -> ShoreGuardClient

Connect using raw certificate bytes (from DB or registry).

Parameters:

Name	Type	Description	Default
endpoint	str	gRPC endpoint address (host:port).	required
ca_cert	bytes | None	CA certificate bytes for TLS.	None
client_cert	bytes | None	Client certificate bytes for mTLS.	None
client_key	bytes | None	Client private key bytes for mTLS.	None
timeout	float	Default gRPC call timeout in seconds.	30.0

Returns:

Name	Type	Description
ShoreGuardClient	ShoreGuardClient	Connected client instance.

from_active_cluster classmethod

from_active_cluster(
    *, cluster: str | None = None, timeout: float = 30.0
) -> ShoreGuardClient

Connect to the active OpenShell gateway using mTLS credentials.

Parameters:

Name	Type	Description	Default
cluster	str | None	Cluster name override. Defaults to the active gateway.	None
timeout	float	Default gRPC call timeout in seconds.	30.0

Returns:

Name	Type	Description
ShoreGuardClient	ShoreGuardClient	Connected client instance.

Raises:

Type	Description
GatewayNotConnectedError	If gateway metadata is missing or invalid.

health

health() -> dict

Check gateway health.

Returns:

Name	Type	Description
dict	dict	Status and version of the gateway.

get_cluster_inference

get_cluster_inference(*, route_name: str = '') -> dict

Get current cluster inference configuration.

Parameters:

Name	Type	Description	Default
route_name	str	Optional route name to filter by.	''

Returns:

Name	Type	Description
dict	dict	Inference configuration with provider, model, and route.

get_gateway_config

get_gateway_config() -> dict

Get the global gateway configuration (settings and revision).

Returns:

Name	Type	Description
dict	dict	Settings map and settings revision number.

set_cluster_inference

set_cluster_inference(
    *,
    provider_name: str,
    model_id: str,
    verify: bool = True,
    route_name: str = "",
    timeout_secs: int = 0,
) -> dict

Set cluster inference configuration.

Parameters:

Name	Type	Description	Default
provider_name	str	Name of the inference provider.	required
model_id	str	Model identifier to use.	required
verify	bool	Whether to validate endpoints before saving.	True
route_name	str	Optional route name to configure.	''
timeout_secs	int	Per-route request timeout in seconds (0 = default 60s).	0

Returns:

Name	Type	Description
dict	dict	Updated inference configuration with validation results.

close

close() -> None

Close the underlying gRPC channel.

__enter__

__enter__() -> ShoreGuardClient

Support usage as a context manager.

Returns:

Name	Type	Description
ShoreGuardClient	ShoreGuardClient	This client instance.

__exit__

__exit__(*args: object) -> None

Close the channel on context exit.

Parameters:

Name	Type	Description	Default
*args	object	Exception info (exc_type, exc_val, exc_tb).	()

sandboxes

Sandbox CRUD and execution operations.

SandboxManager

SandboxManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Sandbox CRUD, execution, and lifecycle operations.

Parameters:

Name	Type	Description	Default
stub	OpenShellStub	OpenShell gRPC stub.	required
timeout	float	gRPC call timeout in seconds.	30.0

list

list(
    *, limit: int = 100, offset: int = 0
) -> list[dict[str, Any]]

List all sandboxes.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of sandboxes to return.	100
offset	int	Pagination offset.	0

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of sandbox dicts.

get

get(name: str) -> dict[str, Any]

Get a sandbox by name.

Parameters:

Name	Type	Description	Default
name	str	Sandbox name.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Sandbox data dict.

create

create(
    *,
    name: str = "",
    image: str = "",
    policy: dict | None = None,
    providers: list[str] | None = None,
    gpu: bool = False,
    environment: dict[str, str] | None = None,
) -> dict[str, Any]

Create a new sandbox.

Parameters:

Name	Type	Description	Default
name	str	Sandbox name.	''
image	str	Container image to use.	''
policy	dict | None	Optional sandbox policy dict.	None
providers	list[str] | None	Optional list of provider names.	None
gpu	bool	Whether to request GPU resources.	False
environment	dict[str, str] | None	Optional environment variable key-value pairs.	None

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Created sandbox data dict.

delete

delete(name: str) -> bool

Delete a sandbox by name.

Parameters:

Name	Type	Description	Default
name	str	Sandbox name.	required

Returns:

Name	Type	Description
bool	bool	True if the sandbox was deleted.

wait_ready

wait_ready(
    name: str, *, timeout_seconds: float = 300.0
) -> dict[str, Any]

Block until a sandbox reaches READY phase.

Parameters:

Name	Type	Description	Default
name	str	Sandbox name.	required
timeout_seconds	float	Maximum time to wait in seconds.	300.0

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Sandbox data dict once ready.

Raises:

Type	Description
SandboxError	If the sandbox enters an error phase.
TimeoutError	If the sandbox is not ready within the timeout.

exec

exec(
    sandbox_id: str,
    command: list[str],
    *,
    workdir: str = "",
    env: dict[str, str] | None = None,
    timeout_seconds: int = 0,
) -> dict[str, Any]

Execute a command in a sandbox and return the result.

Parameters:

Name	Type	Description	Default
sandbox_id	str	Sandbox identifier.	required
command	list[str]	Command and arguments to execute.	required
workdir	str	Working directory inside the sandbox.	''
env	dict[str, str] | None	Optional environment variables for the command.	None
timeout_seconds	int	Command timeout in seconds (0 for default).	0

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Execution result with exit_code, stdout, and stderr.

create_ssh_session

create_ssh_session(sandbox_id: str) -> dict[str, Any]

Create a temporary SSH session for shell access to a sandbox.

Parameters:

Name	Type	Description	Default
sandbox_id	str	Sandbox identifier.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: SSH session details including token and gateway connection info.

revoke_ssh_session

revoke_ssh_session(token: str) -> bool

Revoke an active SSH session.

Parameters:

Name	Type	Description	Default
token	str	Session token to revoke.	required

Returns:

Name	Type	Description
bool	bool	True if the session was revoked.

get_logs

get_logs(
    sandbox_id: str,
    *,
    lines: int = 200,
    since_ms: int = 0,
    sources: list[str] | None = None,
    min_level: str = "",
) -> list[dict[str, Any]]

Fetch recent sandbox logs.

Parameters:

Name	Type	Description	Default
sandbox_id	str	Sandbox identifier.	required
lines	int	Maximum number of log lines to return.	200
since_ms	int	Only return logs after this timestamp (ms).	0
sources	list[str] | None	Optional list of log sources to filter by.	None
min_level	str	Minimum log level to include.	''

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of log entry dicts.

watch

watch(
    sandbox_id: str,
    *,
    follow_status: bool = True,
    follow_logs: bool = True,
    follow_events: bool = True,
    log_tail_lines: int = 50,
) -> Iterator[dict[str, Any]]

Stream live sandbox events (status, logs, platform events, draft updates).

Parameters:

Name	Type	Description	Default
sandbox_id	str	Sandbox identifier.	required
follow_status	bool	Subscribe to status changes.	True
follow_logs	bool	Subscribe to log output.	True
follow_events	bool	Subscribe to platform events.	True
log_tail_lines	int	Number of existing log lines to replay.	50

Yields:

Type	Description
dict[str, Any]	dict[str, Any]: Event dict with type and data keys.

policies

Policy management operations.

PolicyManager

PolicyManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Policy read/write operations against OpenShell gateway.

Parameters:

Name	Type	Description	Default
stub	OpenShellStub	OpenShell gRPC stub.	required
timeout	float	gRPC call timeout in seconds.	30.0

get

get(sandbox_name: str) -> dict[str, Any]

Get the current active policy for a sandbox.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Active policy status with revision details.

get_version

get_version(
    sandbox_name: str, version: int
) -> dict[str, Any]

Get a specific policy revision by version number.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
version	int	Policy version number to retrieve.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Policy status with revision details.

list_revisions

list_revisions(
    sandbox_name: str, *, limit: int = 20, offset: int = 0
) -> list[dict[str, Any]]

List policy revision history for a sandbox.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
limit	int	Maximum number of revisions to return.	20
offset	int	Pagination offset.	0

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of policy revision summary dicts.

update

update(
    sandbox_name: str,
    policy: SandboxPolicy,
    *,
    global_scope: bool = False,
) -> dict[str, Any]

Push a new policy version to a sandbox (or globally).

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
policy	SandboxPolicy	SandboxPolicy protobuf message.	required
global_scope	bool	If True, apply policy globally.	False

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Version and policy hash of the new revision.

approvals

Draft policy approval flow operations.

ApprovalManager

ApprovalManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Draft policy approval flow: review, approve, reject blocked requests.

Parameters:

Name	Type	Description	Default
stub	OpenShellStub	OpenShell gRPC stub.	required
timeout	float	gRPC call timeout in seconds.	30.0

get_draft

get_draft(
    sandbox_name: str, *, status_filter: str = ""
) -> dict[str, Any]

Get draft policy recommendations for a sandbox.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
status_filter	str	Optional status to filter chunks by.	''

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Draft policy with chunks, summary, and version.

get_pending

get_pending(sandbox_name: str) -> list[dict[str, Any]]

Get only pending (unapproved) draft chunks.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of pending chunk dicts.

approve

approve(sandbox_name: str, chunk_id: str) -> dict[str, Any]

Approve a single draft policy chunk (merges into active policy).

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
chunk_id	str	Chunk identifier to approve.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: New policy version and hash.

reject

reject(
    sandbox_name: str, chunk_id: str, *, reason: str = ""
) -> None

Reject a single draft policy chunk.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
chunk_id	str	Chunk identifier to reject.	required
reason	str	Optional rejection reason.	''

approve_all

approve_all(
    sandbox_name: str,
    *,
    include_security_flagged: bool = False,
) -> dict[str, Any]

Approve all pending draft chunks.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
include_security_flagged	bool	Whether to also approve security-flagged chunks.	False

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Policy version, hash, and approval counts.

edit

edit(
    sandbox_name: str, chunk_id: str, proposed_rule: dict
) -> None

Edit a pending draft chunk in-place.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
chunk_id	str	Chunk identifier to edit.	required
proposed_rule	dict	Network rule dict to replace the existing rule.	required

undo

undo(sandbox_name: str, chunk_id: str) -> dict[str, Any]

Reverse an approval (remove merged rule from active policy).

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required
chunk_id	str	Chunk identifier to undo.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Updated policy version and hash.

clear

clear(sandbox_name: str) -> dict[str, int]

Clear all pending draft chunks for a sandbox.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required

Returns:

Type	Description
dict[str, int]	dict[str, int]: Number of chunks cleared.

get_history

get_history(sandbox_name: str) -> list[dict[str, Any]]

Get decision history for a sandbox's draft policy.

Parameters:

Name	Type	Description	Default
sandbox_name	str	Sandbox name.	required

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of history entry dicts.

providers

Provider management operations.

ProviderManager

ProviderManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Provider CRUD operations against OpenShell gateway.

Parameters:

Name	Type	Description	Default
stub	OpenShellStub	OpenShell gRPC stub.	required
timeout	float	gRPC call timeout in seconds.	30.0

list

list(
    *, limit: int = 100, offset: int = 0
) -> list[dict[str, Any]]

List all providers.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of providers to return.	100
offset	int	Pagination offset.	0

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of provider dicts.

get

get(name: str) -> dict[str, Any]

Get a provider by name.

Parameters:

Name	Type	Description	Default
name	str	Provider name.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Provider data dict.

create

create(
    *,
    name: str,
    provider_type: str,
    credentials: dict[str, str] | None = None,
    config: dict[str, str] | None = None,
) -> dict[str, Any]

Create a new provider.

Parameters:

Name	Type	Description	Default
name	str	Provider name.	required
provider_type	str	Provider type identifier.	required
credentials	dict[str, str] | None	Provider credential key-value pairs.	None
config	dict[str, str] | None	Provider configuration key-value pairs.	None

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Created provider data dict.

update

update(
    *,
    name: str,
    provider_type: str = "",
    credentials: dict[str, str] | None = None,
    config: dict[str, str] | None = None,
) -> dict[str, Any]

Update an existing provider.

Parameters:

Name	Type	Description	Default
name	str	Provider name.	required
provider_type	str	Provider type identifier.	''
credentials	dict[str, str] | None	Provider credential key-value pairs.	None
config	dict[str, str] | None	Provider configuration key-value pairs.	None

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Updated provider data dict.

delete

delete(name: str) -> bool

Delete a provider by name.

Parameters:

Name	Type	Description	Default
name	str	Provider name.	required

Returns:

Name	Type	Description
bool	bool	True if the provider was deleted.