Last Updated: 3/5/2026

Python Client

A Python client for interacting with the Pie server.

Installation


pip install pie-client

Quick Start


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

PieClient

The main client class for connecting to a Pie server.

Constructor


PieClient(server_uri: str)

Parameter	Type	Description
`server_uri`	`str`	WebSocket URI (e.g., `"ws://127.0.0.1:8080"`)

Methods

Method	Description
`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:
    # client is connected
    await client.authenticate("username")
    # ...
# client is automatically closed

Instance

Represents a running program instance on the server.

Methods

Method	Description
`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(
    "text-completion",
    ["--prompt", "Hello"]
)
 
# Send additional input
await instance.send("Continue the story...")
 
# Receive streaming output
while True:
    event, data = await instance.recv()
    
    if event.name == "Stdout":
        print(data, end="")
    elif event.name == "Completed":
        break

Event Types

Events returned by instance.recv():

Event	Description
`Message`	Text message from instance
`Stdout`	Streaming stdout output
`Stderr`	Streaming stderr output
`Blob`	Binary data received
`Completed`	Instance finished successfully
`Aborted`	Instance was aborted
`Exception`	Instance raised an exception
`ServerError`	Server-side error
`OutOfResources`	Resource limit reached

ParsedPrivateKey

Handles SSH private keys for authentication.

Supported Key Types

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

Methods


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

Example


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

Detached Instances

Launch instances that persist after disconnection:


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

Error Handling


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