agentbroker 1.1.0

Python SDK for AgentBroker — programmatic crypto trading for autonomous agents

AgentBroker Python SDK

Official Python SDK for AgentBroker — the non-custodial crypto trading platform built for autonomous agents on Jupiter DEX (Solana).

Features

• Async-first client (AgentBrokerClient) built on httpx — runs cleanly under asyncio, FastAPI, Discord bots, etc.
• Sync wrapper (AgentBroker) — backward-compatible with the existing v1.x API.
• Non-custodial trading — your wallet signs the Jupiter swap; AgentBroker never holds your private key.
• Typed response models — every API response is parsed into a dataclass you can autocomplete against.
• Typed exceptions — AuthenticationError, NotFoundError, RateLimitError, ServerError, etc. — all subclasses of AgentBrokerError.
• WebSocket helpers — decorator-style event handlers with auto-reconnect (optional agentbroker[ws] extra).

Installation

pip install agentbroker

With WebSocket support:

pip install "agentbroker[ws]"

Quickstart (5 lines)

pip install agentbroker → working trade in 5 lines of Python:

from agentbroker import AgentBrokerClient
import asyncio
client = AgentBrokerClient(api_key="ab_live_...")
async def main():
    acct = await client.get_account()
    trade = await client.trade("SOL", "USDC", amount_sol=0.1)
asyncio.run(main())

That's it — the asyncio plumbing is already wired up.

Sync wrapper (backward-compatible)

Existing v1.x users keep working unchanged:

from agentbroker import AgentBroker
client = AgentBroker(api_key="ab_live_...")
acct   = client.get_account()
order  = client.place_order("BTC-USDC", "buy", "market", quantity=0.001)

Both classes ship from agentbroker — pick the one that matches your runtime.

Non-custodial registration (new agents)

import asyncio
from agentbroker import AgentBrokerClient

async def register():
    async with AgentBrokerClient() as client:
        agent = await client.register(
            wallet_address="9B5X5AF5zxkXk7gw7Xj7t3rK4vN2mL1pQ8sR6tU5vW4",
            name="my-trading-bot",
            email="[e-mail]",
        )
        print(agent.api_key)   # ← save this — shown ONCE

asyncio.run(register())

The returned api_key authenticates every subsequent request. The agent's own Solana wallet signs and submits the unsigned trade() output — AgentBroker never touches your private key.

Features in depth

• Full API coverage — registration (legacy + non-custodial), orders, deposits, market data, WebSocket streaming.
• Type hints throughout — autocomplete-friendly with dataclass response models (Agent, Account, DiscoveredToken, TradeResult, ...).
• Smart error handling — typed exceptions for auth failures, insufficient balance, not found, rate limit, server errors, etc.

---

API Reference

AgentBrokerClient(api_key=None, base_url=None, timeout=30.0)

Async client. Use this for all new code.

client = AgentBrokerClient(api_key="ab_live_...", timeout=30.0)
async with client:
    ...

Parameter	Type	Default	Description
api_key	str	None	API key (required for authenticated endpoints; not needed for register or discover)
base_url	str	https://agentbroker.polsia.app	API base URL
timeout	float	30.0	HTTP request timeout in seconds

---

Registration & Account

Method	Returns	Description
await client.register(wallet_address, name=None, email=None, ...)	Agent	Register a non-custodial agent using your own Solana wallet (no API key needed)
await client.get_account()	Account	Balance + totals summary
await client.get_balance()	float	Current USDC balance, as a float

# Non-custodial register
agent = await client.register(
    wallet_address="9B5X5AF5zxkXk7gw7Xj7t3rK4vN2mL1pQ8sR6tU5vW4",
    name="my-bot",
    email="[e-mail]",
)
print(agent.api_key)   # save it!

# Account
acct = await client.get_account()
print(acct.balance_usdc)

---

Discovery

Method	Returns	Description
await client.discover(limit=20, strategy='trending')	list[DiscoveredToken]	Trending tokens + recent launches + recently-active

tokens = await client.discover(limit=10)
for t in tokens[:5]:
    print(f"{t.symbol} ({t.type}) — price: {t.price} — liq: {t.liquidity_usd}")

---

Trading (non-custodial)

Method	Returns	Description
await client.trade(token_in, token_out, amount_sol, slippage_bps=50, side=None)	TradeResult	Build an unsigned Jupiter swap — your wallet signs + submits

side is auto-detected from the symbol pair (SOL→USDC = buy, USDC→SOL = sell); pass it explicitly to override. The result contains a base64 unsigned_transaction ready for your wallet to sign, plus the Jupiter quote, expected_output_amount, and the trade_id you'll later confirm via POST /v1/trades/swap-confirm.

trade = await client.trade(
    token_in="SOL",
    token_out="USDC",
    amount_sol=0.1,
    slippage_bps=50,     # 0.5% default
)
print(f"trade_id {trade.trade_id}: {trade.unsigned_transaction[:40]}…")
print(f"expected out: {trade.expected_output_amount} atomic units")

To execute end-to-end: sign unsigned_transaction in your wallet, broadcast on Solana mainnet, then call swap-confirm with the resulting signature.

---

Trades (history)

Method	Returns	Description
await client.get_trades(base_symbol=None, limit=50)	list[Trade]	Recent trades for this agent

---

WebSocket

ws = await client.connect_ws()  # requires agentbroker[ws]

@ws.on_trade
def on_trade(event): ...

ws.run_forever()

Channels: "trades", "balance", "orderbook", "prices".

---

Exceptions

from agentbroker.exceptions import (
    AgentBrokerError,        # Base class
    AuthenticationError,     # 401 — missing/invalid API key
    ValidationError,         # 400 — bad request params
    InsufficientBalanceError,# 400 with code INSUFFICIENT_BALANCE
    NotFoundError,           # 404 — resource not found
    RateLimitError,          # 429 — too many requests
    ServerError,             # 5xx — server-side error
    ConnectionError,         # Network failure
    TimeoutError,            # Request timed out
    WebSocketError,          # WS connection issues
)

try:
    trade = await client.trade("SOL", "USDC", amount_sol=10_000)
except InsufficientBalanceError as e:
    print(f"Need {e.required} USDC, have {e.available} USDC")
except AuthenticationError:
    print("Check your API key")
except AgentBrokerError as e:
    print(f"[{e.code}] {e.message}")

---

Legacy sync API (AgentBroker)

If you're not running under asyncio, the AgentBroker class is a thin synchronous wrapper around AgentBrokerClient (v1.x users keep their existing code).

from agentbroker import AgentBroker

client = AgentBroker(api_key="ab_live_...")

# Same names, same dataclasses — backward-compatible
profile = client.get_profile()
order   = client.place_order("BTC-USDC", "buy", "market", quantity=0.001)

The original classmethod registration path is preserved:

agent = AgentBroker.register(
    name="quickstart-bot",
    email="[e-mail]",
    preferred_pairs=["BTC-USDC", "ETH-USDC"],
)
print(agent.api_key)

---

Examples

File	Description
examples/async_basics.py	Register / account / discover / trade (async)
examples/basic_trade.py	Register, deposit, place market + limit orders (sync)
examples/momentum_bot.py	Trend-following bot with stop-loss + WebSocket balance updates
examples/arbitrage_scanner.py	Scan order books for wide spreads + triangular arb signals

Run any example:

export AGENTBROKER_API_KEY=ab_live_your_key_here
python examples/async_basics.py

---

Environment Variables

Variable	Description
AGENTBROKER_API_KEY	Your API key (used by examples)
AGENTBROKER_BASE_URL	Override base URL (default: https://agentbroker.polsia.app)

---

Development

git clone https://github.com/Polsia-Inc/agentbroker
cd agentbroker/python-sdk

pip install -e ".[dev]"

# Run tests
pytest

# Format
black agentbroker/ examples/
isort agentbroker/ examples/

# Type check
mypy agentbroker/

# Build wheel + sdist
python -m build

---

License

MIT © AgentBroker