Skip to main content

Connecting clients

A client is the application that calls your inferlet. Three official client libraries cover the common cases: Python (pie-client), JavaScript (pie-client), and Rust (pie-client crate). This page is task-shaped: how to connect, launch, stream, and close. The full API surface is in the per-language reference. Read this after Run a server.

Install​

[dependencies]
pie-client = "*"
tokio = { version = "1", features = ["full"] }
serde_json = "1"
anyhow = "1"

Connect and authenticate​

use pie_client::{Client, ParsedPrivateKey};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let client = Client::connect("ws://127.0.0.1:8080").await?;

    // Public-key auth (Ed25519 / RSA / ECDSA).
    let key = ParsedPrivateKey::from_file("~/.ssh/id_ed25519")?;
    client.authenticate("alice", &Some(key)).await?;

    // ... use the client ...

    client.close().await?;
    Ok(())
}

If the engine is running with --no-auth, the authenticate call is a no-op (Python and Rust) and you skip it entirely (JavaScript).

Launch a process​

launch_process(name@version, input) starts an inferlet. The returned handle streams events.

let input = serde_json::json!({"question": "What is 2+2?"}).to_string();
let mut process = client
    .launch_process(
        "research-agent@0.1.0".into(),
        input,
        true,           // capture stdout/stderr
        None,           // token budget (None = unlimited)
    )
    .await?;

input becomes the inferlet's typed Input struct. The launch resolves once the engine has confirmed the inferlet started; events stream after.

Stream events​

The handle yields events one by one. Loop until Return (success) or Error (failure).

use pie_client::ProcessEvent;

loop {
    match process.recv().await? {
        ProcessEvent::Stdout(s) => print!("{s}"),
        ProcessEvent::Stderr(s) => eprint!("{s}"),
        ProcessEvent::Message(m) => println!("[message] {m}"),
        ProcessEvent::File(b) => save_blob(&b),
        ProcessEvent::Return(v) => {
            let answer: String = serde_json::from_str(&v)?;
            println!("{answer}");
            break;
        }
        ProcessEvent::Error(e) => anyhow::bail!("{e}"),
    }
}

Stdout events arrive in the order the inferlet emitted them. Per-event ordering is preserved across stdout / stderr / messages.

Send signals and files​

The client side mirrors the inferlet's session surface.

process.signal("stop").await?;
process.transfer_file(&pdf_bytes).await?;

The inferlet receives signals via session.receive_signal() and files via session.receive_file().

Concurrent calls on one connection​

A single connection multiplexes many processes. Launch several without awaiting, drain them in parallel.

let mut handles = Vec::new();
for q in questions {
    let input = serde_json::json!({"question": q}).to_string();
    let p = client.launch_process(
        "research-agent@0.1.0".into(), input, true, None,
    ).await?;
    handles.push(p);
}

// Drain in parallel via tokio tasks.
let answers = futures::future::join_all(
    handles.into_iter().map(|mut p| async move {
        loop {
            match p.recv().await? {
                ProcessEvent::Return(v) => return Ok::<_, anyhow::Error>(v),
                ProcessEvent::Error(e) => anyhow::bail!("{e}"),
                _ => {}
            }
        }
    })
).await;

The engine batches forward passes from concurrent processes. Three runs in parallel against the same engine take roughly the same wall-clock time as one run, plus per-step overhead.

Upload a local build​

For client-uploaded builds (no registry):

use std::path::Path;

if !client.check_program("research-agent@0.1.0", None, None).await? {
    client.add_program(
        Path::new("./research-agent.wasm"),
        Path::new("./Pie.toml"),
        false,
    ).await?;
}

The upload chunks at 256 KiB. The build stays loaded as long as the engine runs.

Detach and reattach​

Long-running processes can be detached and reattached later by ID.

let id = process.id().to_string();
drop(process);  // detach

// later, on a new client connection:
let mut process = client.attach_process(&id).await?;

The process keeps running on the engine while detached. Reattachment resumes the event stream from where it left off (with the buffered events the engine has held).

Next​