Typed calls and subscriptions

The runtime wraps the official Convex Rust client with generated function references. These references constrain queries, mutations, and actions to the matching argument and return models, expose query subscriptions as streams, and can declare a Rust type for application error payloads.

Use TypedClient when the generated Rust types should validate arguments and decode results for you. Create the official SDK client first and give the adapter a mutable borrow. Use the first project for a runnable consumer with the functions below.

Calls

Pass the generated function reference as the first argument and its matching args as the second. The Rust compiler rejects a query reference passed to mutation, or args from a different endpoint. A declared empty args object still has its own type: pass tasks::ListArgs {} to tasks::list. A declared return validator determines the result type. See generated models for undeclared validators.

Like the SDK, query opens a subscription, returns its first result, and then drops the subscription.

Subscriptions

Use subscribe when a view should stay current as data changes. Only queries can be subscribed to. The tutorial opens the stream, writes a new row, and waits for the write to appear in a result. futures::StreamExt provides next().

    let mut updates = client.subscribe(tasks::list, tasks::ListArgs {}).await?;
    let text = format!("Write Rust ({})", SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos());
    let id: Id<tables::Tasks> = client
        .mutation(
            add,
            tasks::AddArgs {
                text: text.clone(),
                priority: Some(Some(tasks::AddArgsPriority::High)),
            },
        )
        .await?;
    println!("Created: {}", id.as_str());

    // The stream can first yield the initial result. Wait for this write to appear.
    timeout(Duration::from_secs(10), async {
        while let Some(update) = updates.next().await {
            if update?.contains(&text) {
                println!("Observed: {text}");
                return Ok::<(), Box<dyn std::error::Error>>(());
            }
        }
        Err("subscription ended before observing the write".into())
    })
    .await??;

The stream delivers an initial result and then reflects changes to the data read by the query. Each item decodes independently. Keep the SDK client alive while polling the stream; the stream owns the subscription and releases it on drop. A closed subscription ends the stream. In an integration check, bound the wait and look for the expected change rather than assuming it is the next item; the tutorial does both.

See application errors to distinguish application failures from encoding, decoding, and transport failures.

Using the rest of the Convex SDK

TypedClient borrows a &mut convex::ConvexClient. Use its typed methods for generated calls and client_mut() whenever you need the SDK directly. Set up the connection and make any SDK clones before borrowing it.

Import the SDK with use convex_typegen_runtime::convex; to use the exact version and types the runtime uses. See the semver policy and the Convex Rust SDK docs.

The SDK calls an auth callback with force_refresh = true on reconnect. Pass that flag to your token provider so it can fetch a fresh token. For example, if auth_provider is an Arc to your provider with an async token(force_refresh) method returning a JWT:

let fetch_token: convex::AuthTokenFetcher = Box::new(move |force_refresh| {
    let provider = auth_provider.clone();
    Box::pin(async move {
        let jwt = provider.token(force_refresh).await?;
        Ok(convex::AuthenticationToken::User(jwt))
    })
});
client.client_mut().set_auth_callback(Some(fetch_token)).await;

Use the generated function’s path() and encode its args as a Convex object for an untyped SDK call. The SDK returns a FunctionResult, so you can decode a successful value separately or inspect the SDK’s error variants.

use convex::{FunctionResult, Value};
use convex_typegen_runtime::value_serde::{from_value, to_value};

let Value::Object(raw_args) = to_value(&tasks::ListArgs {})? else {
    return Err("expected a Convex argument object".into());
};
let result = client.client_mut().query(tasks::list.path(), raw_args).await?;
if let FunctionResult::Value(value) = result {
    let tasks: Vec<String> = from_value(value)?;
    println!("{tasks:?}");
}

A builder accepts a Tokio channel sender for connection-state updates. Keep the receiver wherever your app handles connection status.

use convex::{ConvexClient, ConvexClientBuilder};
use convex_typegen_runtime::TypedClient;

let (state_tx, mut state_rx) = tokio::sync::mpsc::channel(8);
let builder = ConvexClientBuilder::new(url)
    .with_client_id("tasks")
    .with_on_state_change(state_tx);
let mut sdk = ConvexClient::new_from_builder(builder).await?;
let mut other_sdk = sdk.clone();
let mut client = TypedClient::new(&mut sdk);
let mut other_client = TypedClient::new(&mut other_sdk);