Python SDK

from blockrun_llm import LLMClient

client = LLMClient(
    private_key="0x...",                # Wallet key (or use env var / ~/.blockrun/.session)
    api_url="https://blockrun.ai/api",  # Optional
    timeout=600.0,                      # Chat request timeout in seconds (default 600)
    search_timeout=300.0,               # Timeout when Live Search is enabled (default 300)
    transaction_log=None,               # True → ./log/, or a directory path (default: BLOCKRUN_TX_LOG)
    max_cost_per_call=None,             # USD ceiling per quote (default: unset)
    max_session_cost=None,              # USD ceiling per client session (default: unset)
)

response = client.chat(
    "openai/gpt-5.5",
    "Explain quantum computing",
    system="You are a physics teacher.",  # Optional system prompt
    max_tokens=500,                        # Optional max output
    temperature=0.7,                       # Optional temperature
    response_format={"type": "json_object"},  # Optional JSON mode (honoured on every model)
    stop=["###"],                          # Optional stop sequence(s), str or list of ≤4
    fallback_models=["openai/gpt-5.4"],    # Optional chain walked on timeout / 429 / 5xx
    search=True,                           # Optional Live Search grounding (uses search_timeout)
)

messages = [
    {"role": "system", "content": "You are helpful."},
    {"role": "user", "content": "What is 2+2?"}
]

result = client.chat_completion(
    "openai/gpt-5.5",
    messages,
    max_tokens=100,
    temperature=0.7,
    top_p=0.9,
    tools=None,          # OpenAI-format tool definitions
    tool_choice=None,    # "auto" | "required" | {"type": "function", ...}
)

print(result.choices[0].message.content)
print(f"Tokens used: {result.usage.total_tokens}")

for chunk in client.chat_completion_stream("openai/gpt-5.5", messages):
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)

models = client.list_models()
for model in models:
    print(f"{model['id']}: ${model['pricing']['input']}/M in, ${model['pricing']['output']}/M out")

address = client.get_wallet_address()
print(f"Paying from: {address}")

print(f"USDC balance: ${client.get_balance():.2f}")   # on-chain balance of the active chain

spent = client.get_spending()                          # this client session only
print(f"Spent ${spent['total_usd']:.4f} across {spent['calls']} calls")

from blockrun_llm import LLMClient, SpendLimitError

client = LLMClient(max_cost_per_call=0.25, max_session_cost=10.00)
# or per-deployment: BLOCKRUN_MAX_COST_PER_CALL / BLOCKRUN_MAX_SESSION_COST

try:
    client.chat("openai/gpt-5.5", "...")
except SpendLimitError as e:
    print(e.scope, e.quoted_usd, e.limit_usd)   # "call" | "session"

from blockrun_llm import LLMClient

client = LLMClient()

result = client.smart_chat("Summarize this changelog entry in one line")

print(result.response)
print(result.model)              # "google/gemini-2.5-flash"
print(result.routing.tier)       # "SIMPLE"
print(result.routing.task_type)  # "chat"
print(result.routing.savings)    # 0.90 (90% savings vs the baseline flagship)

decision = client.route("Prove that the square root of 2 is irrational")

print(decision.model)       # "deepseek/deepseek-v4-pro"
print(decision.tier)        # "REASONING"
print(decision.task_type)   # "reasoning"
print(decision.method)      # "portfolio"
print(decision.candidates)  # ordered chain; smart_chat walks it on a transient failure
print(decision.reasoning)   # human-readable explanation of the pick

result = client.smart_chat_completion(
    [{"role": "user", "content": "Cancel order B-42 using the tool."}],
    tools=[{"type": "function", "function": {"name": "cancel_order", "parameters": {}}}],
    tool_choice="required",
)

print(result.model)                              # "openai/gpt-5-mini" — tool-capable
print(result.routing.task_type)                  # "tool_agent"
print(result.response.choices[0].message.content)

response = client.chat_completion("blockrun/auto", messages)

# Free models only — a paid model can never leak into this profile
result = client.smart_chat("Explain recursion", routing_profile="free")
print(result.model)                 # "nvidia/nemotron-3.5-lightning"
print(result.routing.cost_estimate) # 0.0

# Maximum savings
result = client.smart_chat("Summarize this article: ...", routing_profile="eco")
print(result.model)  # "google/gemini-3.1-flash-lite"

# Premium for critical tasks
result = client.smart_chat("Review this contract for legal issues...", routing_profile="premium")

result = client.smart_chat("Prove that the square root of 2 is irrational")
routing = result.routing

print(routing.model)             # the model that served the request
print(routing.tier)              # "REASONING"
print(routing.task_type)         # "reasoning"
print(routing.method)            # "portfolio" ("rules" for the free profile)
print(routing.router_version)    # "v3-portfolio"
print(routing.confidence)        # 0.85
print(routing.reasoning)         # why this model won
print(routing.candidates)        # ordered candidate chain
print(routing.candidate_scores)  # per-model quality / cost / speed / reliability
print(routing.fallbacks)         # candidates[1:], the runtime retry chain
print(f"${routing.cost_estimate:.4f} vs ${routing.baseline_cost:.4f}")
print(f"Savings: {routing.savings:.0%}")

from blockrun_llm import (
    RoutingProfile,              # Literal["free", "eco", "auto", "premium"]
    RoutingTier,                 # Literal["SIMPLE", "MEDIUM", "COMPLEX", "REASONING"]
    RoutingDecision,             # Full routing details
    CandidateScore,              # One row of routing.candidate_scores
    SmartChatResponse,           # response + model + routing
    SmartChatCompletionResponse, # ChatResponse + model + routing
)

import asyncio
from blockrun_llm import AsyncLLMClient, SolanaLLMClient

# Async, Base
async def main():
    async with AsyncLLMClient() as client:
        result = await client.smart_chat("What's the weather like?", routing_profile="eco")
        print(result.response)

asyncio.run(main())

# Solana — same routing, USDC on Solana
solana = SolanaLLMClient()
print(solana.route("Prove this theorem").model)

pip install "blockrun-llm[solana]"
export SOLANA_WALLET_KEY="..."   # bs58 keypair or seed, ~/.config/solana/id.json array, or 64-byte hex

from blockrun_llm import SolanaLLMClient, AsyncSolanaLLMClient, setup_agent_solana_wallet

client = SolanaLLMClient()                    # SOLANA_WALLET_KEY → ~/.blockrun/.solana-session
client = SolanaLLMClient(private_key="...")   # or pass the key
client = setup_agent_solana_wallet()          # creates ~/.blockrun/.solana-session if missing

print(client.chat("openai/gpt-5.5", "gm Solana"))
img = client.image("a fox in snow", model="openai/gpt-image-2", quality="low")  # quality is Solana-only
print(img.data[0].url)

from blockrun_llm import ImageClient

img = ImageClient()  # timeout defaults to 200s (gpt-image-2 at high res is slow)

# Generate — default model google/nano-banana, default size 1024x1024
res = img.generate("A cute cat astronaut, studio lighting", model="google/nano-banana-pro", size="1024x1024", n=1)
print(res.data[0].url)

# Edit / fusion — pass one data URI, or 2–4 to fuse (OpenAI ≤4, Nano Banana ≤3)
res = img.edit("Place this logo on the t-shirt", image=["data:image/png;base64,...", "data:image/png;base64,..."])
print(res.data[0].url)

from blockrun_llm import VideoClient

vid = VideoClient()  # timeout 360s; submit→poll handled for you (budget 900s, re-signs mid-poll)

res = vid.generate(
    "a red apple spinning on a marble counter",
    model="bytedance/seedance-2.0",
    duration_seconds=5,
    resolution="720p",          # 360p|480p|540p|720p|1080p|1K (Seedance); 4K only on seedance-2.0
    aspect_ratio="16:9",        # adaptive | 16:9 | 9:16 | 1:1 | 4:3 | 3:4 | 21:9 | 9:21
    generate_audio=True,
)
print(res.data[0].url)

from blockrun_llm import MusicClient

music = MusicClient()
res = music.generate("upbeat synthwave, driving bassline", model="minimax/music-2.5+", instrumental=True)
print(res.data[0].url)   # URL expires ~24h; download promptly. ~$0.1585/track
# For vocals: instrumental=False with lyrics="..." (passing both instrumental=True and lyrics raises ValueError)

from blockrun_llm import SpeechClient

tts = SpeechClient()
res = tts.generate("Hello from BlockRun!", model="elevenlabs/flash-v2.5", voice="sarah", response_format="mp3", speed=1.0)
print(res.data[0].url)

# Sound effects (flat $0.0535/generation)
sfx = tts.sound_effect("rain on a tin roof", duration_seconds=6.0)

voices = tts.list_voices()  # free, 60 req/min/IP

from blockrun_llm import SearchClient

search = SearchClient()
res = search.search("latest agent-payments news", sources=["web", "news"], max_results=10)  # web | news
print(res.summary)
for c in res.citations:
    print(c)

from blockrun_llm import PriceClient

px = PriceClient()                              # set require_wallet=False to use only free categories
btc = px.price("crypto", "BTC-USD")             # crypto, fx, commodity are FREE; usstock, stocks are paid
print(btc.price, btc.publishTime)

bars = px.history("crypto", "BTC-USD", resolution="D", from_ts=1700000000, to_ts=1710000000)
symbols = px.list_symbols("crypto", q="ETH", limit=20)

from blockrun_llm import SurfClient

surf = SurfClient()
ranking = surf.call("market/ranking", params={"limit": 20})   # auto GET/POST from the catalog
catalog = surf.endpoints()                                     # static: every path + tier + price

from blockrun_llm import RpcClient

rpc = RpcClient()
res = rpc.call("ethereum", "eth_blockNumber")          # $0.003/call
print(int(res.result, 16), "cache_hit:", res.cache_hit)

# JSON-RPC 2.0 batch — billed $0.003 x N
batch = rpc.batch("polygon", [{"method": "eth_blockNumber"}, {"method": "eth_gasPrice"}])

from blockrun_llm import PhoneClient

phone = PhoneClient()
info  = phone.lookup("+14155552671")          # $0.011 - carrier + line type
fraud = phone.lookup_fraud("+14155552671")    # $0.051 - + SIM-swap / call-forwarding signals
num   = phone.buy_number(country="US", area_code="415")  # $5 / 30 days (settles after Twilio confirms)
phone.renew_number(num["phone_number"])       # $5 / +30 days
phone.list_numbers()                          # $0.003
phone.release_number(num["phone_number"])     # free

from blockrun_llm import VoiceClient

voice = VoiceClient()
call = voice.call(
    to="+14155552671",
    task="Confirm the 3pm dental appointment and offer to reschedule if needed.",
    voice="maya",            # nat | josh | maya | june | paige | derek | florian, or a Bland.ai id
    max_duration=5,          # 1–30 minutes
    language="en-US",
)
print(call["call_id"])
status = voice.get_status(call["call_id"])   # free; transcript + recording_url once completed

from blockrun_llm import PortraitClient, RealFaceClient

# Virtual Portrait — AI character, no liveness check, $0.011 one-time
portrait = PortraitClient()
p = portrait.enroll("My Spokesperson", "https://example.com/character.jpg")
print(p.asset_id)   # ta_xxxxxxxx → pass to VideoClient(real_face_asset_id=...)

# RealFace — real person, requires on-phone liveness check, $0.011
rf = RealFaceClient()
init = rf.init("Jane Doe")            # render init.h5_link as a QR for the subject
rf.wait_for_active(init.group_id)     # blocks until liveness passes (default 180s)
asset = rf.enroll("Jane Doe", "https://example.com/jane.jpg", init.group_id)
print(asset.asset_id)                 # ta_xxxxxxxx

from blockrun_llm import LLMClient
client = LLMClient()

# Modal secure sandbox
sb = client.modal_sandbox_create()
out = client.modal_sandbox_exec(sb["id"], code="print(2+2)")
client.modal_sandbox_status(sb["id"]); client.modal_sandbox_terminate(sb["id"])

# DeFi (DeFiLlama) + DEX (0x)
yields = client.defi_yields(); protocols = client.defi_protocols()
quote  = client.dex_quote(...); gasless = client.dex_gasless_quote(...)

# Wallet helpers
print(client.get_balance())    # USDC on the active chain
print(client.get_spending())   # session totals: {"total_usd": ..., "calls": ...}
print(client.onramp())         # Coinbase on-ramp link

from blockrun_llm import LLMClient

client = LLMClient()

# List Polymarket markets
markets = client.pm("polymarket/markets")

# List Polymarket events
events = client.pm("polymarket/events")

# Get Polymarket trades
trades = client.pm("polymarket/trades")

# Get candlestick data for a specific condition
candles = client.pm("polymarket/candlesticks/0xabc123...")

# Get wallet profile
wallet = client.pm("polymarket/wallet/0x1234...")

# Get wallet P&L
pnl = client.pm("polymarket/wallet/pnl/0x1234...")

# Get Polymarket leaderboard
leaders = client.pm("polymarket/leaderboard")

# List Kalshi markets
kalshi_markets = client.pm("kalshi/markets")

# Get Kalshi trades
kalshi_trades = client.pm("kalshi/trades")

# Get Binance candles for a symbol
btc_candles = client.pm("binance/candles/BTCUSDT")
eth_candles = client.pm("binance/candles/ETHUSDT")

# Cross-venue search (matching-markets was sunset by Predexon 2026-07-20)
results = client.pm("markets/search", q="Fed rate")

# Bulk wallet identity lookup ($0.0085)
batch = client.pm_query("polymarket/wallet/identities", {
    "addresses": ["0xabc...", "0xdef...", "0x123..."],  # up to 200
})

# Cross-venue search (Tier 2)
found     = client.pm("markets/search", q="bitcoin 2026")

# Polymarket keyset pagination (Tier 1)
page      = client.pm_polymarket_markets_keyset(limit="100")
next_page = client.pm_polymarket_events_keyset(pagination_key=page["pagination"]["next_key"])

# Wallet identity & on-chain clustering (Tier 2)
ident   = client.pm_wallet_identity("0xabc...")
batch   = client.pm_wallet_identities(["0xabc...", "0xdef..."])  # up to 200
cluster = client.pm_wallet_cluster("0xabc...")

import asyncio
from blockrun_llm import AsyncLLMClient

async def main():
    async with AsyncLLMClient() as client:
        markets = await client.pm("polymarket/markets")
        events = await client.pm("polymarket/events")
        candles = await client.pm("binance/candles/SOLUSDT")

asyncio.run(main())

from blockrun_llm import SolanaLLMClient

client = SolanaLLMClient()
markets = client.pm("polymarket/markets")

from blockrun_llm import testnet_client

# Create testnet client (uses Base Sepolia)
client = testnet_client()  # Uses BLOCKRUN_WALLET_KEY

# Chat with testnet model
response = client.chat("openai/gpt-oss-20b", "Hello!")
print(response)

# Check testnet USDC balance
balance = client.get_balance()
print(f"Testnet USDC: ${balance:.4f}")

# Verify you're on testnet
print(f"Is testnet: {client.is_testnet()}")  # True

from blockrun_llm import LLMClient

# Configure manually with testnet API URL
client = LLMClient(api_url="https://testnet.blockrun.ai/api")
response = client.chat("openai/gpt-oss-20b", "Hello!")

import asyncio
from blockrun_llm import AsyncLLMClient

async def main():
    async with AsyncLLMClient() as client:
        # Single request
        response = await client.chat("openai/gpt-5.5", "Hello!")

        # Concurrent requests
        tasks = [
            client.chat("openai/gpt-5.5", "What is 2+2?"),
            client.chat("anthropic/claude-sonnet-4.6", "What is 3+3?"),
        ]
        responses = await asyncio.gather(*tasks)

asyncio.run(main())

from blockrun_llm import LLMClient, APIError, PaymentError, SpendLimitError, RetiredEndpointError

client = LLMClient()

try:
    response = client.chat("openai/gpt-5.5", "Hello!")
except SpendLimitError as e:
    print(f"Refused before paying: {e.scope} limit ${e.limit_usd}, quote ${e.quoted_usd}")
except PaymentError as e:
    print(f"Payment failed: {e}")            # e.status_code / e.response carry the gateway's
    # Check your USDC balance                # code + reason when a 402 was rejected
except APIError as e:
    print(f"API error ({e.status_code}): {e}")
    print(f"Details: {e.response}")
except RetiredEndpointError as e:
    print(f"Helper retired upstream: {e}")   # e.g. pm_markets()

class ChatResponse:
    id: str
    object: str
    created: int
    model: str
    choices: List[ChatChoice]
    usage: ChatUsage

class ChatChoice:
    index: int
    message: ChatMessage
    finish_reason: Optional[str]

class ChatMessage:
    role: str                              # "system" | "user" | "assistant" | "tool"
    content: Optional[str]
    tool_calls: Optional[List[ToolCall]]   # assistant tool calls
    tool_call_id: Optional[str]            # tool results

class ChatUsage:
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cache_read_input_tokens: Optional[int]      # prompt-cache hits, when reported
    cache_creation_input_tokens: Optional[int]

from blockrun_llm import LLMClient

client = LLMClient()
messages = [
    {"role": "system", "content": "You are a helpful assistant."}
]

while True:
    user_input = input("You: ")
    if user_input.lower() == "quit":
        break

    messages.append({"role": "user", "content": user_input})
    result = client.chat_completion("openai/gpt-5.5", messages)

    assistant_message = result.choices[0].message.content
    messages.append({"role": "assistant", "content": assistant_message})

    print(f"Assistant: {assistant_message}")

from blockrun_llm import LLMClient

client = LLMClient()

code = client.chat(
    "anthropic/claude-sonnet-4.6",
    "Write a Python function to calculate fibonacci numbers",
    system="You are an expert Python developer. Return only code, no explanations."
)

print(code)

from blockrun_llm import setup_agent_wallet, status, list_discovered_wallets, import_wallet

client = setup_agent_wallet()      # creates ~/.blockrun/.session (0600) if missing, prints address + funding QR
status()                           # "Wallet: 0x…  Balance: $5.30 USDC"

link = client.onramp(client.get_wallet_address())   # one-time Coinbase Onramp link (expires ~5 min)

# Adopt a wallet another application created (never done automatically)
for w in list_discovered_wallets():
    print(w["address"], "from", w["source"])
import_wallet("0x…")               # backs up the current key to ~/.blockrun/.session.backup-<ts> first

pip install "blockrun-llm[anthropic]"

from blockrun_llm import AnthropicClient

client = AnthropicClient()   # same wallet resolution as LLMClient
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.content[0].text)

from blockrun_llm import get_cost_log_summary, export_cost_log_csv, export_cost_log_json

print(get_cost_log_summary())              # grouped by endpoint by default
csv_text = export_cost_log_csv()           # pass output_path=... to also write a file

pytest tests/unit                    # Run unit tests only
pytest tests/unit --cov              # Run with coverage report
pytest tests/unit -v                 # Verbose output

# Set your funded wallet key
export BLOCKRUN_WALLET_KEY=0x...

# Run only integration tests
pytest tests/integration

# Run all tests (unit + integration)
pytest

# .env (add to .gitignore!)
BLOCKRUN_WALLET_KEY=0x...your_private_key_here

# app.py
import os
from blockrun_llm import LLMClient
from dotenv import load_dotenv

load_dotenv()

if not os.getenv("BLOCKRUN_WALLET_KEY"):
    raise ValueError("BLOCKRUN_WALLET_KEY not set")

client = LLMClient()  # Reads from environment

from blockrun_llm import LLMClient, APIError

client = LLMClient()

try:
    response = client.chat('invalid-model', 'Hello')
except APIError as e:
    # Error messages only contain safe, user-facing information
    # No internal stack traces, file paths, or sensitive data
    print(e.message)

client = LLMClient()
address = client.get_wallet_address()
print(f"View transactions: https://basescan.org/address/{address}")

pip install --upgrade blockrun-llm