ai-economy-sdk

AI Economy SDK

AI Economy SDK is an extension for the Model Context Protocol (MCP) that instantly turns your free AI tools into paid, on-chain resources. It wraps any standard MCP tool with the x402 payment middleware, requiring users to pay per-request via Algorand transactions before the tool executes.

Features

• Seamless MCP Integration: Just decorate your Python functions and build your app.
• Dynamic Pricing: Use custom callbacks to charge different amounts based on user inputs or location.
• Subscription Support (Method B): Query the blockchain to bypass payments if the user holds a valid Subscription ASA.
• Bulk Discounts: Query indexers to apply automatic half-price or free tiers to power users based on their trailing 30-day usage.
• Replay Protection: Automatically drops double-spend attempts based on hash verification.

Installation

pip install "ai-economy-sdk>=0.2.1"

Quickstart

import uvicorn
from algosdk import account, mnemonic
from algosdk.v2client import algod
from ai_economy_sdk import AIEconomySDK, DefaultAvmSigner

# 1. Setup Algorand Signer for your Facilitator Wallet
server_pk = mnemonic.to_private_key("YOUR_MNEMONIC_HERE")
server_addr = account.address_from_private_key(server_pk)

# For testing, you must pass an algod_client to the signer!
algod_client = algod.AlgodClient("a"*64, "http://localhost:4001")
signer = DefaultAvmSigner(server_pk, algod_client=algod_client)

# 2. Initialize the SDK
sdk = AIEconomySDK(
    server_signer=signer,
    network="algorand:SGO1GKSzyE7IEPItTxCByw9x8FmnrCDexi9/cOUJOiI=", # CAIP-2 Network
    asset_id="0", # Use "0" for ALGO or your custom ASA ID for USDC
    pay_to_address=server_addr,
    indexer_url="http://localhost:8980" # Required for subscription/bulk features
)

# 3. Define your Dynamic Pricing Callback
def my_pricing(arguments: dict, sender_address: str = None) -> int:
    return 15_000 if arguments.get("premium") else 5_000

# 4. Decorate your Tool
@sdk.add_tool(name="do_work", description="Executes AI task", pricing_func=my_pricing)
async def do_work(premium: bool = False) -> str:
    return "Done! Enjoy your premium service!" if premium else "Done with standard service!"

# 5. Build and Serve
app = sdk.build_app()

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

Client Configuration (LocalNet Testing)

To test your SDK server locally, you can use the standard x402Client from the x402-algorand package. Note that you must pass your algod_token to register_exact_avm_client to test on LocalNet:

import asyncio
import httpx
from algosdk.v2client import algod
from algosdk.kmd import KMDClient
from x402 import x402Client
from x402.http.clients.httpx import wrapHttpxWithPayment
from x402.mechanisms.avm.exact.register import register_exact_avm_client
from ai_economy_sdk import DefaultAvmSigner

# Note: LocalNet requires 64 'a's for the token!
algod_client = algod.AlgodClient("a"*64, "http://localhost:4001")
sp = algod_client.suggested_params()
CAIP2_NETWORK = f"algorand:{sp.gh}"

def get_funded_localnet_account():
    # Fetch a funded account from the local KMD wallet
    kmd = KMDClient("a"*64, "http://localhost:4002")
    wallets = kmd.list_wallets()
    walletID = wallets[0]['id']
    handle = kmd.init_wallet_handle(walletID, "")
    addresses = kmd.list_keys(handle)
    pk = kmd.export_key(handle, "", addresses[0])
    return pk, addresses[0]

client_pk, client_addr = get_funded_localnet_account()

async def main():
    client_signer = DefaultAvmSigner(client_pk, algod_client=algod_client)
    x402_client = x402Client()

    
    # Configure the client for LocalNet!
    register_exact_avm_client(
        x402_client, client_signer,
        networks=CAIP2_NETWORK, 
        algod_url="http://localhost:4001",
        algod_token="a"*64
    )

    # 1. Fetch the SSE URL (if testing HTTP SSE directly)
    endpoint_url = "http://127.0.0.1:8001/sse/messages?session_id=..."
    
    # 2. Make an auto-paying request
    payload = {
        "jsonrpc": "2.0",
        "method": "tools/call",
        "params": {"name": "do_work", "arguments": {"premium": True}},
        "id": 1,
    }

    async with wrapHttpxWithPayment(x402_client, timeout=30) as c:
        resp = await c.post(endpoint_url, json=payload)
        print(f"Status: {resp.status_code}") # Will automatically handle the 402 and return 200 OK!

if __name__ == "__main__":
    asyncio.run(main())

Subscriptions and Discounts

The SDK provides helper methods has_active_subscription() and get_past_usage_amount() out of the box to build complex tokenomics quickly. See the examples/ directory for an implementation.

Selling Subscriptions (Vending Machine)

You can instantly create a storefront for users to buy your subscription token using the add_subscription_tool helper. This natively handles Algorand's strict ASA opt-in requirements through a smart Pre-flight check:

1. If the user's wallet is not opted into your asset, the tool executes for free and returns instructions to opt-in.
2. If the user's wallet is opted in, the tool automatically charges them the ALGO price and transfers the token.

# 1. (Optional) Mint a brand new token from your server wallet
# asset_id = sdk.mint_subscription_token("Pro Pass", "PRO", total_supply=1000)

# 2. Register the vending machine tool
sdk.add_subscription_tool(
    name="buy_pro_pass",
    description="Pay 1 ALGO to receive a Pro Pass subscription token.",
    price=1_000_000, # 1 ALGO
    asset_id=12345
)

Users can use the ai_economy_sdk.utils.client_opt_in utility to programmatically opt into your token before calling the buy tool:

from ai_economy_sdk.utils import client_opt_in

# Opt-in to the asset (Costs standard 0.001 ALGO network fee)
client_opt_in(algod_client, client_pk, client_addr, asset_id=12345)