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
with pyyus.stream("app.wasm") as sr:
    with sr.text() as out:        # stdout + stderr combined
        for line in out:
            print(line, end="")

# Paged output (page-by-page reading)
with pyyus.paged("app.wasm", max_lines=50) as pr:
    for page in pr.paged():
        print(page, end="")

# Stateful session (WASI P2 function calls)
with pyyus.session("kv.wasm") as s:
    s("set", key="k1", value="v1")
    val = s("get", 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 — run_async / stream_async / paged_async, plus async session calls (call_async / stream_async)
• App registry — register and discover apps by name
• Skill support — query skill metadata from packaged apps

API Reference

Full, up-to-date reference: docs/pypackage/api-reference.md (中文). This section is a summary; the docs directory is authoritative.

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, stdin=None, env=None, workdir=None, timeout=-1, kwargs=None, **kv) -> RunResult

Run a WASM app and return the complete result.

Returns: RunResult

result = pyyus.run("hello.wasm")
result = pyyus.run("hello.zip", "greet", "world")     # positional arg
result = pyyus.run("myapp", "set", key="k1", value="v1")  # keyword args
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, stdin=None, env=None, workdir=None, timeout=-1, kwargs=None, **kv) -> StreamingResult

Run a WASM app with streaming output. stream() returns immediately — the app runs on a background thread — so the result is consumed through reader factories instead of an aggregated stdout field.

with pyyus.stream("app.wasm") as sr:
    with sr.text() as ss:               # CombinedStdStream (stdout + stderr)
        for line in ss:                 # line-by-line
            print(line, end="")
        err = ss.stderr.text().read(-1)

    result = sr.wait()                  # RunResult, blocks until done

Readers: sr.stdout_text() / sr.stdout_bytes() / sr.stdout_paged(), the same three for stderr_*, plus the combined text() / bytes() / paged(). Lifecycle: wait() / wait_async() / is_done() / kill() / elapsed.

pyyus.paged(path_or_name, action="*", *args, ..., timeout=-1, max_chars=4096, max_lines=50) -> PagedRunResult

Same parameters as stream(), plus page size:

The whole event stream (stdout raw, stderr tagged [stderr], other events tagged [event]) is piped into a single PagedReader:

with pyyus.paged("app.wasm", max_lines=50) as pr:
    reader = pr.paged()
    for page in reader:            # page by page
        print(page, end="")
    result = pr.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, token=None, env=None, workdir=None) -> SessionWrapper

Create a stateful session.

with pyyus.session("kv.wasm", ttl=300) as s:
    # lib entries take named arguments (kwargs= or **kv)
    s("set", key="k1", value="v1")
    result = s("get", key="k1")
    print(result.text)      # "v1"

    # Streaming call
    with s.stream("list-keys") as ss:
        print(ss.text().read())

Async sessions

There is no separate session_async() factory — session() returns a SessionWrapper that supports both context-manager protocols, and the async methods are on the wrapper:

async with pyyus.session("kv.wasm") as s:
    await s.call_async("set", kwargs={"key": "k1", "value": "v1"})
    result = await s.call_async("get", kwargs={"key": "k1"})
    print(result.text)

Session Methods

Session Properties

Session Management

Cached sessions live in the pyyus.sessions submodule:

# List sessions for an app
indices = pyyus.sessions.list("myapp")
infos = pyyus.sessions.list_info("myapp")

# Check TTL / token
ttl = pyyus.sessions.ttl("myapp", "default")
token = pyyus.sessions.get_token("myapp", "default")
ok = pyyus.sessions.check_token("myapp", "default", token)

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

# Clear all expired sessions
pyyus.sessions.clear_expired()

# Start background cleaner (process-wide, only the first call takes effect)
pyyus.sessions.start_cleaner(interval=60)

App Registry

Register and discover WASM apps by name (pyyus.apps submodule):

# 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 (cached)
cfg = pyyus.apps.load("myapp") # load from disk
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

Streaming results hand out reader objects rather than a raw stdout attribute. Three reader types exist — BytesReader, TextReader and PagedReader — obtained from StreamingResult / PagedRunResult / SessionStreamResult (.bytes() / .text() / .paged()).

BytesReader / TextReader

File-like readers, both iterable and usable as context managers.

data = reader.read()              # read everything (-1 = all), blocking
data = reader.read(4096)          # up to 4096 bytes / chars
print(reader.readall())           # BytesReader only
for chunk in reader:              # iterate until EOF
    ...

reader.closed      # stream closed?
reader.is_eof      # BytesReader: no more data
reader.available   # BytesReader: buffered bytes ready without blocking

TextReader additionally has readline() / readlines() / encoding.

PagedReader

Low-level paged reader with navigation:

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

# Sequential reading
page = next(reader)               # iterator yields pages (StopIteration at end)

# Navigation
reader.next_page()
reader.prev_page()
reader.first_page()
reader.last_page()
reader.goto_page(3)          # 1-based
reader.cache_all_pages()     # drain the stream, return page count

# 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[ChannelWriter, PagedReader]

Create a writer/reader pipe pair for custom streaming. pyyus.bytes_pipe() and pyyus.text_pipe() return BytesReader / TextReader instead.

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

App/function arguments are passed positionally (*args: Scalar) or as keyword arguments (kwargs={"k": ...} / **kv); kwarg values may also be list[Scalar] or nested dict.

License

MIT