Skip to Content
docsClient-APIsPython

Last Updated: 3/9/2026

Skip to main content

Python Client

A Python client for interacting with the Pie server.

Installation​

pip install pie-clientpip install pie-client

Quick Start​

import asyncio import  asynciofrom pie import PieClient, ParsedPrivateKey from  pie import  PieClient,  ParsedPrivateKey async def main(): async  def  main():     async with PieClient("ws://127.0.0.1:8080") as client:     async  with  PieClient("ws://127.0.0.1:8080")  as  client:         # Authentication is always required         
# Authentication is always required         # If server auth is enabled, provide a valid private key:         
# If server auth is enabled, provide a valid private key:         key = ParsedPrivateKey.from_file("~/.ssh/id_ed25519")         key =  ParsedPrivateKey. from_file("~/.ssh/id_ed25519")         await client.authenticate("username", key)         await  client. authenticate("username",  key)                         # If server auth is disabled, any username works without a key:         
# If server auth is disabled, any username works without a key:         # await client.authenticate("any_username")         
# await client.authenticate("any_username")                         # Upload and launch a program         
# Upload and launch a program         with open("my_program.wasm", "rb") as f:         with  open("my_program.wasm",  "rb")  as  f:             await client.upload_program(f.read())             await  client. upload_program(f. read())                         program_hash = "..."  # blake3 hash of the wasm binary         program_hash =  "..."   
# blake3 hash of the wasm binary         instance = await client.launch_instance(program_hash)         instance =  await  client. launch_instance(program_hash)                         # Interact with the instance         
# Interact with the instance         await instance.send("hello")         await  instance. send("hello")         event, message = await instance.recv()         event,  message =  await  instance. recv()         print(f"Received: {event.name} - {message}")         print(f"Received: {event. name} - {message} ")  asyncio.run(main()) asyncio. run(main())

PieClient​

The main client class for connecting to a Pie server.

Constructor​

PieClient(server_uri: str) PieClient(server_uri:  str)
ParameterTypeDescription
server_uristrWebSocket URI (e.g., "ws://127.0.0.1:8080")

Methods​

MethodDescription
authenticate(username, private_key=None)Public key authentication (challenge-response)
internal_authenticate(token)Token-based internal authentication
upload_program(bytes)Upload a WASM program
program_exists(hash)Check if program is uploaded
launch_instance(hash, args=[], detached=False)Launch a program instance by hash
launch_instance_from_registry(name, args=[], detached=False)Launch from registry
attach_instance(instance_id)Attach to a detached instance
list_instances()List running instances
terminate_instance(instance_id)Terminate an instance
ping()Check server connectivity
query(subject, record)Generic server query

Context Manager​

Use as an async context manager for automatic cleanup:

async with PieClient("ws://127.0.0.1:8080") as client: async  with  PieClient("ws://127.0.0.1:8080")  as  client:     # client is connected     
# client is connected     await client.authenticate("username")     await  client. authenticate("username")     # ...     
# ... # client is automatically closed 
# client is automatically closed

Instance​

Represents a running program instance on the server.

Methods​

MethodDescription
send(message)Send a string message to the instance
upload_blob(bytes)Upload binary data to the instance
recv()Receive next event (returns (Event, data))
terminate()Request termination

Example​

instance = await client.launch_instance_from_registry(instance =  await  client. launch_instance_from_registry(    "text-completion",     "text-completion",     ["--prompt", "Hello"]     ["--prompt",  "Hello"] ) )  # Send additional input 
# Send additional input await instance.send("Continue the story...") await  instance. send("Continue the story...")  # Receive streaming output 
# Receive streaming output while True: while  True:     event, data = await instance.recv()     event,  data =  await  instance. recv()             if event.name == "Stdout":     if  event. name ==  "Stdout":         print(data, end="")         print(data,  end = "")     elif event.name == "Completed":     elif  event. name ==  "Completed":         break         break

Event Types​

Events returned by instance.recv():

EventDescription
MessageText message from instance
StdoutStreaming stdout output
StderrStreaming stderr output
BlobBinary data received
CompletedInstance finished successfully
AbortedInstance was aborted
ExceptionInstance raised an exception
ServerErrorServer-side error
OutOfResourcesResource limit reached

ParsedPrivateKey​

Handles SSH private keys for authentication.

Supported Key Types​

  • RSA (≥2048 bits)
  • ED25519
  • ECDSA (P-256, P-384)

Methods​

# From file
# From file key = ParsedPrivateKey.from_file("~/.ssh/id_ed25519") key =  ParsedPrivateKey. from_file("~/.ssh/id_ed25519")  # From string 
# From string key = ParsedPrivateKey.parse(key_content) key =  ParsedPrivateKey. parse(key_content)

Example​

from pie import PieClient, ParsedPrivateKey from  pie import  PieClient,  ParsedPrivateKey key = ParsedPrivateKey.from_file("~/.ssh/id_ed25519") key =  ParsedPrivateKey. from_file("~/.ssh/id_ed25519")  async with PieClient("ws://127.0.0.1:8080") as client: async  with  PieClient("ws://127.0.0.1:8080")  as  client:     await client.authenticate("username", key)     await  client. authenticate("username",  key)

Detached Instances​

Launch instances that persist after disconnection:

# Launch detached
# Launch detached instance = await client.launch_instance_from_registry(instance =  await  client. launch_instance_from_registry(    "long-running-task",     "long-running-task",     detached=True     detached = True ) ) instance_id = instance.id instance_id =  instance. id  # Later, reattach 
# Later, reattach instance = await client.attach_instance(instance_id) instance =  await  client. attach_instance(instance_id) event, data = await instance.recv() event,  data =  await  instance. recv()

Error Handling​

from pie import PieClient, PieError from  pie import  PieClient,  PieError try: try:     async with PieClient("ws://127.0.0.1:8080") as client:     async  with  PieClient("ws://127.0.0.1:8080")  as  client:         await client.authenticate("username")         await  client. authenticate("username")         instance = await client.launch_instance("invalid_hash")         instance =  await  client. launch_instance("invalid_hash") except PieError as e: except  PieError as  e:     print(f"Pie error: {e}")     print(f"Pie error: {e} ") except ConnectionError as e: except  ConnectionError as  e:     print(f"Connection failed: {e}")     print(f"Connection failed: {e} ")
JavascriptClient Basics