Skip to main content

​
Installation

pip install lasersell-sdk
Optional extras for additional features: 
# Transaction signing (requires solders)
pip install lasersell-sdk[tx]

# WebSocket stream (requires websockets)
pip install lasersell-sdk[stream]

# Both
pip install lasersell-sdk[tx,stream]

​
Modules

Import PathPurpose
lasersell_sdk.exit_apiExitApiClient, request/response types, ExitApiError
lasersell_sdk.txsign_unsigned_tx, send_transaction, SendTarget* classes
lasersell_sdk.stream.clientStreamClient, StreamConfigure, StreamSender
lasersell_sdk.stream.sessionStreamSession, StreamEvent, PositionHandle
lasersell_sdk.stream.protoMessage types (ServerMessage, ClientMessage, StrategyConfigMsg)

​
API Client

All API client methods are async and must be awaited. 
import asyncio
from lasersell_sdk.exit_api import ExitApiClient, BuildSellTxRequest, SellOutput

async def main():
    client = ExitApiClient.with_api_key("YOUR_API_KEY")

    request = BuildSellTxRequest(
        mint="TOKEN_MINT",
        user_pubkey="WALLET",
        amount_tokens=1_000_000,
        slippage_bps=2_000,
        output=SellOutput.SOL,
    )

    response = await client.build_sell_tx(request)
    print("Unsigned tx:", response.tx)

asyncio.run(main())

​
Buy Transactions

from lasersell_sdk.exit_api import BuildBuyTxRequest

request = BuildBuyTxRequest(
    mint="TOKEN_MINT",
    user_pubkey="WALLET",
    amount=0.1,  # 0.1 SOL
    slippage_bps=2_000,
)

response = await client.build_buy_tx(request)

​
Custom Options

from lasersell_sdk.exit_api import ExitApiClient, ExitApiClientOptions
from lasersell_sdk.retry import RetryPolicy

client = ExitApiClient.with_options(
    "YOUR_API_KEY",
    ExitApiClientOptions(
        attempt_timeout_s=2.0,
        retry_policy=RetryPolicy(
            max_attempts=3,
            initial_backoff_ms=50,
            max_backoff_ms=200,
            jitter_ms=50,
        ),
    ),
)

​
Exit Intelligence Stream Session

event.message is a TypedDict union. At runtime, access fields using dict syntax (e.g. msg["unsigned_tx_b64"]). 
import asyncio
from lasersell_sdk.stream.client import StreamClient, StreamConfigure
from lasersell_sdk.stream.session import StreamSession
from lasersell_sdk.tx import SendTargetHeliusSender, send_transaction, sign_unsigned_tx

async def main():
    client = StreamClient("YOUR_API_KEY")
    session = await StreamSession.connect(
        client,
        StreamConfigure(
            wallet_pubkeys=["WALLET_PUBKEY"],
            strategy={
                "target_profit_pct": 5.0,
                "stop_loss_pct": 1.5,
            },
            deadline_timeout_sec=45,
            send_mode="helius_sender",
            tip_lamports=1000,
        ),
    )

    while True:
        event = await session.recv()
        if event is None:
            break

        if event.type == "position_opened" and event.handle is not None:
            print(f"New position: {event.handle.mint}")

        if event.type == "exit_signal_with_tx":
            msg = event.message  # TypedDict, use dict access
            signed = sign_unsigned_tx(str(msg["unsigned_tx_b64"]), signer)
            sig = await send_transaction(SendTargetHeliusSender(), signed)
            print(f"Exit submitted: {sig}")

asyncio.run(main())

​
Liquidity Snapshots and Partial Sells

Tier 1+ only. Requires a Professional or Advanced subscription. See the announcement for full details.
StreamSession caches the latest liquidity snapshot per position. Query slippage bands, maximum sellable amounts, and liquidity trends: 
bands = session.get_slippage_bands(position_id)
max_tokens = session.get_max_sell_at_slippage(position_id, 500)  # 5% slippage
trend = session.get_liquidity_trend(position_id)  # "growing" | "stable" | "draining"
Use build_partial_sell_tx() to sell a portion of a position based on slippage data: 
max_tokens = session.get_max_sell_at_slippage(position_id, 500)
if max_tokens is not None:
    response = await client.build_partial_sell_tx(handle, max_tokens, 500, "SOL")

​
Transaction Helpers

from lasersell_sdk.tx import (
    sign_unsigned_tx,
    send_transaction,
    send_transaction_b64_to,
    encode_signed_tx,
    SendTargetRpc,
    SendTargetHeliusSender,
    SendTargetAstralane,
)

# Sign
signed_tx = sign_unsigned_tx(unsigned_tx_b64, keypair)

# Send
signature = await send_transaction(SendTargetHeliusSender(), signed_tx)

# Send with custom RPC
signature = await send_transaction(
    SendTargetRpc(url="https://rpc.example.com"),
    signed_tx,
)

# Astralane
signature = await send_transaction(
    SendTargetAstralane(api_key="KEY", region="fr"),
    signed_tx,
)

​
Error Handling

from lasersell_sdk.exit_api import ExitApiError
from lasersell_sdk.tx import TxSubmitError
from lasersell_sdk.stream.client import StreamClientError

try:
    response = await client.build_sell_tx(request)
except ExitApiError as e:
    print(f"Kind: {e.kind}, Retryable: {e.is_retryable()}")
    print(f"Status: {e.status}, Body: {e.body}")
except TxSubmitError as e:
    print(f"Kind: {e.kind}, Target: {e.target}")
except StreamClientError as e:
    print(f"Kind: {e.kind}")

​
Loading a Keypair

import json
from pathlib import Path
from solders.keypair import Keypair

raw = json.loads(Path("./keypair.json").read_text())
keypair = Keypair.from_bytes(bytes(raw))
Requires the solders package, which is included in the [tx] extra.

​
Complete Example

See the Quickstart for a complete build, sign, and submit workflow.