steamship.utils package#


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

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[source]#
steamship.utils.binary_utils.to_b64(obj: Any) str[source]#

steamship.utils.context_length module#

steamship.utils.context_length.filter_blocks_for_prompt_length(max_tokens: int, blocks: List[Block]) List[int][source]#
steamship.utils.context_length.token_length(block: Block, tiktoken_encoder: str = 'p50k_base') int[source]#

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

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

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')[source]#

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.


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 to let us know, and we’ll up-prioritize adding a proper key-value API.

client: Steamship#
delete(key: str) bool[source]#

Delete the entry represented by key

get(key: str) Dict | None[source]#

Get the value represented by key.

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

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

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


Delete all key-values.

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

Set the entry (key, value).

store_identifier: str#

steamship.utils.metadata module#

steamship.utils.metadata.hash_dict(d: Dict) str[source]#

Returns the MD5 hash of a dictionary.

steamship.utils.metadata.metadata_to_str(m: int | float | bool | str | List | Dict) str | None[source]#
steamship.utils.metadata.str_to_metadata(s: str) int | float | bool | str | List | Dict | None[source]#

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)[source]#

Bases: SteamshipREPL

agent_class: Type[AgentService]#
agent_instance: AgentService | None#

alias of Steamship

config = None#
context_id: str | None = None#
print_history(client: Steamship, *args, **kwargs)[source]#
run(dump_history_on_exit: bool | None = False, **kwargs)[source]#
run_with_client(client: Steamship, **kwargs)[source]#
class steamship.utils.repl.HttpREPL(prompt_url: str, context_id: str | None = None, client: Steamship | None = None, **kwargs)[source]#

Bases: SteamshipREPL

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


alias of Steamship

config = None#
context_id: str | None = None#
prompt_url: str | None#
run_with_client(client: Steamship, **kwargs)[source]#
class steamship.utils.repl.SteamshipREPL(log_level=None, dev_logging_handler=None)[source]#

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)[source]#

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)[source]#

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

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

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

abstract run()[source]#
temporary_workspace() Steamship[source]#
class steamship.utils.repl.ToolREPL(tool: Tool, client: Steamship | None = None, **kwargs)[source]#

Bases: SteamshipREPL


alias of Steamship

run_with_client(client: Workspace, context: AgentContext | None = None)[source]#
tool: Tool#
steamship.utils.repl.colored(text: str, color: str, **kwargs)[source]#

steamship.utils.signed_urls module#

steamship.utils.signed_urls.download_from_signed_url(url: str, to_file: Path = None) Path[source]#

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)[source]#

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

steamship.utils.signed_urls.url_to_bytes(url: str) bytes[source]#

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

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)[source]#

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)[source]#

Bases: str, Enum

GET = 'GET'#
steamship.utils.url.apply_localstack_url_fix(url: str | None) str | None[source]#
steamship.utils.url.is_local(base: str) bool[source]#

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

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

steamship.utils.utils.format_uri(uri: str | None) str | None[source]#
steamship.utils.utils.is_valid_uuid4(candidate: str) bool[source]#
steamship.utils.utils.safe_get(d: Dict, key: str, default: Any = None) Any | None[source]#

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

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

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.