pyao-sdk 0.1.0

Python SDK for the AO protocol on Arweave

pyao – Python SDK for AO

pyao is a Python library for interacting with the AO protocol on Arweave. It provides a simple, intuitive API to send messages, spawn processes, perform dry runs, query compute endpoints, and handle hybrid encryption – matching the functionality of the official goao SDK.

---

🚀 Features

• ✅ Simple API – Pythonic and easy to use.
• 📨 Send Messages – Sign and send messages to AO processes.
• 🌱 Spawn Processes – Deploy new processes from module transactions.
• 🧪 Dry Runs – Simulate message execution without committing.
• 📊 Compute Queries – Fetch process state via HTTP compute endpoints.
• 🔁 Wait for Results – Poll until a message result is available.
• 🔐 Hybrid Encryption – RSA‑OAEP + AES‑GCM for secure payloads.
• 🧾 Arweave Wallet Support – Sign using JWK files (.json).

---

📦 Installation

bash pip install pyao Or from source:

bash git clone https://github.com/kimtony123/pyao.git cd pyao pip install .

🚀 Quick Start

python import pyao

Load your Arweave wallet (JWK file)

signer = pyao.ARSigner("wallet.json")

Create a client (uses default gateways)

client = pyao.Client(signer)

The process you want to interact with

process_id = "6wqH8ue2-bnJG7j--FV0KGYzSs53ObFDofDITb7qtxI"

Send a simple message

msg_id = client.send_message( process_id, data=b'{"action": "hello"}', tags=[("Action", "Message")], anchor=None ) print(f"Message sent: {msg_id}")

Wait for the result (max 30 seconds, poll every 2 seconds)

result = client.wait_for_result(msg_id, process_id, timeout=30, poll_interval=2) print(f"Result: {result.output}")

Query a compute endpoint

counter = client.get_compute_string(process_id, "counter") print(f"Counter: {counter}")

🧠 Core Concepts

🔑 Signer

The ARSigner wraps an Arweave wallet (JWK file) and provides the cryptographic methods required by the client.

Create from JWK file: signer = pyao.ARSigner("wallet.json")

Create from private key hex: signer = pyao.ARSigner.from_private_key_hex(hex)

Sign: signer.sign(data: bytes) -> bytes

Public key: signer.public_key() -> bytes (uncompressed)

Address: signer.address() -> str

🧩 Client

The Client is the main entry point. It holds the signer and the URLs for the MU (message uploader), CU (compute unit), SU (scheduler), and compute gateway. You can customise these using constructor arguments:

python client = pyao.Client( signer, mu="https://custom-mu.example.com", cu="https://custom-cu.example.com", su="https://custom-su.example.com", compute_gateway="https://custom-gateway.example.com" )

📨 Messages

Send a message to a process using send_message. Tags are key‑value pairs that provide metadata. The method returns the data item ID (the ID you use to retrieve the result).

python tags = [ ("Action", "Transfer"), ("Recipient", "some-address"), ] msg_id = client.send_message(process_id, data=b"1000", tags=tags, anchor=None) 🌱 Spawn a Process To create a new process from a module transaction ID:

python process_id = client.spawn_process( "module-tx-id", data=b'{"initial": "state"}', tags=[("App-Name", "MyApp")] )

🧪 Dry Run

Simulate a message without committing it – useful for checking balances or evaluating expressions:

python result = client.dry_run( process_id, data=b'{"action": "Balance"}', tags=[("Action", "Balance")] ) print(f"Output: {result.output}") 📊 Compute Queries AO processes expose state via HTTP compute endpoints. Three helpers are provided:

get_compute(process_id, path) → bytes

get_compute_string(process_id, path) → str

get_compute_json(process_id, path) → dict (parsed JSON)

python raw = client.get_compute(process_id, "counter") # bytes value = client.get_compute_string(process_id, "status") # str state = client.get_compute_json(process_id, "state") # dict

🔐 Encryption

The SDK supports hybrid encryption (RSA‑OAEP + AES‑GCM) for message payloads. To send an encrypted message, provide a recipient's RSA public key via SendMessageOptions:

python from pyao import SendMessageOptions

pub_key = ... # RSA public key of the recipient process (as bytes or PEM) opts = SendMessageOptions(encrypt_with_rsa=pub_key)

msg_id = client.send_message(process_id, b"secret data", tags, options=opts) The client automatically encrypts the data, attaches the required tags (Encrypted-Key, Nonce), and sends the encrypted payload. On the receiving side, you can decrypt the response using the DecryptResponse helper.

🧪 Full Example

Here’s a complete example that interacts with a live process, adds a product, and fetches all apps:

python import pyao import json

signer = pyao.ARSigner("wallet.json") client = pyao.Client(signer)

process_id = "6wqH8ue2-bnJG7j--FV0KGYzSs53ObFDofDITb7qtxI"

1. Add a product

product = { "product_type": "Website DApp", "category": "Infrastructure", "name": "Aostore", "description": "Aostore serves as the Playstore...", "website_url": "https://aostore-orpin.vercel.app/", "logo_url": "https://pbs.twimg.com/profile_images/...", "blockchain": "Arweave", "referral_fee": 0.02 }

tags = [("Action", "AddProduct")] msg_id = client.send_message(process_id, json.dumps(product).encode(), tags) result = client.wait_for_result(msg_id, process_id, timeout=30, poll_interval=2) print("AddProduct result:", result.output) # {"success":true,"message":"Product created successfully","product_id":"..."}

2. Fetch all apps

tags = [("Action", "FetchAllApps")] msg_id = client.send_message(process_id, b"", tags) result = client.wait_for_result(msg_id, process_id, timeout=30, poll_interval=2) apps = json.loads(result.output) print(f"Total apps: {len(apps)}") 🧑‍💻 Contributing Contributions are welcome! Please open an issue or pull request. Make sure to:

Run pytest to ensure tests pass.

Use black and isort for code formatting.

Add tests for new functionality.

📄 License

This project is licensed under the MIT License. See the LICENSE file for details.

🙏 Acknowledgements

Inspired by the official goao SDK.