Python Client for Sui gRPC

# Create project directory
mkdir sui-grpc-python
cd sui-grpc-python

# Create virtual environment
python3 -m venv venv

# Activate virtual environment
source venv/bin/activate  # On macOS/Linux
# or
.\venv\Scripts\activate   # On Windows

pip install grpcio grpcio-tools protobuf python-dotenv

# Clone the Sui APIs repository
git clone https://github.com/MystenLabs/sui-apis.git

# Copy proto files to your project
cp -r sui-apis/proto .

# Generate Python code from proto files
python -m grpc_tools.protoc \
  -I./proto \
  --python_out=. \
  --grpc_python_out=. \
  ./proto/sui/rpc/v2/*.proto \
  ./proto/google/rpc/*.proto

# Create __init__.py files for sui package only
find sui -type d -exec touch {}/__init__.py \;

sui-grpc-python/
├── .env.example          # Environment variables template
├── .env                  # Your config (gitignored)
├── client.py             # gRPC client wrapper
├── example_get_checkpoint.py
├── example_get_balance.py
├── proto/                # Protocol buffer definitions
│   ├── sui/rpc/v2/
│   └── google/rpc/
├── sui/                  # Generated Python code
│   └── rpc/v2/
└── google/               # Generated Google proto code
    └── rpc/

# Copy the example
cp .env.example .env

# Edit and add your API key
# .env
DWELLIR_ENDPOINT=api-sui-mainnet-full.n.dwellir.com:443
DWELLIR_API_KEY=your_api_key_here

"""Simple Sui gRPC Client"""
import grpc
import os
from dotenv import load_dotenv

load_dotenv()

class SuiGrpcClient:
    """Basic Sui gRPC client"""

    def __init__(self, endpoint=None, api_key=None):
        self.endpoint = endpoint or os.getenv('DWELLIR_ENDPOINT')
        self.api_key = api_key or os.getenv('DWELLIR_API_KEY')

        if not self.endpoint or not self.api_key:
            raise ValueError(
                "Missing DWELLIR_ENDPOINT or DWELLIR_API_KEY. "
                "Set them in .env file or pass as arguments."
            )

        self._channel = None

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def connect(self):
        """Establish gRPC connection"""
        credentials = grpc.ssl_channel_credentials()
        self._channel = grpc.secure_channel(
            self.endpoint,
            credentials,
            options=[
                ('grpc.max_receive_message_length', 50 * 1024 * 1024),  # 50MB
                ('grpc.max_send_message_length', 50 * 1024 * 1024),
            ]
        )

    def close(self):
        """Close gRPC connection"""
        if self._channel:
            self._channel.close()
            self._channel = None

    def get_metadata(self):
        """Generate authentication metadata"""
        return [('x-api-key', self.api_key)]

    @property
    def channel(self):
        """Get or create channel"""
        if self._channel is None:
            self.connect()
        return self._channel

#!/usr/bin/env python3
"""Example: Fetch the latest checkpoint from Sui"""
import grpc
from client import SuiGrpcClient
from sui.rpc.v2 import ledger_service_pb2
from sui.rpc.v2 import ledger_service_pb2_grpc

def get_current_checkpoint():
    """Fetch the current checkpoint"""
    print("Connecting to Sui gRPC API...")

    with SuiGrpcClient() as client:
        # Create ledger service stub
        stub = ledger_service_pb2_grpc.LedgerServiceStub(client.channel)

        # Create request - get latest checkpoint
        request = ledger_service_pb2.GetCheckpointRequest()

        # Execute request with authentication
        try:
            print("Fetching latest checkpoint...")
            response = stub.GetCheckpoint(
                request,
                metadata=client.get_metadata(),
                timeout=30
            )

            checkpoint = response.checkpoint
            print("\n" + "="*50)
            print("Successfully fetched checkpoint!")
            print("="*50)
            print(f"Sequence Number: {checkpoint.summary.sequence_number}")
            print(f"Epoch: {checkpoint.summary.epoch}")
            print(f"Timestamp: {checkpoint.summary.timestamp}")

            if checkpoint.contents and checkpoint.contents.transactions:
                print(f"Transactions in checkpoint: {len(checkpoint.contents.transactions)}")
            else:
                print("No transaction details in response")

            print("="*50)
            return checkpoint

        except grpc.RpcError as e:
            print(f"\nError: {e.code()} - {e.details()}")
            if e.code() == grpc.StatusCode.UNAUTHENTICATED:
                print("\nAuthentication failed. Please check:")
                print("1. You have created a .env file (copy from .env.example)")
                print("2. Your DWELLIR_API_KEY is correct")
                print("3. Your API key is active")
            raise

if __name__ == '__main__':
    get_current_checkpoint()

# Activate virtual environment (if not already active)
source venv/bin/activate  # On macOS/Linux
# or
.\venv\Scripts\activate   # On Windows

# Run the example
python example_get_checkpoint.py

Connecting to Sui gRPC API...
Fetching latest checkpoint...

==================================================
Successfully fetched checkpoint!
==================================================
Sequence Number: 0
Epoch: 0
Timestamp:
No transaction details in response
==================================================

import time
from decimal import Decimal
from typing import Optional
from client import SuiGrpcClient
from sui.rpc.v2 import state_service_pb2, state_service_pb2_grpc

class BalanceMonitor:
    """Monitor balance changes for an address"""

    def __init__(self, address: str, coin_type: str):
        self.address = address
        self.coin_type = coin_type
        self.previous_balance: Optional[int] = None

    def check_balance(self) -> dict:
        """Check current balance and detect changes"""
        with SuiGrpcClient() as client:
            stub = state_service_pb2_grpc.StateServiceStub(client.channel)

            request = state_service_pb2.GetBalanceRequest(
                owner=self.address,
                coin_type=self.coin_type
            )

            response = stub.GetBalance(
                request,
                metadata=client.get_metadata()
            )

            # response.balance is a Balance object, access its balance field
            if response.balance and response.balance.balance is not None:
                current_balance = response.balance.balance
            else:
                current_balance = 0

            result = {
                'balance': current_balance,
                'balance_sui': Decimal(current_balance) / Decimal(1_000_000_000),
                'changed': False,
                'change_amount': 0
            }

            if self.previous_balance is not None:
                change = current_balance - self.previous_balance
                if change != 0:
                    result['changed'] = True
                    result['change_amount'] = change

            self.previous_balance = current_balance
            return result

    def monitor(self, interval: int = 10):
        """Continuously monitor balance changes"""
        print(f"Monitoring balance for: {self.address}")
        print(f"Coin type: {self.coin_type}")
        print("Press Ctrl+C to stop\n")

        try:
            while True:
                result = self.check_balance()

                if result['changed']:
                    change = result['change_amount']
                    direction = '+' if change > 0 else ''
                    print(f"[{time.strftime('%H:%M:%S')}] Balance changed: "
                          f"{direction}{change} (Total: {result['balance_sui']} SUI)")
                else:
                    print(f"[{time.strftime('%H:%M:%S')}] Balance: "
                          f"{result['balance_sui']} SUI")

                time.sleep(interval)

        except KeyboardInterrupt:
            print("\nMonitoring stopped")

# Usage
if __name__ == '__main__':
    monitor = BalanceMonitor(
        address='0x935029ca5219502a47ac9b69f556ccf6e2198b5e7815cf50f68846f723739cbd',
        coin_type='0x0000000000000000000000000000000000000000000000000000000000000002::sui::SUI'
    )
    monitor.monitor(interval=5)

from typing import List, Dict
from dataclasses import dataclass
import grpc
from client import SuiGrpcClient
from sui.rpc.v2 import state_service_pb2, state_service_pb2_grpc

@dataclass
class TokenBalance:
    coin_type: str
    balance: int
    symbol: str
    decimals: int

    @property
    def formatted_balance(self) -> str:
        """Human-readable balance"""
        value = self.balance / (10 ** self.decimals)
        return f"{value:.{self.decimals}f} {self.symbol}"

class PortfolioTracker:
    """Track all token balances for an address"""

    def __init__(self, address: str):
        self.address = address

    def get_portfolio(self) -> List[TokenBalance]:
        """Fetch all token balances"""
        with SuiGrpcClient() as client:
            state_stub = state_service_pb2_grpc.StateServiceStub(client.channel)

            # Get all balances
            balance_request = state_service_pb2.ListBalancesRequest(
                owner=self.address
            )

            balance_response = state_stub.ListBalances(
                balance_request,
                metadata=client.get_metadata()
            )

            portfolio = []

            # Enrich with coin metadata
            for balance_item in balance_response.balances:
                # Get coin info
                info_request = state_service_pb2.GetCoinInfoRequest(
                    coin_type=balance_item.coin_type
                )

                try:
                    info_response = state_stub.GetCoinInfo(
                        info_request,
                        metadata=client.get_metadata()
                    )

                    metadata = info_response.metadata
                    portfolio.append(TokenBalance(
                        coin_type=balance_item.coin_type,
                        balance=int(balance_item.balance),
                        symbol=metadata.symbol,
                        decimals=metadata.decimals
                    ))

                except grpc.RpcError as e:
                    print(f"Warning: Could not get info for {balance_item.coin_type}")

            return portfolio

    def print_portfolio(self):
        """Display portfolio in formatted table"""
        portfolio = self.get_portfolio()

        print(f"\nPortfolio for: {self.address}\n")
        print(f"{'Token':<15} {'Balance':<30}")
        print("-" * 45)

        for token in portfolio:
            print(f"{token.symbol:<15} {token.formatted_balance:<30}")

        print(f"\nTotal tokens: {len(portfolio)}")

# Usage
if __name__ == '__main__':
    tracker = PortfolioTracker('0x935029ca5219502a47ac9b69f556ccf6e2198b5e7815cf50f68846f723739cbd')
    tracker.print_portfolio()