Skip to main content

Official Python SDK for building and deploying agents on the Circuit platform

Project description

Circuit Agent SDK - Python

Clean, unified, type-safe Python SDK for building cross-chain agents on Circuit

A simplified Python SDK for building automated agents to deploy on Circuit. Agents receive a single AgentContext object containing everything they need - request data, SDK methods, and unified logging. No boilerplate, no return values to manage, just write your logic.

💡 Best used with Circuit Agents CLI - Deploy, manage, and test your agents with ease

📑 Table of Contents

🚀 Quick Start

Install the SDK

pip install circuit-agent-sdk
# or with uv
uv pip install circuit-agent-sdk

Create Your First Agent

Every agent receives a single AgentContext object that provides:

Session Data:

  • agent.sessionId - Unique session identifier
  • agent.sessionWalletAddress - The wallet address for this session
  • agent.currentPositions - Assets allocated to the agent at the start of this execution

Available Methods:

  • agent.log() - Send messages to users and log locally
  • agent.memory - Persist data across executions (.set(), .get(), .list(), .delete())
  • agent.platforms.polymarket - Trade prediction markets (.market_order(), .redeem_positions())
  • agent.platforms.hyperliquid - trading (.place_order(), .positions(), .balances(), .transfer())
  • agent.swidge - Cross-chain swaps and bridges (.quote(), .execute())
  • agent.sign_and_send() - Execute custom built transactions on any supported chain
  • agent.sign_message() - Sign messages (EVM only)
  • agent.transactions() - Get transaction history with asset changes

Important: currentPositions reflects your allocated assets at the start of each execution. You will need to use the agent.get_current_positions() method to pull fresh values after executing transactions. It's important to consider the hasPendingTxs if you had recently submitted transactions. If hasPendingTxs is true, the balances you received may not be 100% up to date.

Required Asset setup

Note: For native tokens, use the following null addresses for solana/ethereum

// Requiring 1 SOL
[[requiredAssets]]
network = "solana"
address = "11111111111111111111111111111111"
minimumAmount = "1000000000"

// Requiring 1 ETH
[[requiredAssets]]
network = "ethereum:<chainId>"
address = "0x0000000000000000000000000000000000000000"
minimumAmount = "1000000000000000000"
from agent_sdk import Agent, AgentContext
import time

def run(agent: AgentContext) -> None:
    """
    Main agent logic - receives AgentContext with everything needed.
    No return value - errors are caught automatically.
    """
    # Access session data
    agent.log(f"Starting execution for session {agent.sessionId}")
    agent.log(f"Wallet: {agent.sessionWalletAddress}")
    agent.log(f"Managing {len(agent.currentPositions)} allocated positions")

    # Use SDK methods
    agent.memory.set("last_run", str(time.time()))

    # Your agent logic here - track position changes within this execution
    # Circuit will provide updated positions on the next run

def stop(agent: AgentContext) -> None:
    """Cleanup when agent is stopped."""
    agent.log("Cleaning up resources")
    agent.memory.delete("temp_data")

# Boilerplate - This should never change unless you want to change the names of your run and stop functions
agent = Agent(
    run=run,
    stop=stop
)

handler = agent.get_handler()

if __name__ == "__main__":
    agent.run()

🎯 Core Concepts

The AgentContext Object

Every agent function receives a single AgentContext object that contains:

Request Data:

  • agent.sessionId - Unique session identifier
  • agent.sessionWalletAddress - Wallet address for this session
  • agent.currentPositions - Current positions allocated to this agent

SDK Methods:

  • agent.log() - Unified logging (console + backend)
  • agent.memory - Session-scoped key-value storage
  • agent.platforms.polymarket - Prediction market operations
  • agent.swidge - Cross-chain swap operations
  • agent.sign_and_send() - Sign and broadcast transactions
  • agent.sign_message() - Sign messages on EVM
  • agent.transactions() - Get transaction history with asset changes

Run/Stop Function Requirements

  1. Signature: def my_function(agent: AgentContext) -> None:
  2. Return: Always return None (or no return statement)
  3. Errors: You should surface any relevant errors via agent.log('error message',error=True). All errors from built-in sdk functions will be caught gracefully and provided in the return data for you to handle as necessary.

📝 Unified Logging with agent.log()

Use agent.log() to communicate with your users and debug your agent. Every message appears in your terminal, and by default also shows up in the Circuit UI for your users to see. Simply pass debug=True to skip sending the message to the user.

def run(agent: AgentContext) -> None:
    # Standard log: Shows to user in Circuit UI
    agent.log("Processing transaction")

    # Error log: Shows to user in Circuit UI (as an error)
    result = agent.memory.get("key")
    if not result.success:
        agent.log(result.error_message, error=True)

    # Debug log: Only you see this in your terminal
    agent.log("Internal state: processing...", debug=True)

    # Logging dicts and pydantic models
    # Both are pretty-printed to console and serialized/truncated for backend
    agent.log({"wallet": agent.sessionWalletAddress, "status": "active"})

    # Check if message reached the user
    log_result = agent.log("Important message")
    if not log_result.success:
        # Rare - usually means Circuit UI is unreachable
        pass

What Your Users See:

Code You See (Terminal) User Sees (Circuit UI)
agent.log("msg") ✅ In your terminal ✅ In Circuit UI
agent.log("msg", error=True) ✅ As error in terminal ✅ As error in Circuit UI
agent.log("msg", debug=True) ✅ In terminal ❌ Hidden from user
agent.log("msg", error=True, debug=True) ✅ As error in terminal ❌ Hidden from user

💾 Session Memory Storage

Store and retrieve data for your agent's session with simple operations. Memory is automatically scoped to your session ID, and for now is simple string storage. You will need to handle serialization of whatever data you want to store here.

def run(agent: AgentContext) -> None:
    # Set a value
    result = agent.memory.set("lastSwapNetwork", "ethereum:42161")
    if not result.success:
        agent.log(result.error_message, error=True)

    # Get a value
    result = agent.memory.get("lastSwapNetwork")
    if result.success and result.data:
        network = result.data.value
        agent.log(f"Using network: {network}")
    else:
        agent.log("No saved network found")

    # List all keys
    result = agent.memory.list()
    if result.success and result.data:
        agent.log(f"Found {result.data.count} keys: {result.data.keys}")

    # Delete a key
    result = agent.memory.delete("tempData")

All memory operations return responses with .success and .error_message:

result = agent.memory.set("key", "value")
if not result.success:
    agent.log(f"Failed to save: {result.error_message}", error=True)

🌉 Cross-Chain Swaps with Swidge

Built-in Swidge integration for seamless cross-chain token swaps and bridges.

Get and execute a quote

Note: It is important to always validate quotes before executing. Circuit will always do its best to return a quote, and as of now, will only filter out quotes with price impacts exceeding 100% to ensure maximum flexibility. It is on the agent to makes sure a quote is valid, given its own parameters

Bulk Execution: execute() accepts both single quotes and lists. Pass a list to execute multiple swaps in parallel: agent.swidge.execute([quote1.data, quote2.data]) returns a list of results. Note: Use explicit is not None checks for proper type narrowing.

def run(agent: AgentContext) -> None:
    # 1. Get quote
    quote = agent.swidge.quote({
        "from": {"network": "ethereum:8453", "address": agent.sessionWalletAddress},
        "to": {"network": "ethereum:137", "address": agent.sessionWalletAddress},
        "amount": "1000000000000000",  # 0.001 ETH
        "toToken": "0x2791bca1f2de4661ed88a30c99a7a9449aa84174",
        "slippage": "2.0",
        "priceImpact": "100.0"
    })

    if not quote.success:
        _ = agent.log(f"Quote failed: {quote.error_message}", error=True)
        return

    if abs(float(quote.data.priceImpact.percentage)) > 10:
        _ = agent.log(f"Warning: Price impact is too high: {quote.data.priceImpact.percentage}%", error=True, debug=True)
    else:
        _ = agent.log(f"You'll receive: {quote.data.assetReceive.amountFormatted}")
        _ = agent.log(f"Fees: {', '.join([f.name for f in quote.data.fees])}")

        # 2. Execute the swap
        result = agent.swidge.execute(quote.data)

        if result.success and result.data:
            agent.log(f"Swap status: {result.data.status}")
            if result.data.status == "success":
                agent.log("✅ Swap completed!")
                agent.log(f"In tx: {result.data.in_.txs}")
                agent.log(f"Out tx: {result.data.out.txs}")
            elif result.data.status == "failure":
                agent.log("❌ Swap failed", error=True)
        else:
            agent.log(result.error_message, error=True)

📈 Polymarket Prediction Markets

Trade prediction markets on Polygon.

Place Market Orders

Note: Right now, polymarket's API accepts different decimal precision for buys and sells, this will result in dust positions if you are selling out of a position before expiry. Once expired, dust can be cleaned up during the redeem step.

def run(agent: AgentContext) -> None:
    # Buy order - size is USD amount to spend
    buy_order = agent.platforms.polymarket.market_order({
        "tokenId": "86192057611122246511563653509192966169513312957180910360241289053249649036697",
        "size": 3,  # Spend $3
        "side": "BUY"
    })

    if buy_order.success and buy_order.data:
        agent.log(f"Order ID: {buy_order.data.orderInfo.orderId}")
        agent.log(f"Price: ${buy_order.data.orderInfo.priceUsd}")
        agent.log(f"Total: ${buy_order.data.orderInfo.totalPriceUsd}")
    else:
        agent.log(buy_order.error_message, error=True)

    # Sell order - size is number of shares to sell
    sell_order = agent.platforms.polymarket.market_order({
        "tokenId": "86192057611122246511563653509192966169513312957180910360241289053249649036697",
        "size": 5.5,  # Sell 5.5 shares
        "side": "SELL"
    })

Redeem Positions

def stop(agent: AgentContext) -> None:
    """Redeem all settled positions."""
    redemption = agent.platforms.polymarket.redeem_positions()

    if redemption.success and redemption.data:
        successful = [r for r in redemption.data if r.success]
        agent.log(f"Redeemed {len(successful)} positions")
    else:
        agent.log(redemption.error_message, error=True)

Hyperliquid Trading

Trade perpetual futures (with leverage) and spot markets on Hyperliquid DEX.

Market Types:

  • Perp: Perpetual futures trading with leverage (use market: "perp" in order parameters)
  • Spot: Spot trading (use market: "spot" in order parameters)

Asset Naming for Spot:

  • Non-Hypercore-native assets use "Unit" prefix: UBTC (Unit BTC), UETH (Unit ETH)
  • Example: To trade BTC spot, use symbol UBTC-USDC

Account Information

def run(agent: AgentContext) -> None:
    # Check balances
    balances = agent.platforms.hyperliquid.balances()
    if balances.success and balances.data:
        agent.log(f"Account value: ${balances.data.perp.accountValue}")
        agent.log(f"Withdrawable: ${balances.data.perp.withdrawable}")

    # View open positions
    positions = agent.platforms.hyperliquid.positions()
    if positions.success and positions.data:
        for pos in positions.data:
            agent.log(f"{pos.symbol}: {pos.side} {pos.size} @ {pos.entryPrice}")
            agent.log(f"Unrealized PnL: ${pos.unrealizedPnl}")

Place Orders

def run(agent: AgentContext) -> None:
    # Perp market order
    perp_order = agent.platforms.hyperliquid.place_order({
        "symbol": "BTC-USD",
        "side": "buy",
        "size": 0.0001,
        "price": 110000,  # Acts as slippage limit for market orders
        "market": "perp",
        "type": "market"
    })

    if perp_order.success and perp_order.data:
        agent.log(f"Perp Order {perp_order.data.orderId}: {perp_order.data.status}")
    else:
        agent.log(perp_order.error, error=True)

    # Spot market order (for non-Hypercore-native assets like BTC)
    spot_order = agent.platforms.hyperliquid.place_order({
        "symbol": "UBTC-USDC",  # Unit BTC
        "side": "buy",
        "size": 0.0001,
        "price": 110000,
        "market": "spot",  # Changed to spot
        "type": "market"
    })

    if spot_order.success and spot_order.data:
        agent.log(f"Spot Order {spot_order.data.orderId}: {spot_order.data.status}")
    else:
        agent.log(spot_order.error, error=True)

Order Management

def run(agent: AgentContext) -> None:
    # Get open orders
    open_orders = agent.platforms.hyperliquid.open_orders()

    # Get specific order details
    order = agent.platforms.hyperliquid.order("12345")

    # Cancel an order
    result = agent.platforms.hyperliquid.delete_order("12345", "BTC-USD")

    # View historical orders
    history = agent.platforms.hyperliquid.orders()

    # Check order fills
    fills = agent.platforms.hyperliquid.order_fills()

Transfer Between Accounts

def run(agent: AgentContext) -> None:
    # Transfer USDC from spot to perp account
    transfer = agent.platforms.hyperliquid.transfer({
        "amount": 1000.0,
        "toPerp": True
    })

    if transfer.success:
        agent.log("Transfer completed")

📊 Transaction History

Get a list of asset changes for all confirmed transactions during your session. This is useful for tracking what assets have moved in and out of the agent's wallet.

Note: The system needs to index new transactions, so there may be a slight delay between when you execute a transaction and when the resulting asset changes are returned in this method. Make sure you are taking that into consideration if dealing with assets the agent just transacted with.

def run(agent: AgentContext) -> None:
    # Get all transaction history for this session
    result = agent.transactions()

    if result.success and result.data:
        agent.log(f"Found {len(result.data)} asset changes")

        # Filter for outgoing transfers
        outgoing = [c for c in result.data if c.from_ == agent.sessionWalletAddress]
        agent.log(f"Outgoing transfers: {len(outgoing)}")

        # Calculate total USD value (where price data is available)
        total_usd = sum(
            float(c.amount) * float(c.tokenUsdPrice)
            for c in result.data
            if c.tokenUsdPrice
        )
        agent.log(f"Total USD value: ${total_usd:.2f}")

        # View specific transaction details
        for change in result.data:
            agent.log(f"{change.network}: {change.from_}{change.to}")
            agent.log(f"  Amount: {change.amount} {change.tokenType}")
            if change.token:
                agent.log(f"  Token: {change.token}")
            agent.log(f"  Tx: {change.transactionHash}")
            agent.log(f"  Time: {change.timestamp}")
    else:
        agent.log(result.error or "Failed to fetch transactions", error=True)

AssetChange Structure:

Each asset change in the response contains:

  • network - Network identifier (e.g., "ethereum:1", "solana")
  • transactionHash - Transaction hash
  • from_ - Sender address (note the underscore to avoid Python keyword)
  • to - Recipient address
  • amount - Amount transferred (as string to preserve precision)
  • token - Token contract address (None for native tokens)
  • tokenId - Token ID for NFTs (None for fungible tokens)
  • tokenType - Token type (e.g., "native", "ERC20", "ERC721")
  • tokenUsdPrice - Token price in USD at time of transaction (None if unavailable)
  • timestamp - Transaction timestamp

🚀 Sign & Send Transactions

Ethereum (any EVM chain)

def run(agent: AgentContext) -> None:
    # Self-send demo - send a small amount to yourself
    response = agent.sign_and_send({
        "network": "ethereum:1",
        "request": {
            "to_address": agent.sessionWalletAddress,  # Send to self
            "data": "0x",
            "value": "100000000000000"  # 0.0001 ETH
        },
        "message": "Self-send demo"
    })

    if response.success:
        agent.log(f"Transaction sent: {response.tx_hash}")
        if response.transaction_url:
            agent.log(f"View: {response.transaction_url}")
    else:
        agent.log(response.error_message, error=True)

Solana

def run(agent: AgentContext) -> None:
    response = agent.sign_and_send({
        "network": "solana",
        "request": {
            "hex_transaction": "010001030a0b..."  # serialized VersionedTransaction
        }
    })

    if response.success:
        agent.log(f"Transaction: {response.tx_hash}")
    else:
        agent.log(response.error_message, error=True)

⚡ Error Handling

All SDK methods return response objects with .success and .error_message:

Uncaught Exceptions:

If your function throws an uncaught exception, the Agent SDK automatically:

  1. Logs the error to console (visible in local dev and cloud logs)
  2. Updates the job in the circuit backend with status='failed' and logs the error.
def run(agent: AgentContext) -> None:
    # This typo will be caught and logged automatically
    agent.memmory.set("key", "value")  # AttributeError

    # No need for try/except around everything!

🛠️ Deployment

What You Write

Your agent code should look like this - just define your functions and add the boilerplate at the bottom:

from agent_sdk import Agent, AgentContext

def run(agent: AgentContext) -> None:
    agent.log("Hello from my agent!")
    # Your agent logic here

def stop(agent: AgentContext) -> None:
    agent.log("Cleaning up...")
    # Optional cleanup logic

# ============================================================================
# BOILERPLATE - Don't modify, this handles local testing AND Circuit deployment
# ============================================================================
agent = Agent(
    run=run,
    stop=stop
)

handler = agent.get_handler()

if __name__ == "__main__":
    agent.run()

That's it! The boilerplate code automatically handles the rest

Deploy to Circuit

circuit publish

See the circuit cli docs for more details

Test Locally

🧪 Manual Instantiation (Jupyter/Testing)

For testing in Jupyter notebooks or scripts, you can manually create an AgentContext:

from agent_sdk import AgentContext
from agent_sdk.agent_context import CurrentPosition

# Create agent context with test data
# Tip: Get a real session ID and wallet from running 'circuit run -m local -x execute'
SESSION_ID = <your-session-id>
WALLET_ADDRESS = "your-wallet-address"

# Create sample position data - helpful for testing your agents treatment of different scenarios
# for example if you want to test some re-balancing logic, you can build a sample set of positions here
# In production, or when using the cli, these will be live values for the session
SAMPLE_POSITION = CurrentPosition(
    network="ethereum:137",
    assetAddress="0x4d97dcd97ec945f40cf65f87097ace5ea0476045",
    tokenId="86192057611122246511563653509192966169513312957180910360241289053249649036697",
    avgUnitCost="0.779812797920499100",
    currentQty="41282044"
)

agent = AgentContext(
    sessionId=SESSION_ID,
    sessionWalletAddress=WALLET_ADDRESS,
    currentPositions=[SAMPLE_POSITION]
)

# Use it just like in production!
agent.log("Testing in Jupyter!")
agent.memory.set("test_key", "test_value")

result = agent.memory.get("test_key")
if result.success and result.data:
    print(f"Value: {result.data.value}")

Key Points:

  • Use real session IDs from the CLI for proper testing
  • currentPositions is optional but helps test position-related logic
  • All SDK methods work the same way as in production

📚 Working Examples

The SDK includes complete working examples to help you get started:

demo-agent.py

An agent demonstrating all main features of the sdk:

  • Memory operations (set, get, list, delete)
  • Polymarket integration (market orders, position redemption)
  • Swidge
  • Self sending using sign and send
  • Unified logging patterns

This is a great starting point for building your own agent.

examples/kitchen-sink.ipynb

An interactive Jupyter notebook showing all SDK features:

  • Manual AgentContext instantiation for testing
  • All memory operations with examples
  • Cross-chain swaps with Swidge (quotes, execution, price impact checks)
  • Polymarket operations (buy/sell orders, redemptions)
  • Custom transactions with sign_and_send
  • Logging patterns (standard, error, debug)

Run this notebook to experiment with the SDK interactively before deploying your agent.

🎯 Key Takeaways

  1. Single Interface: Everything you need is in the AgentContext object
  2. No Return Values: Functions return None - uncaught errors are handled automatically by circuit
  3. Unified Logging: agent.log() with simple debug and error flags
  4. Check .success: All SDK methods return response objects with .success and .error_message
  5. Let Errors Bubble: Uncaught exceptions are automatically logged and reported
  6. Copy the Boilerplate: Use the same boilerplate code for local testing and Circuit deployment

📖 Additional Resources

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

circuit_agent_sdk-1.3.2.tar.gz (322.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

circuit_agent_sdk-1.3.2-py3-none-any.whl (72.8 kB view details)

Uploaded Python 3

File details

Details for the file circuit_agent_sdk-1.3.2.tar.gz.

File metadata

  • Download URL: circuit_agent_sdk-1.3.2.tar.gz
  • Upload date:
  • Size: 322.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for circuit_agent_sdk-1.3.2.tar.gz
Algorithm Hash digest
SHA256 12236010436dec36cf418a4d57a938895458bd5ca87a4a8043caaf6d50b7fd76
MD5 c2a48aa422c167551a38a1ff93f2c402
BLAKE2b-256 0a22b6d8d67a0cf8eb1bfb4ad33a00635f16945a1f481f76a23322d49ad3a1e1

See more details on using hashes here.

File details

Details for the file circuit_agent_sdk-1.3.2-py3-none-any.whl.

File metadata

File hashes

Hashes for circuit_agent_sdk-1.3.2-py3-none-any.whl
Algorithm Hash digest
SHA256 ded8da2616bc51179daff85c8e27940b8361ef1de3a98cedaf90a3ba7876150e
MD5 e8f98ddc4a36f2bf02b526b8bfd2cc43
BLAKE2b-256 c7b7a3bead5ff98862146fa9b6ab53010a75059d2c5be4f0265e2eda88a38490

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page