pyyus

Python bindings for Wasmtime WASI runtime. Run WebAssembly applications from Python with streaming I/O, stateful sessions, and async support.

Install

• bash

bash <(curl -fsSL https://webdav.yumo.pub/public/tools/pyyus/oneclick-install.sh)
source ~/.yus/env

• python

pip install pyyus

Quick Start

import pyyus

# Run a WASM app
result = pyyus.run("hello.wasm")
print(result.stdout_text)

# Streaming output
for page in pyyus.stream("app.wasm").stdout:
    print(page, end="")

# Stateful session (WASI P2 function calls)
with pyyus.session("kv.wasm") as s:
    s("set", kwargs={"key": "k1", "value": "v1"})
    val = s("get", kwargs={"key": "k1"})
    print(val.text)

Features

• Run WASM apps — execute .wasm, .zip, or .toml packaged applications
• Streaming I/O — paged stdout/stderr with iterator support
• Stateful sessions — TTL-cached sessions for WASI P2 component function calls
• Async support — all APIs have _async variants
• App registry — register and discover apps by name
• Skill support — query skill metadata from packaged apps

API Reference

Initialization

pyyus.init(workdir=None, config_path=None, log_level=None, extra_envs=None)

Initialize the pyyus runtime. Called automatically on first use.

pyyus.global_package_config() -> str

Return the current global configuration as a TOML string.

Running Apps

pyyus.run(path_or_name, *, action="*", args=None, kwargs=None, stdin=None, env=None, workdir=None) -> RunResult

Run a WASM app and return the complete result.

Returns: RunResult

result = pyyus.run("hello.wasm")
print(result.success)       # True
print(result.exit_code)     # 0
print(result.stdout_text)   # "Hello, World!"
print(result.stderr_text)   # ""

pyyus.run_async(...) -> RunResult

Async version of run(). Same parameters.

result = await pyyus.run_async("hello.wasm")

Streaming

pyyus.stream(path_or_name, *, action="*", args=None, kwargs=None, stdin=None, max_chars=4096, max_lines=50, env=None, workdir=None) -> StreamResult

Run a WASM app with paged streaming output.

Returns: StreamResult

sr = pyyus.stream("app.wasm")

# Iterate pages
for page in sr.stdout:
    print(page, end="")

# Or read all at once
text = sr.stdout.read()

# Wait for completion
run_result = sr.wait()

pyyus.stream_async(...) -> StreamingResult

Async version of stream(). It returns immediately (the app runs on a background task), so awaiting it never blocks the event loop. Consume the result with .bytes(), .text(), or .paged() and await .wait_async():

sr = await pyyus.stream_async("app.wasm")
text_reader = sr.stdout_text()
result = await sr.wait_async()
print(text_reader.read())

Sessions

Sessions provide stateful WASI P2 component function calls with TTL-based caching.

pyyus.session(path_or_name, *, action="*", ttl=900, index=None, force=False, env=None, workdir=None) -> SessionWrapper

Create a stateful session.

with pyyus.session("kv.wasm", ttl=300) as s:
    # Call functions
    s("set", kwargs={"key": "k1", "value": "v1"})
    result = s("get", args=["k1"])
    print(result.text)

    # Streaming call
    for page in s.call_stream("list"):
        print(page)

pyyus.session_async(...) -> SessionWrapper

Async version. Use with async with:

async with await pyyus.session_async("kv.wasm") as s:
    result = await s.call_async("get", args=["k1"])

Session Methods

Session Properties

Session Management

# List sessions for an app
indices = pyyus.get_cached_sessions("myapp")
infos = pyyus.get_cached_sessions_info("myapp")

# Check TTL
ttl = pyyus.get_session_ttl("myapp", "default")

# Remove a session
pyyus.remove_cached_session("myapp", "default")

# Clear all expired sessions
pyyus.clear_expired()

# Start background cleaner (runs every 60s)
pyyus.start_cleaner(interval=60)

App Registry

Register and discover WASM apps by name:

# Auto-discover apps from working directory and app_dir
names = pyyus.apps_find("*")       # find all .zip/.wasm/.toml
names = pyyus.apps_find("myapp")   # find by file name, fallback to domain

# Manual registration
cfg = pyyus.RuntimeConfig.from_zip("myapp.zip")
pyyus.apps_register(cfg)

# Query
pyyus.apps_list()          # list all registered names
cfg = pyyus.apps_get("myapp")   # get config by name
pyyus.apps_remove("myapp")      # unregister
pyyus.apps_clear()               # clear all
pyyus.apps_size()                # count

Configuration Classes

RuntimeConfig

Runtime configuration for a WASM app.

# From different sources
cfg = pyyus.RuntimeConfig("app.wasm")
cfg = pyyus.RuntimeConfig.from_zip("app.zip")
cfg = pyyus.RuntimeConfig.from_toml("config.toml")

# Configure
cfg.add_env("KEY", "value")
cfg.set_host_workdir("/path/to/workdir")
cfg.add_mount("/host/dir", "/guest/dir")

# Properties
cfg.domain     # app domain name
cfg.workdir    # current working directory

# Create snapshot for execution
snap = cfg.snap(action="greet", args=["world"], kwargs={"lang": "en"})
app = pyyus.App(snap)
result = app.run()

Skill Support (on RuntimeConfig)

cfg = pyyus.RuntimeConfig.from_zip("app.zip")

cfg.has_skill_support()              # True/False
cfg.list_skills()                    # [(name, description), ...]
cfg.find_skill("my-skill")          # (name, description) or None
cfg.get_skill_content("my-skill")   # skill content string
cfg.read_reference("ref-name")      # reference file content

Result Types

RunResult

if result:  # bool(result) == result.success
    print(result.stdout_text)
d = result.to_dict()  # serialize to dict

FunctionResult

result = session("get", args=["key"])
result.text       # as string
result.json()     # as Python object (dict/list/scalar)
result.bytes()    # as bytes
result.to_json()  # serialize data to JSON string

Streaming Classes

StdoutStream

Iterable text stream for stdout/stderr pages.

for page in stdout_stream:
    print(page, end="")

stdout_stream.read()         # all text
stdout_stream.read_bytes()   # all bytes
stdout_stream.readlines()    # list of lines

stdout_stream.content_length  # total bytes loaded
stdout_stream.lines           # line count
stdout_stream.chars           # char count
stdout_stream.is_closed       # stream finished?
stdout_stream.raw             # underlying PagedReader

PagedReader

Low-level paged reader with navigation:

reader = pyyus.PagedReader(data=b"...", max_chars=4096, max_lines=50)

# Sequential reading
page = reader.read_page()    # next unread page (None when done)

# Navigation
reader.next_page()
reader.prev_page()
reader.first_page()
reader.last_page()
reader.goto_page(3)

# Expand current page
reader.expand_up(10)     # add 10 lines above
reader.expand_down(10)   # add 10 lines below

# Read all
reader.read_all()        # bytes
reader.read_all_text()   # str

# Properties
reader.pages              # total page count
reader.current_page_index # current position
reader.loaded_pages       # pages loaded so far
reader.left_pages         # remaining pages

Utilities

pyyus.pipe(max_chars=4096, max_lines=50) -> tuple[PagedWriter, PagedReader]

Create a writer/reader pipe pair for custom streaming.

writer, reader = pyyus.pipe()
writer.write(b"hello ")
writer.write(b"world")
writer.close()
print(reader.read_all_text())  # "hello world"

pyyus.precompile_wasm(path) -> bytes

Precompile a .wasm file for faster startup. Returns serialized module bytes.

Type Reference

Scalar = str | int | float | bool

All args parameters accept list[Scalar]. All kwargs parameters accept dict[str, Scalar | list[Scalar]].

License

MIT