steamship.utils package

Submodules

steamship.utils.binary_utils module

steamship.utils.binary_utils.flexi_create(base64string: str = None, data: Any = None, string: str = None, json: Any = None, _bytes: bytes | BytesIO = None, mime_type=None, force_base64=False) → Tuple[Any, None | str, None | str]

It’s convenient for some constructors to accept a variety of input types: - data (your choice) - string - json - bytes

steamship.utils.binary_utils.guess_mime(obj: Any, provided_mime: str = None) → str

steamship.utils.binary_utils.to_b64(obj: Any) → str

steamship.utils.context_length module

steamship.utils.context_length.filter_blocks_for_prompt_length(max_tokens: int, blocks: List[Block]) → List[int]

steamship.utils.context_length.token_length(block: Block, tiktoken_encoder: str = 'p50k_base') → int

Calculate num tokens with tiktoken package.

steamship.utils.file_tags module

steamship.utils.file_tags.update_file_status(client: Steamship, file: File, status: str) → None

steamship.utils.huggingface_helper module

This class is a helper for plugins to use models hosted on Hugging Face.

It uses asyncio parallelism to make many http requests simultaneously.

steamship.utils.huggingface_helper.get_huggingface_results(blocks: List[Block], hf_model_path: str, hf_bearer_token: str, additional_params: dict = None, timeout_seconds: int = 30, use_gpu: bool = False) → List[list]

steamship.utils.kv_store module

A simple key-value store implemented atop Files and Tags.

class steamship.utils.kv_store.KeyValueStore(client: Steamship, store_identifier: str = 'KeyValueStore')

Bases: object

A simple key value store implemented in Steamship.

Instances of the KeyValueStore are identified by its namespace. This store_identifier corresponds to a File that will be created with a special tag identifying it.

Entries of the KeyValueStore are saved as Tag objects with:

• Kind = “KeyValueStore”

• Name = the key of the (kv) pair

• Value = a dict set to the value

Note that the value is always saved as a dict object. To save a string or int, wrap it in a dict.

WARNING:

This is essentially a clever hack atop Steamship’s tag system to provide mutable key-value storage. It is in the steamship.utils package because it’s proven useful once or twice. But in general, if you find yourself heavily relying upon it, consider reaching out to us at hello@steamship.com to let us know, and we’ll up-prioritize adding a proper key-value API.

client: Steamship

delete(key: str) → bool

Delete the entry represented by key

get(key: str) → Dict | None

Get the value represented by key.

items(filter_keys: List[str] | None = None) → List[Tuple[str, Dict[str, Any]]]

Return all key-value entries as a list of (key, value) tuples.

If filter_keys is provided, only returns keys within that list.

reset()

Delete all key-values.

set(key: str, value: Dict[str, Any])

Set the entry (key, value).

store_identifier: str

steamship.utils.metadata module

steamship.utils.metadata.hash_dict(d: Dict) → str

Returns the MD5 hash of a dictionary.

steamship.utils.metadata.metadata_to_str(m: int | float | bool | str | List | Dict) → str | None

steamship.utils.metadata.str_to_metadata(s: str) → int | float | bool | str | List | Dict | None

steamship.utils.repl module

class steamship.utils.repl.AgentREPL(agent_class: Type[AgentService], method: str | None = None, agent_package_config: Dict[str, Any] | None = None, client: Steamship | None = None, context_id: str | None = None, **kwargs)

Bases: SteamshipREPL

agent_class: Type[AgentService]

agent_instance: AgentService | None

client

alias of Steamship

config = None

context_id: str | None = None

print_history(client: Steamship, *args, **kwargs)

run(dump_history_on_exit: bool | None = False, **kwargs)

run_with_client(client: Steamship, **kwargs)

class steamship.utils.repl.HttpREPL(prompt_url: str, context_id: str | None = None, client: Steamship | None = None, **kwargs)

Bases: SteamshipREPL

REPL that uses an HTTP endpoint. Best for the ship serve command.

client

alias of Steamship

config = None

context_id: str | None = None

prompt_url: str | None

run(**kwargs)

run_with_client(client: Steamship, **kwargs)

class steamship.utils.repl.SteamshipREPL(log_level=None, dev_logging_handler=None)

Bases: ABC

Base class for building REPLs that facilitate running Steamship code in the IDE.

client: Steamship

dev_logging_handler: DevelopmentLoggingHandler

print_object(obj: Task | Block | str | dict, metadata: Dict[str, Any] | None = None)

Print an object, returned by the agent or tool, to the console.

Various epochs of the Agent SDK development have included Agents returning, to the repl: Blocks, strings, and Tasks. Since this is something that users can write (e.g. not controlled by the SDK) the REPL needs to handle all three cases in displaying output.

print_object_or_objects(output: List | Any, metadata: Dict[str, Any] | None = None)

Print Agent or Tool output, whether a list or a single object.

print_string(output: str, metadata: Dict[str, Any] | None = None)

Print a string to console. All REPL output should ideally route through this method.

abstract run()

temporary_workspace() → Steamship

class steamship.utils.repl.ToolREPL(tool: Tool, client: Steamship | None = None, **kwargs)

Bases: SteamshipREPL

client

alias of Steamship

run()

run_with_client(client: Workspace, context: AgentContext | None = None)

tool: Tool

steamship.utils.repl.colored(text: str, color: str, **kwargs)

steamship.utils.signed_urls module

steamship.utils.signed_urls.download_from_signed_url(url: str, to_file: Path = None) → Path

Downloads the Signed URL to the filename desired_filename in a temporary directory on disk.

steamship.utils.signed_urls.upload_to_signed_url(url: str, _bytes: bytes | None = None, filepath: Path | None = None)

Uploads either the bytes or filepath contents to the provided Signed URL.

steamship.utils.signed_urls.url_to_bytes(url: str) → bytes

Downloads the Signed URL and returns the contents as bytes.

This is a helper function to consolidate Steamship Client URL fetching to ensure a single point of handling for:

• Error messages

• Any required manipulations for URL signed URLs

• Any required manipulations for localstack-based environments

Note that the base API Client does not use this method on purpose: in the event of error code, it inspects the contents of the response for a SteamshipError.

steamship.utils.signed_urls.url_to_json(url: str) → any

Downloads the Signed URL and returns the contents as JSON.

steamship.utils.text_chunker module

steamship.utils.text_chunker.chunk_text(text: str, chunk_size: int = 200, chunk_overlap: int = 50)

Chunk text for embedding and insertion into an embedding index.

steamship.utils.url module

class steamship.utils.url.Verb(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

GET = 'GET'

POST = 'POST'

steamship.utils.url.apply_localstack_url_fix(url: str | None) → str | None

steamship.utils.url.is_local(base: str) → bool

Check if we are running the client locally.

steamship.utils.utils module

steamship.utils.utils.create_instance_handle(invocable_handle: str, version_handle: str, invocable_config: Dict[str, Any] | None) → str

Create an instance handle deterministically from package/plugin metadata and configuration.

steamship.utils.utils.format_uri(uri: str | None) → str | None

steamship.utils.utils.is_valid_uuid4(candidate: str) → bool

steamship.utils.utils.safe_get(d: Dict, key: str, default: Any = None) → Any | None

Safely a value from dictionairy using a specific key

steamship.utils.zip_archives module

steamship.utils.zip_archives.unzip_folder(zip_file: Path, into_folder: Path | None) → Path

Unzips a folder on disk, returning the path to the new folder resulting.

steamship.utils.zip_archives.zip_folder(folder: Path, into_file: Path | None) → Path

Zips a folder on disk to a co-located zip-file of the same name.

The resulting zip file does not contain the enclosing folder name provided. It contains only the children of that folder as its root elements.

Module contents

Collection of utility functions.