ai-economy-sdk 0.1.6

Payment-gated MCP server SDK for Algorand

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.1.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.