pyyus

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

Install

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.

Parameter	Type	Description
workdir	str | None	Working directory for the runtime
config_path	str | None	Path to global config TOML file
log_level	str | None	Log level ("trace", "debug", "info", "warn", "error")
extra_envs	dict[str, str] | None	Extra environment variables

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.

Parameter	Type	Description
path_or_name	str	Path to .wasm/.zip/.toml, or a registered app name
action	str	Entry point name (default "*")
args	list[Scalar] | None	Positional arguments
kwargs	dict[str, Scalar | list[Scalar]] | None	Keyword arguments
stdin	bytes | str | IO | None	Input data
env	dict[str, str] | None	Extra environment variables
workdir	str | None	Host working directory

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.

Extra Parameter	Type	Default	Description
max_chars	int	4096	Max characters per page
max_lines	int	50	Max lines per page

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(...) -> AsyncStreamResult

Async version. Iterate with async for:

asr = await pyyus.stream_async("app.wasm")
async for page in asr.stdout:
    print(page, end="")

---

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.

Parameter	Type	Default	Description
ttl	int	900	Time-to-live in seconds (0 = permanent)
index	str | None	None	Session index for multiple instances
force	bool	False	Force create new session even if cached

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

Method	Description
s.call(func, args=, kwargs=, data=)	Sync function call → FunctionResult
s.call_async(func, args=, kwargs=, data=)	Async function call → FunctionResult
s.call_stream(func, ..., max_chars=, max_lines=)	Streaming call → StdoutStream
s.call_stream_async(func, ...)	Async streaming call → AsyncStdoutStream
s(func, args=, kwargs=, data=)	Shorthand for call()
s.close()	Remove session from cache

Session Properties

Property	Type	Description
s.session_id	str	Unique session ID
s.app_name	str	App domain name
s.index	str	Session index
s.is_active	bool	Whether session is active
s.ttl	int	Remaining TTL in seconds

---

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 .zip apps from working directory
names = pyyus.apps_find("*")       # find all
names = pyyus.apps_find("myapp")   # find specific

# 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")
cfg = pyyus.RuntimeConfig.from_app_config(app_config)

# 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()

AppConfig

Lower-level app configuration:

ac = pyyus.AppConfig("app.wasm")
ac = pyyus.AppConfig.from_zip("app.zip")
ac = pyyus.AppConfig.from_toml("config.toml")

ac.domain           # domain name
ac.add_env("K", "V")
ac.add_mount("/host", "/guest")

cfg = ac.build(rename="custom-name")  # → RuntimeConfig

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

Attribute	Type	Description
success	bool	Whether execution succeeded
exit_code	int | None	Process exit code
error_message	str | None	Error message if failed
stdout	bytes | None	Raw stdout
stderr	bytes | None	Raw stderr
stdout_text	str	stdout as UTF-8 string
stderr_text	str	stderr as UTF-8 string

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

FunctionResult

Attribute	Type	Description
success	bool	Whether the call succeeded
data	ResultTypes | None	Result data
error	str | None	Error message
text	str | None	Data as text

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