> ## Documentation Index
> Fetch the complete documentation index at: https://docs.metastreams.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Rust SDK

> A typed Rust client for every REST endpoint and for the WebSocket stream.

The Rust SDK is a typed client for every `/v1` REST endpoint and for the `/v1/stream` WebSocket. It consists of two crates, published from the [`metastreams-sdk`](https://github.com/metastreams/metastreams-sdk) repository:

| Crate | Contents |
| - | - |
| `metastreams-types` | Every request and response of the API as serde types, without an HTTP or WebSocket dependency |
| `metastreams-sdk` | `MetastreamsProvider`, with one method per REST endpoint, and `StreamConnection`, with typed stream subscriptions. It re-exports the types crate as `metastreams_sdk::types`. |

The API uses the same types, so they match its responses field for field.

## Installation

The crates are not on crates.io yet. Add them as a git dependency pinned to a tag:

```toml Cargo.toml theme={null}
[dependencies]
metastreams-sdk = { git = "https://github.com/metastreams/metastreams-sdk", tag = "v0.1.0" }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
futures = "0.3" # StreamExt, for subscriptions
```

To track a commit, pin a `rev` instead of a `tag`. The minimum supported Rust version is 1.93. TLS uses rustls with the webpki root certificates, so you don't install a crypto provider yourself.

## Provider

`MetastreamsProvider::new(key)` creates a client for the production API, and `MetastreamsProvider::new_with_base_url(key, url)` creates one for another origin. The provider implements `Clone`, and its `Debug` output hides the key.

Each method sends one request, with a 30-second request timeout and a 10-second connect timeout. A method never retries.

```rust theme={null}
use metastreams_sdk::{MetastreamsProvider, MultiChainAddress, SpotChain};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;
    let bonk: MultiChainAddress = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263".parse()?;

    let token = provider.token_details(SpotChain::Solana, &bonk).await?;
    println!("{} {:?}", token.symbol, token.price.usd);
    Ok(())
}
```

The examples in the SDK repository read the key from `METASTREAMS_API_KEY`. The samples on this page read it from `API_KEY`, as the rest of these docs do.

### Methods

| Method | Endpoint | Returns |
| - | - | - |
| `token_details(chain, &address)` | [`GET /v1/tokens/{chain}/{address}`](https://docs.metastreams.io/reference/rest/token-details) | `UiSpotToken` |
| `token_ohlcv(chain, &address, &SpotCandlesQuery)` | [`GET /v1/tokens/{chain}/{address}/ohlcv`](https://docs.metastreams.io/reference/rest/token-ohlcv) | `UiCandlesResponse` |
| `token_trades(chain, &address, &TokenTradesQuery)` | [`GET /v1/tokens/{chain}/{address}/trades`](https://docs.metastreams.io/reference/rest/token-trades) | `UiSpotTrades` |
| `token_holders(chain, &address, &TokenHoldersQuery)` | [`GET /v1/tokens/{chain}/{address}/holders`](https://docs.metastreams.io/reference/rest/token-holders) | `UiTokenHolderList` |
| `identity_trades(&IdentityTradesQuery)` | [`GET /v1/trades`](https://docs.metastreams.io/reference/rest/identity-trades) | `UiSpotTrades` |
| `wallet_balances(&WalletBalancesQuery)` | [`GET /v1/wallets/balances`](https://docs.metastreams.io/reference/rest/wallet-balances) | `UiWalletBalances` |
| `positions(&PositionsQuery)` | [`GET /v1/positions`](https://docs.metastreams.io/reference/rest/positions) | `UiSpotPositions` |
| `native_prices(&NativePricesQuery)` | [`GET /v1/native-prices`](https://docs.metastreams.io/reference/rest/native-prices) | `UiNativePrices` |
| `discovery_board(&SpotPulseSubscriptionConfig)` | [`POST /v1/discovery`](https://docs.metastreams.io/reference/rest/discovery) | `UiSpotPulseResponse` |
| `trending_board(&SpotTrendingFilter)` | [`POST /v1/trending`](https://docs.metastreams.io/reference/rest/trending) | `UiSpotTrendingUpdate` |
| `session_token_create(&CreateSessionTokenRequest)` | [`POST /v1/session-tokens`](https://docs.metastreams.io/reference/rest/session-tokens) | `UiSessionToken` |
| `connect()` | [`GET /v1/stream`](https://docs.metastreams.io/streams/overview) | `StreamConnection` |

When `identity_trades`, `wallet_balances` or `positions` leaves `chains` at `None`, the SDK sends every chain it knows.

### Pagination

A paginated response carries `has_more` and `next_cursor`. Copy `next_cursor` into the query to read the next page:

```rust theme={null}
use metastreams_sdk::types::trade::TokenTradesQuery;
use metastreams_sdk::{MetastreamsProvider, MultiChainAddress, SpotChain};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;
    let bonk: MultiChainAddress = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263".parse()?;

    let mut query = TokenTradesQuery { limit: Some(100), ..Default::default() };
    loop {
        let page = provider.token_trades(SpotChain::Solana, &bonk, &query).await?;
        for trade in &page.trades {
            println!("{} {:?} {:?}", trade.id, trade.side, trade.volume.usd);
        }
        if !page.has_more {
            break;
        }
        query.cursor = page.next_cursor;
    }
    Ok(())
}
```

### Session tokens

A backend creates a [session token](https://docs.metastreams.io/get-started/authentication.md#session-tokens) for a browser user with `session_token_create`:

```rust theme={null}
use metastreams_sdk::MetastreamsProvider;
use metastreams_sdk::types::session::CreateSessionTokenRequest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;

    let request = CreateSessionTokenRequest {
        origins: vec!["https://app.example.com".into()],
        subject: Some("user-42".into()),
        ttl_seconds: Some(900),
    };
    let session = provider.session_token_create(&request).await?;
    println!("expires at {}", session.expires_at.0);
    Ok(())
}
```

## Streams

`provider.connect()` opens one WebSocket, which carries many subscriptions. The filter type selects the channel and the payload types:

| Filter | Channel | `Exit` payload |
| - | - | - |
| `TradeFilter` | [`spot.trades`](https://docs.metastreams.io/reference/streams/spot-trades) | None |
| `TransferFilter` | [`spot.transfers`](https://docs.metastreams.io/reference/streams/spot-transfers) | None |
| `CandleFilter` | [`spot.candles`](https://docs.metastreams.io/reference/streams/spot-candles) | None |
| `TokenFilter` | [`spot.tokens`](https://docs.metastreams.io/reference/streams/spot-tokens) | `UiTokenExit` |
| `WalletBalancesFilter` | [`wallet.balances`](https://docs.metastreams.io/reference/streams/wallet-balances) | None |
| `SpotPositionsFilter` | [`spot.positions`](https://docs.metastreams.io/reference/streams/spot-positions) | None |
| `SpotHoldersFilter` | [`spot.holders`](https://docs.metastreams.io/reference/streams/spot-holders) | None |
| `NativePricesFilter` | [`chain.nativePrices`](https://docs.metastreams.io/reference/streams/native-prices) | None |
| `SpotTrendingFilter` | [`spot.trending`](https://docs.metastreams.io/reference/streams/spot-trending) | None |

`TradeFilter`, `TransferFilter`, `CandleFilter` and `TokenFilter` are in `metastreams_sdk::types::filter`. `SpotTrendingFilter` is in `metastreams_sdk::types::trending`, and the other filter types are in `metastreams_sdk::types::stream`. Each subscription is a `futures::Stream` of `Event`s. An `Event::Update` carries one payload, and an `Event::Exit` means a token the subscription matched no longer matches. A channel without exits uses an empty exit type, so its `Exit` arm is `match exit {}`.

```rust theme={null}
use futures::StreamExt;
use metastreams_sdk::types::filter::TradeFilter;
use metastreams_sdk::{Event, MetastreamsProvider, MultiChainAddress, SpotChain};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let provider = MetastreamsProvider::new(std::env::var("API_KEY")?)?;
    let bonk: MultiChainAddress = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263".parse()?;

    let connection = provider.connect().await?;
    let filter = TradeFilter {
        chains: vec![SpotChain::Solana],
        tokens: Some(vec![bonk]),
        ..Default::default()
    };
    let mut trades = connection.subscribe(&filter).await?;
    while let Some(event) = trades.next().await {
        match event? {
            Event::Update(trade) => println!("{:?} {} {}", trade.side, trade.amount, trade.symbol),
            Event::Exit(exit) => match exit {},
        }
    }
    Ok(())
}
```

### Subscription options

`subscribe_throttled(&filter, Duration)` sets [`throttleMs`](https://docs.metastreams.io/streams/overview#client-messages) on a `TokenFilter` or `CandleFilter`, rounded up to whole milliseconds. Exits are never throttled. `subscribe_candles(&CandleFilter, CandleOptions)` also sets the candle `metric`, `denomination` and `closed_only`.

### Subscription lifecycle

The stream sends no snapshot, except on `spot.trending`, whose first update is the full board. On other channels, read the REST endpoint after `subscribe` returns, as [Initial snapshot](https://docs.metastreams.io/streams/overview.md#initial-snapshot) describes. `id()` returns the subscription ID the server assigned, and dropping a subscription unsubscribes it.

A subscription ends with one `Err(MetastreamsError::Closed(reason))`, followed by `None`. A payload that fails to decode yields `Err(MetastreamsError::FrameDecode { .. })`, and the subscription continues.

`connection.closed()` resolves to the `CloseReason` when the socket closes. `connection.close()` closes the socket normally, and each subscription then ends with `CloseReason::Client`.

The SDK sends a ping every 15 seconds and treats the socket as lost after 45 seconds without a frame. Each subscription buffers up to 1,024 unread events. When the buffer is full, the SDK closes that subscription with `CloseReason::Lagged` and leaves the others open.

### Close reasons

| `CloseReason` | Cause | Transient |
| - | - | - |
| `ServerRestarting` | Close code `1001` | Yes |
| `SlowConsumer` | Close code `4008` | Yes |
| `KeyRevoked` | Close code `1008` | No |
| `CreditsExhausted` | Close code `4402` | No |
| `Lagged` | The subscription's buffer filled | Yes |
| `Lost(LostCause)` | The socket ended without a close code, or went silent | Yes |
| `Other { code, reason }` | A close code this SDK version does not name | Yes |
| `Client` | You closed the connection | Yes |

### Reconnection

The SDK does not reconnect. Subscriptions end with their connection, and missed updates are not replayed. After a transient close, connect, subscribe again and re-read the snapshot from REST. Reconnect at once after `ServerRestarting`, and back off after other transient reasons. After `KeyRevoked` or `CreditsExhausted`, stop. See [Reconnection](https://docs.metastreams.io/streams/overview.md#reconnection).

## Errors

Every method returns `metastreams_sdk::Result<T>`, whose error is `MetastreamsError`:

| Variant | Cause |
| - | - |
| `Api { status, error, request_id }` | The API refused the request with an [error envelope](https://docs.metastreams.io/concepts/errors.md). Match on `error.code`, never on `error.message`. |
| `RateLimited { retry_after, request_id }` | A `429`, including a stream handshake past the connection limit, with the `Retry-After` wait |
| `Rejected { channel, error }` | The server rejected a subscribe request with an `error` frame |
| `Closed(CloseReason)` | A subscription ended |
| `FrameDecode { channel, source }` | A stream payload did not decode |
| `Http`, `WebSocket` | A transport failure |
| `Decode`, `Unexpected` | A response body the SDK could not read |
| `Url`, `UnsupportedScheme`, `InvalidKey` | The base URL or the key cannot be used |

`error.code` is an `ErrorCode` from `metastreams_sdk::types::error`. A code that this SDK version does not know becomes `ErrorCode::Unknown`.

`is_transient()` reports whether the same call can succeed later: a transport failure, a `429`, a `5xx`, or a transient close. The example below retries transient errors and waits for `Retry-After` on a `429`:

```rust theme={null}
use std::time::Duration;

use metastreams_sdk::types::token::UiSpotToken;
use metastreams_sdk::{MetastreamsError, MetastreamsProvider, MultiChainAddress, SpotChain};

async fn token_with_retries(
    provider: &MetastreamsProvider,
    address: &MultiChainAddress,
    attempts: u32,
) -> metastreams_sdk::Result<UiSpotToken> {
    let mut attempt = 0;
    loop {
        match provider.token_details(SpotChain::Solana, address).await {
            Err(error) if error.is_transient() && attempt + 1 < attempts => {
                let delay = match &error {
                    MetastreamsError::RateLimited { retry_after: Some(wait), .. } => *wait,
                    _ => Duration::from_millis((1000 * 2u64.pow(attempt.min(5))).min(30_000)),
                };
                tokio::time::sleep(delay).await;
                attempt += 1;
            }
            result => return result,
        }
    }
}
```

The sample also needs the `time` feature of `tokio`.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.