Worker API Reference

Complete reference for the ctx object available in worker code.

Code Structure

def setup(ctx):
    """Called once when the worker starts."""
    pass

def tick(ctx):
    """Called every tick_interval seconds. Required."""
    pass

def on_error(ctx, error):
    """Optional. Called when tick() raises an exception."""
    pass

---

ctx.state -- Persistent Storage

Redis-backed key-value store. Data persists across restarts and redeployments. Values are JSON-serialized.

Methods

Method	Signature	Description
get	get(key: str, default=None) -> Any	Get value by key. Returns default if not found.
set	set(key: str, value: Any) -> None	Set a value. Must be JSON-serializable.
delete	delete(key: str) -> None	Delete a key.
keys	keys() -> list[str]	List all state keys for this worker.
get_all	get_all() -> dict	Get all key-value pairs.

Example

def setup(ctx):
    ctx.state.set("prices", [])
    ctx.state.set("config", {"threshold": 5.0, "symbol": "BTCUSDT"})

def tick(ctx):
    config = ctx.state.get("config", {})
    prices = ctx.state.get("prices", [])

    # Append new price
    prices.append({"time": "2024-01-01", "price": 42000})
    ctx.state.set("prices", prices[-100:])  # keep last 100

    # List all keys
    all_keys = ctx.state.keys()  # ["prices", "config"]

---

ctx.log -- Logging

Batched logging system. Logs are flushed automatically after each tick. Visible in the worker's Log tab.

Methods

Method	Signature	Description
info	info(msg: str)	Info level message
warn	warn(msg: str)	Warning level message
error	error(msg: str)	Error level message
debug	debug(msg: str)	Debug level message
flush	flush()	Force flush logs (usually automatic)

Example

def tick(ctx):
    ctx.log.info("Starting tick")
    ctx.log.debug(f"State keys: {ctx.state.keys()}")

    try:
        result = do_something()
        ctx.log.info(f"Success: {result}")
    except Exception as e:
        ctx.log.error(f"Failed: {e}")
        ctx.log.warn("Will retry next tick")

TIP

Logs are batched (up to 20 entries) and sent in a single HTTP call for performance. Messages are truncated at 4000 characters.

---

ctx.monitor -- Real-time Dashboard

Send structured widget data to connected Monitor block(s). Use this to build live dashboards.

Requires: Monitor block connected via edge.

Methods

Method	Signature	Description
render	render(widgets: list[dict])	Replace the entire display with new widgets
patch	patch(updates: list[dict])	Update specific widgets by id
log	log(line: str, level: str = "info")	Append a line to the monitor log
clear	clear()	Clear the display

Widget Builders

Use these static methods to build widget dicts:

Builder	Signature	Description
metric	metric(label, value, color="blue", icon=None, id=None, subtitle=None)	Metric card
status	status(label, state, detail="", id=None)	Status indicator. state: "ok", "warning", "error"
table	table(headers: list, rows: list[list], id=None)	Data table
text	text(content, style="normal", id=None)	Text block. style: "normal" or "monospace"
progress	progress(label, value: float, color="gold", id=None)	Progress bar (0.0 to 1.0)
list_widget	list_widget(items: list, ordered=False, id=None)	Bullet or numbered list
log_widget	log_widget(lines: list, max_lines=50, id=None)	Scrolling log widget
separator	separator()	Visual divider line

Colors

Available colors for metric and progress: "blue", "green", "red", "gold", "orange", "purple", "gray".

Example

def tick(ctx):
    price = 42150.50
    change = +2.35

    ctx.monitor.render([
        ctx.monitor.metric("BTC Price", f"${price:,.2f}",
                          color="green" if change >= 0 else "red",
                          id="btc_price"),
        ctx.monitor.metric("24h Change", f"{change:+.2f}%",
                          color="green" if change >= 0 else "red"),
        ctx.monitor.separator(),
        ctx.monitor.status("Exchange", "ok", "Bybit connected"),
        ctx.monitor.status("Bot", "warning", "Waiting for signal"),
        ctx.monitor.separator(),
        ctx.monitor.table(
            headers=["Symbol", "Side", "Size", "PnL"],
            rows=[
                ["BTCUSDT", "Long", "0.5", "+$120"],
                ["ETHUSDT", "Short", "2.0", "-$45"],
            ]
        ),
        ctx.monitor.progress("Daily Target", 0.65, color="gold"),
    ])

Partial Updates with patch()

Use patch() to update specific widgets without re-rendering everything:

# First render with IDs
ctx.monitor.render([
    ctx.monitor.metric("Price", "$42,000", id="price"),
    ctx.monitor.status("Bot", "ok", "Running", id="bot_status"),
])

# Later, update only the price
ctx.monitor.patch([
    {"id": "price", "value": "$42,150"},
])

---

ctx.http -- HTTP Requests

Make HTTP requests to external APIs. All requests are routed through an AiSpinner-managed European egress proxy with rotating IPs (Cloud Run, region europe-west1). This protects against exchange-side IP-concentration rate limits and keeps third-party providers happy when many users share the platform — you don't need to configure anything.

Methods

Method	Signature	Returns
get	get(url: str, headers: dict = None) -> dict	{"status": int, "body": str, "url": str, "content_type": str}
post	post(url: str, data: dict = None, headers: dict = None) -> dict	{"status": int, "body": str, "url": str}

Example

def tick(ctx):
    # Fetch JSON API
    resp = ctx.http.get("https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd")
    body = resp.get("body", {})

    # body might be a string — parse it
    if isinstance(body, str):
        import json
        body = json.loads(body)

    price = body.get("bitcoin", {}).get("usd", 0)
    ctx.log.info(f"BTC price: ${price}")

    # POST request with custom headers
    resp = ctx.http.post(
        "https://hooks.slack.com/services/xxx",
        data={"text": f"BTC is ${price}"},
        headers={"Content-Type": "application/json"}
    )

---

ctx.bybit -- Bybit Exchange

Trade on Bybit. 32 methods covering market data, orders, positions, risk management, and account operations across spot, derivatives, and options.

Requires: Bybit block connected via edge, with API keys configured.

📖 Full Bybit SDK reference →

def tick(ctx):
    last = float(ctx.bybit.get_tickers("BTCUSDT")["list"][0]["lastPrice"])
    if last > 70000:
        ctx.bybit.place_order("BTCUSDT", "Sell", "0.001", "Market")

---

ctx.ig -- IG Markets

Trade CFDs, forex, indices, commodities, and equities on IG Markets. 22 methods including pending working orders, historical prices, client sentiment, account activity, and Lightstreamer streaming. Server-only.

📖 Full IG Markets SDK reference →

def tick(ctx):
    # Fade extreme retail sentiment
    s = ctx.ig.get_client_sentiment(market_id="EURUSD")
    if float(s.get("longPositionPercentage", 50)) > 75:
        ctx.ig.open_position(
            epic="CS.D.EURUSD.TODAY.IP", direction="SELL", size=1.0,
            stop_distance=40, limit_distance=80,
        )

---

ctx.binance -- Binance Exchange

Trade on Binance. 39 methods covering spot, USDⓈ-M futures, market data, risk management, and account analytics.

Requires: Binance block connected via edge, with API keys configured.

📖 Full Binance SDK reference →

def tick(ctx):
    px = float(ctx.binance.get_ticker("BTCUSDT")["lastPrice"])
    if px > 70000:
        ctx.binance.place_order("BTCUSDT", "SELL", "0.001", "MARKET")

---

ctx.coinbase -- Coinbase Advanced Trade

Spot, perpetuals (INTX), and CFM futures on Coinbase. 20 methods including order preview, kill switch, and portfolio management. Server-only.

Requires: Coinbase block connected via edge, with API keys configured.

📖 Full Coinbase SDK reference →

def tick(ctx):
    # Always preview before trading on Coinbase — fees vary by tier
    preview = ctx.coinbase.preview_order("BTC-USD", "BUY", "100", "market")
    if not preview.get("errs"):
        ctx.coinbase.place_order("BTC-USD", "BUY", "100", "market")

---

ctx.okx -- OKX Exchange

Spot, perpetual swaps, dated futures, and options on OKX. 25 methods including algo orders (TP/SL/OCO/trailing/iceberg/TWAP). Server-only.

Requires: OKX block connected via edge, with API keys configured.

📖 Full OKX SDK reference →

def tick(ctx):
    last = float(ctx.okx.get_ticker("BTC-USDT")["data"][0]["last"])
    ctx.okx.place_algo_order(
        "BTC-USDT-SWAP", side="sell", sz="1", ord_type="oco",
        tp_trigger_px=str(last * 1.03), tp_ord_px="-1",
        sl_trigger_px=str(last * 0.98), sl_ord_px="-1",
    )

---

ctx.kraken -- Kraken Exchange

Spot + margin trading on Kraken with REST + WebSocket v2 acceleration. 22 methods including kill switch, batch ops, and system-status guard. Server-only.

Requires: Kraken block connected via edge, with API keys configured.

📖 Full Kraken SDK reference →

def tick(ctx):
    if ctx.kraken.get_system_status().get("status") != "online":
        return  # Kraken in cancel_only / maintenance — wait it out
    px = float(ctx.kraken.get_ticker("XBTUSD")["XXBTZUSD"]["c"][0])
    ctx.kraken.place_order("XBTUSD", "buy", 0.001, "market")

---

ctx.deribit -- Deribit Options & Futures

Crypto options, futures, perpetuals + multi-leg combo trading. 38 methods. Server-only (Deribit blocks browser CORS).

Requires: Deribit block connected via edge, with API keys configured.

📖 Full Deribit SDK reference →

def setup(ctx):
    ctx.deribit.enable_cancel_on_disconnect("account")  # safety net

def tick(ctx):
    iv = ctx.deribit.get_ticker("BTC-27JUN25-100000-C")["mark_iv"]
    if iv > 80:
        ctx.deribit.place_order("BTC-27JUN25-100000-C", "sell", 1, "market")

---

ctx.hyperliquid -- Hyperliquid DEX

On-chain perpetuals DEX. Read-only by design — wallet stays in the browser, so workers can read every state endpoint (positions, fills, orders, mids, candles, vaults) but cannot sign trades. Worker emits a signal → canvas block executes via wallet. 21 read-only methods. Server-only.

📖 Full Hyperliquid SDK reference →

def tick(ctx):
    state = ctx.hyperliquid.get_account_state()
    mids = ctx.hyperliquid.get_all_mids()
    if float(mids["BTC"]) > 70000:
        ctx.emit("hl_signal", {"coin": "BTC", "side": "sell", "size": 0.001})

---

ctx.bitget -- Bitget Exchange

Spot + USDT/USDC/COIN-margined perpetuals on Bitget. 15 methods across public market data and authenticated trading. Server-only.

Requires: Bitget block connected via edge, with API keys configured.

📖 Full Bitget SDK reference →

def tick(ctx):
    ctx.bitget.futures_set_leverage("BTCUSDT", 5)
    ctx.bitget.futures_place_order(
        symbol="BTCUSDT", side="buy", size="0.01",
        order_type="market",
    )

---

ctx.ibkr -- Interactive Brokers

Stocks, options, futures, forex, and bonds through Interactive Brokers via Client Portal Gateway. 19 methods including market-data snapshots, historical bars, order modify/reply, multi-currency ledger. Server-only.

📖 Full IBKR SDK reference →

def setup(ctx):
    ctx.ibkr.tickle()
    ctx.ibkr.suppress_messages(["o163"])  # skip fat-finger prompts
    ctx.state.set("aapl_conid", ctx.ibkr.search_contract("AAPL")[0]["conid"])

def tick(ctx):
    ctx.ibkr.tickle()  # keepalive
    ctx.ibkr.place_order(
        account_id="U1234567",
        conid=ctx.state.get("aapl_conid"),
        side="BUY", quantity=10, order_type="MKT",
    )

---

ctx.polymarket -- Polymarket Prediction Markets

On-chain prediction markets via Polymarket. 19 methods covering Gamma discovery, CLOB market data, and signed trading. Server-only.

📖 Full Polymarket SDK reference →

def tick(ctx):
    # Find busiest events of the day
    events = ctx.polymarket.search_events("", limit=200)
    busy = sorted(events, key=lambda e: float(e.get("volume24hr", 0)),
                  reverse=True)[:5]
    for e in busy:
        ctx.log.info(f"{e['title']}: ${e['volume24hr']:,.0f}")

---

ctx.telegram -- Telegram Messaging

Send and receive messages via Telegram bot.

Requires: Telegram block connected via edge, with bot token and chat ID configured.

Methods

Method	Signature	Description
send	send(text: str, parse_mode: str = "HTML") -> dict	Send a message
get_messages	get_messages(limit: int = 10) -> dict	Get recent messages
get_chat_info	get_chat_info() -> dict	Get chat details (title, type)

parse_mode

• "HTML" -- supports <b>bold</b>, <i>italic</i>, <code>code</code>, <pre>block</pre>, <a href="url">link</a>
• "Markdown" -- supports *bold*, _italic_, `code`

Example

def tick(ctx):
    price = 42000
    change = +2.5

    # Send formatted alert
    ctx.telegram.send(
        f"<b>BTC Update</b>\n"
        f"Price: <code>${price:,}</code>\n"
        f"Change: {change:+.1f}%",
        parse_mode="HTML"
    )

    # Read recent messages
    messages = ctx.telegram.get_messages(limit=5)
    for msg in messages.get("messages", []):
        text = msg.get("text", "")
        ctx.log.info(f"Received: {text}")

---

ctx.files -- File Storage

Read and write files in a sandboxed file system (100 MB limit per workspace).

Requires: File Explorer block connected via edge.

Methods

Method	Signature	Description
list	list(path: str = "/") -> dict	List files and dirs. Returns {"files": [...], "dirs": [...]}
read	read(path: str) -> dict	Read file (max 50 KB). Returns {"content": "..."}
write	write(path: str, content: str) -> dict	Write/create file. Returns {"ok": true, "path": "..."}
delete	delete(path: str) -> dict	Delete file or dir. Returns {"ok": true}

Example

def tick(ctx):
    import json
    import datetime

    # Save data to file
    data = {"price": 42000, "time": str(datetime.datetime.now())}
    ctx.files.write("/logs/latest.json", json.dumps(data, indent=2))

    # Read it back
    result = ctx.files.read("/logs/latest.json")
    content = result.get("content", "")
    ctx.log.info(f"Saved: {content}")

    # List directory
    listing = ctx.files.list("/logs/")
    ctx.log.info(f"Files: {listing.get('files', [])}")

---

ctx.llm -- LLM Queries

Send prompts to connected AI models and get text responses. Supports five providers: Claude (Anthropic), Grok (xAI), GPT (OpenAI), DeepSeek, Groq.

Requires: at least one LLM Agent block (Claude Agent, Grok Agent, ChatGPT, DeepSeek, or Groq) connected via edge.

Methods

Method	Signature	Description
ask	ask(message, system="", model="", max_tokens=2048, temperature=0.7) -> str	Use the default connected LLM block
claude	claude(message, system="", model="", max_tokens=2048, temperature=0.7) -> str	Force Claude (requires Claude Agent connected)
grok	grok(message, ...) -> str	Force Grok (requires Grok Agent connected)
deepseek	deepseek(message, ...) -> str	Force DeepSeek
groq	groq(message, ...) -> str	Force Groq
ask_with_tools	ask_with_tools(message, system="", model="", max_tokens=4096) -> str	Ask with tool access (trading, telegram, etc.)

Multi-provider routing

You can connect a Worker to several LLM blocks at once and call them by name: ctx.llm.claude(...) for vision/reasoning, ctx.llm.groq(...) for ultra-fast latency-sensitive tasks, ctx.llm.deepseek(...) for cost-efficient reasoning. Use ctx.llm.ask(...) to use the first connected block as default.

Parameters

Parameter	Type	Default	Description
message	str	required	Your prompt / question
system	str	""	System prompt (instructions for the AI)
model	str	""	Model override (leave empty for block default)
max_tokens	int	2048 / 4096	Maximum response length
temperature	float	0.7	Creativity (0.0 = deterministic, 1.0 = creative)

Example

def tick(ctx):
    # Simple question
    answer = ctx.llm.ask(
        "Summarize the current BTC market sentiment in 2 sentences.",
        system="You are a cryptocurrency analyst.",
        temperature=0.3
    )
    ctx.log.info(f"AI says: {answer}")

    # Use with trading data
    positions = ctx.bybit.get_positions("BTCUSDT")
    analysis = ctx.llm.ask(
        f"Analyze these positions and suggest actions: {positions}",
        system="You are a quantitative trading advisor. Be concise."
    )

    # Send analysis to Telegram
    ctx.telegram.send(f"<b>AI Analysis:</b>\n{analysis}", parse_mode="HTML")

---

ctx.connected_blocks -- Connection Info

Read-only dictionary showing which blocks are connected to this worker.

Example

def setup(ctx):
    blocks = ctx.connected_blocks
    ctx.log.info(f"Connected: {blocks}")
    # Example output:
    # {
    #   "trading_node_id": "bybit_1",
    #   "trading_block_type": "trading.bybit",
    #   "llm_node_id": "claude_1",
    #   "llm_block_type": "ai.claude_agent",
    #   "monitor_node_ids": ["monitor_1"],
    #   "telegram_node_id": "telegram_1",
    #   "file_explorer_node_id": "files_1",
    #   "workspace_id": 42
    # }

    if not blocks.get("trading_node_id"):
        ctx.log.warn("No trading block connected!")

---

Error Handling Pattern

def tick(ctx):
    try:
        ticker = ctx.bybit.get_tickers("BTCUSDT")

        # Check for API errors
        if "error" in ticker:
            ctx.log.error(f"API error: {ticker['error']}")
            return

        price = ticker.get("last_price", "0")
        ctx.log.info(f"BTC: {price}")

    except RuntimeError as e:
        # Block not connected
        ctx.log.error(f"Missing connection: {e}")
    except Exception as e:
        # Unexpected error -- will trigger on_error() if defined
        raise

def on_error(ctx, error):
    ctx.log.error(f"Tick failed: {error}")
    ctx.telegram.send(f"Worker error: {error}")

WARNING

All API methods return dict results. Always check for an "error" key in the response before using the data.