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

# Python SDK

> Install, configure, and use the LaserSell Python SDK with async patterns for building transactions and streaming positions.

## Installation

```bash theme={null}
pip install lasersell-sdk
```

Optional extras for additional features:

```bash theme={null}
# 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 Path                    | Purpose                                                               |
| ------------------------------ | --------------------------------------------------------------------- |
| `lasersell_sdk.exit_api`       | `ExitApiClient`, request/response types, `ExitApiError`               |
| `lasersell_sdk.tx`             | `sign_unsigned_tx`, `send_transaction`, `SendTarget*` classes         |
| `lasersell_sdk.stream.client`  | `StreamClient`, `StreamConfigure`, `StreamSender`                     |
| `lasersell_sdk.stream.session` | `StreamSession`, `StreamEvent`, `PositionHandle`                      |
| `lasersell_sdk.stream.proto`   | Message types (`ServerMessage`, `ClientMessage`, `StrategyConfigMsg`) |

## API Client

All API client methods are `async` and must be awaited.

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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"]`).

```python theme={null}
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

<Note>
  **Tier 1+ only.** Requires a Professional or Advanced subscription. See the [announcement](https://www.lasersell.io/blog/liquidity-snapshots-and-sdk-0-3) for full details.
</Note>

`StreamSession` caches the latest liquidity snapshot per position. Query slippage bands, maximum sellable amounts, and liquidity trends:

```python theme={null}
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:

```python theme={null}
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.

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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](/api/quickstart) for a complete build, sign, and submit workflow.