cryptography and blake3 packages.
This page shows the canonical pattern. A published sigil-py package will track this surface.
Install
pip install httpx cryptography blake3
# or, for async-only workloads:
pip install aiohttp cryptography blake3
Configuration
# sigil/config.py
from dataclasses import dataclass
@dataclass(frozen=True)
class Network:
chain_id: str
rpc_url: str
timeout_seconds: float = 30.0
MAINNET = Network(
chain_id="sigil-mainnet-1",
rpc_url="https://rpc.sigil.ml/rpc",
)
TESTNET = Network(
chain_id="sigil-testnet-1",
rpc_url="https://testnet-rpc.sigil.ml/rpc",
)
export SIGIL_RPC_URL=https://testnet-rpc.sigil.ml/rpc
export SIGIL_CHAIN_ID=sigil-testnet-1
export SIGIL_SIGNING_KEY_HEX=... # 64 hex chars (Ed25519 seed)
export SIGIL_SENDER_DID=did:oas:sigil:agent:...
Identity
# sigil/identity.py
from dataclasses import dataclass
import blake3
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey, Ed25519PublicKey,
)
from cryptography.hazmat.primitives import serialization
@dataclass
class Identity:
did: str
private_key: Ed25519PrivateKey
public_key_bytes: bytes
def _public_key_bytes(key: Ed25519PrivateKey) -> bytes:
return key.public_key().public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
def new_agent() -> Identity:
priv = Ed25519PrivateKey.generate()
pub = _public_key_bytes(priv)
addr = blake3.blake3(pub).digest()
return Identity(
did=f"did:oas:sigil:agent:{addr.hex()}",
private_key=priv,
public_key_bytes=pub,
)
def load_agent(seed_hex: str) -> Identity:
seed = bytes.fromhex(seed_hex)
if len(seed) != 32:
raise ValueError(f"seed must be 32 bytes, got {len(seed)}")
priv = Ed25519PrivateKey.from_private_bytes(seed)
pub = _public_key_bytes(priv)
addr = blake3.blake3(pub).digest()
return Identity(
did=f"did:oas:sigil:agent:{addr.hex()}",
private_key=priv,
public_key_bytes=pub,
)
Reading chain state
# sigil/rpc.py
from typing import Any, TypeVar
import httpx
class RpcError(Exception):
def __init__(self, code: int, message: str):
super().__init__(f"rpc {code}: {message}")
self.code = code
self.message = message
T = TypeVar("T")
async def call(
client: httpx.AsyncClient,
network: "Network",
method: str,
params: Any | None = None,
) -> Any:
if params is None:
params = {}
res = await client.post(
network.rpc_url,
json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params},
timeout=network.timeout_seconds,
)
res.raise_for_status()
body = res.json()
if "error" in body and body["error"]:
raise RpcError(body["error"]["code"], body["error"]["message"])
if "result" not in body:
raise RpcError(-32003, "missing result")
return body["result"]
Typed wrappers
async def get_node_info(client: httpx.AsyncClient, network: Network) -> dict:
return await call(client, network, "sigil_getNodeInfo", {})
async def get_balance(client: httpx.AsyncClient, network: Network, did: str) -> dict:
return await call(client, network, "sigil_getBalance", {"did": did})
async def get_latest_block(client: httpx.AsyncClient, network: Network) -> dict:
return await call(client, network, "sigil_getLatestBlock", {})
httpx.AsyncClient for httpx.Client, drop async/await, and the rest of the file is identical.
Signing and submitting transactions
# sigil/tx.py
from datetime import datetime, timezone
from typing import Any
import json
import blake3
from .identity import Identity
TX_DOMAIN_TAG = "sigil/transaction/v1\n"
def transfer_tx(
chain_id: str,
sender_did: str,
recipient_did: str,
amount_base_units: int,
nonce: int,
memo: str | None = None,
) -> dict[str, Any]:
return {
"chain_id": chain_id,
"sender_did": sender_did,
"nonce": nonce,
"gas_limit": 200_000,
"gas_price": 1,
"tx_type": {
"Transfer": {
"recipient_did": recipient_did,
"amount": amount_base_units,
"memo": memo,
}
},
# Match Rust's chrono::Utc::now() serde shape: RFC3339 with nanoseconds.
"timestamp": datetime.now(timezone.utc).isoformat(),
}
def sign_tx(tx: dict[str, Any], identity: Identity) -> dict[str, Any]:
body = json.dumps(tx, separators=(",", ":"), sort_keys=False).encode("utf-8")
transcript = TX_DOMAIN_TAG.encode("utf-8") + body
digest = blake3.blake3(transcript).digest()
signature = identity.private_key.sign(digest)
return {
"tx": tx,
"signature": signature.hex(),
"signer_public_key": identity.public_key_bytes.hex(),
}
The Rust node accepts the JSON
tx_type enum in two shapes: externally tagged ({"Transfer": {...}}) and adjacently tagged. Match the shape used by sigil-core exactly. If you see expected enum TransactionType on submit, your enum representation drifted.Submitting and waiting
import asyncio
async def send_transaction(
client: httpx.AsyncClient,
network: Network,
signed: dict,
) -> str:
result = await call(client, network, "sigil_sendTransaction", signed)
return result["tx_hash"]
async def wait_receipt(
client: httpx.AsyncClient,
network: Network,
tx_hash: str,
timeout_seconds: float = 30.0,
) -> dict:
deadline = asyncio.get_event_loop().time() + timeout_seconds
while asyncio.get_event_loop().time() < deadline:
try:
receipt = await call(
client, network,
"sigil_getTransactionReceipt",
{"tx_hash": tx_hash},
)
if receipt and receipt.get("status") != "pending":
return receipt
except RpcError as e:
if e.code != -32601:
raise
await asyncio.sleep(1.5)
raise TimeoutError(f"receipt for {tx_hash} did not finalise in {timeout_seconds}s")
Subscribing to events
WebSocket subscriptions are on the roadmap. Until they ship, pollsigil_getLatestBlock:
async def stream_blocks(client: httpx.AsyncClient, network: Network):
last = 0
while True:
block = await get_latest_block(client, network)
height = int(block.get("height", 0))
if height > last:
last = height
yield block
await asyncio.sleep(1.5)
Error handling
from .rpc import RpcError
try:
tx_hash = await send_transaction(client, network, signed)
except RpcError as e:
if e.code == -32002: # mempool full / rate-limited
await asyncio.sleep(0.5)
elif e.code == -32601: # method not found
raise RuntimeError("RPC node missing required method") from e
elif e.code == -32602: # invalid params
raise RuntimeError(f"client bug: {e.message}") from e
else:
raise
| Code | Meaning |
|---|---|
-32600 | Invalid JSON-RPC envelope |
-32601 | Method not found |
-32602 | Invalid params |
-32603 | Internal error (retryable) |
-32002 | Mempool full / rate-limited |
Worked example: send a payment with retry
# main.py
import asyncio
import os
import httpx
from sigil.config import TESTNET, Network
from sigil.identity import load_agent
from sigil.rpc import RpcError, call
from sigil.tx import sign_tx, transfer_tx
async def send_payment_with_retry(
client: httpx.AsyncClient,
network: Network,
identity,
recipient_did: str,
amount_base_units: int,
) -> str:
bal = await call(client, network, "sigil_getBalance", {"did": identity.did})
tx = transfer_tx(
chain_id=network.chain_id,
sender_did=identity.did,
recipient_did=recipient_did,
amount_base_units=amount_base_units,
nonce=bal["nonce"],
)
signed = sign_tx(tx, identity)
delay = 0.25
for attempt in range(5):
try:
result = await call(client, network, "sigil_sendTransaction", signed)
return result["tx_hash"]
except RpcError as e:
if e.code in (-32002, -32603) and attempt < 4:
await asyncio.sleep(delay)
delay *= 2
continue
raise
raise RuntimeError("unreachable")
async def main() -> None:
network = Network(
chain_id=os.environ.get("SIGIL_CHAIN_ID", "sigil-testnet-1"),
rpc_url=os.environ.get("SIGIL_RPC_URL", "https://testnet-rpc.sigil.ml/rpc"),
)
identity = load_agent(os.environ["SIGIL_SIGNING_KEY_HEX"])
recipient = os.environ["RECIPIENT_DID"]
amount = int(os.environ["AMOUNT_BASE_UNITS"])
async with httpx.AsyncClient() as client:
tx_hash = await send_payment_with_retry(client, network, identity, recipient, amount)
print("submitted:", tx_hash)
if __name__ == "__main__":
asyncio.run(main())
Reference
- Wire format: JSON-RPC method index
- Canonical signing transcript:
sigil-wallet-core/src/lib.rs - Parity matrix: SDK overview