Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.lasersell.io/llms.txt

Use this file to discover all available pages before exploring further.

​
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")

​
Mirror Trading

Configure watch wallets to mirror trades from other wallets. When a watched wallet opens a position on a supported market, the stream sends a mirror_buy_signal with an unsigned transaction for your wallet to execute. 
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():
    wallet_pubkey = "YOUR_WALLET_PUBKEY"

    client = StreamClient("YOUR_API_KEY")
    session = await StreamSession.connect(
        client,
        StreamConfigure(
            wallet_pubkeys=[wallet_pubkey],
            strategy={
                "target_profit_pct": 50.0,
                "stop_loss_pct": 25.0,
            },
            watch_wallets=[
                {
                    "pubkey": "WatchedWalletPubkey...",
                    "auto_buy": {
                        "wallet_pubkey": wallet_pubkey,
                        "amount_quote_units": 100_000_000,
                        "amount_usd1_units": None,
                    },
                    "mirror_sell": False,
                },
            ],
            mirror_config={
                "max_positions_per_wallet": 1,
                "cooldown_sec": 30,
                "max_active_sol": 5.0,
                "buy_slippage_bps": 2500,
            },
        ),
    )

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

        if event.type == "mirror_buy_signal":
            msg = event.message
            signed = sign_unsigned_tx(str(msg["unsigned_tx_b64"]), signer)
            sig = await send_transaction(SendTargetHeliusSender(), signed)
            session.sender().mirror_buy_result(str(msg["mint"]), True)
            print(f"Mirror buy submitted: {sig}")

        if event.type == "mirror_buy_failed":
            msg = event.message
            print(f"Mirror buy failed: {msg['reason']}")

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

asyncio.run(main())
The watch_wallets list specifies which wallets to mirror. Each entry includes an auto_buy configuration that controls how much to spend when mirroring a trade, and a mirror_sell flag that determines whether to also mirror sell transactions. The mirror_config dict sets global limits: max_positions_per_wallet caps concurrent mirrored positions, cooldown_sec enforces a delay between mirror buys, max_active_sol limits total SOL exposure, and buy_slippage_bps sets the slippage tolerance for mirror buy transactions.

​
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.