DEV Community

L_X_1
L_X_1

Posted on • Originally published at policylayer.com

How to Add Spending Limits to AutoGPT Agents

#autogpt #tutorial

AutoGPT was built for autonomy. It loops, plans, and executes without human intervention. That's powerful—until your agent decides to "optimise" your finances by executing 500 transactions overnight.

This guide shows you how to add hard spending limits to AutoGPT agents.

The AutoGPT Risk Profile

AutoGPT's architecture makes it particularly risky for financial operations:

  1. Recursive execution: Agent plans and executes in a loop
  2. Tool chaining: One tool's output feeds another's input
  3. Minimal oversight: Designed to run without human intervention
  4. Persistent memory: Can remember and retry failed transactions

Give an AutoGPT agent wallet access, and it will use it. Repeatedly.

# Typical AutoGPT command registration
@command("send_crypto", "Send cryptocurrency to an address")
def send_crypto(recipient: str, amount: float, currency: str) -> str:
    # Direct wallet access - no limits!
    wallet.send(recipient, amount, currency)
    return f"Sent {amount} {currency} to {recipient}"
Enter fullscreen mode Exit fullscreen mode

Adding Policy Enforcement

We'll wrap the crypto command with PolicyLayer to enforce limits:

import os
import json
import requests
from autogpt.commands import command

class PolicyEnforcedWallet:
    def __init__(self):
        self.api_url = "https://api.policylayer.com"
        self.api_key = os.environ["POLICYLAYER_API_KEY"]
        self.agent_id = os.environ.get("AUTOGPT_AGENT_ID", "autogpt-main")

    def send(self, to: str, amount: str, asset: str) -> dict:
        # Gate 1: Validate against policy
        validation = self._validate_intent(to, amount, asset)

        if validation["decision"] != "allow":
            return {
                "success": False,
                "error": f"Policy blocked: {validation.get('reason', 'Unknown')}"
            }

        # Gate 2: Verify authorisation
        verification = self._verify_auth(
            validation["authorisation"],
            validation["fingerprint"]
        )

        if verification["status"] != "valid":
            return {
                "success": False,
                "error": f"Verification failed: {verification.get('reason')}"
            }

        # Execute the actual transaction
        tx_hash = self._execute_transaction(to, amount, asset)

        return {
            "success": True,
            "tx_hash": tx_hash,
            "amount": amount,
            "asset": asset
        }

    def _validate_intent(self, to: str, amount: str, asset: str) -> dict:
        response = requests.post(
            f"{self.api_url}/validate-intent",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "intent": {
                    "chain": "ethereum",
                    "asset": asset.lower(),
                    "to": to,
                    "amount": amount,
                    "nonce": str(uuid.uuid4()),
                    "expWindowSeconds": 60
                },
                "idempotencyKey": str(uuid.uuid4()),
                "metadata": {
                    "agentId": self.agent_id,
                    "source": "autogpt"
                }
            }
        )
        return response.json()

    def _verify_auth(self, auth: str, fingerprint: str) -> dict:
        response = requests.post(
            f"{self.api_url}/verify-authorisation",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "auth": auth,
                "intentFingerprint": fingerprint
            }
        )
        return response.json()

    def _execute_transaction(self, to: str, amount: str, asset: str) -> str:
        # Your actual transaction execution logic
        # This could be ethers.js, web3.py, etc.
        pass


# Global policy-enforced wallet instance
policy_wallet = PolicyEnforcedWallet()


@command(
    "send_crypto",
    "Send cryptocurrency with spending limits enforced",
    {
        "recipient": {"type": "string", "description": "Wallet address"},
        "amount": {"type": "string", "description": "Amount to send"},
        "currency": {"type": "string", "description": "Currency (eth, usdc)"}
    }
)
def send_crypto(recipient: str, amount: str, currency: str) -> str:
    result = policy_wallet.send(recipient, amount, currency)

    if result["success"]:
        return f"SUCCESS: Sent {amount} {currency}. TX: {result['tx_hash']}"
    else:
        return f"BLOCKED: {result['error']}"
Enter fullscreen mode Exit fullscreen mode

Configuring the Policy

Set up a policy appropriate for autonomous agents:

{
  "name": "autogpt-policy",
  "limits": {
    "perTransactionLimit": "100000000",
    "dailyLimit": "1000000000",
    "hourlyLimit": "250000000",
    "maxTxPerHour": 20
  },
  "recipientWhitelist": [
    "0x...known-safe-addresses"
  ]
}
Enter fullscreen mode Exit fullscreen mode

Why these limits:

  • $100 per transaction: Limits blast radius of any single decision
  • $1,000 daily: Caps total exposure per day
  • $250 hourly: Prevents rapid drain scenarios
  • 20 tx/hour: Stops infinite loop attacks

Handling Rejections in AutoGPT

AutoGPT will retry failed commands. Configure it to handle policy rejections gracefully:

@command("send_crypto", "Send cryptocurrency with spending limits")
def send_crypto(recipient: str, amount: str, currency: str) -> str:
    result = policy_wallet.send(recipient, amount, currency)

    if result["success"]:
        return f"SUCCESS: Transaction completed. Hash: {result['tx_hash']}"

    error = result["error"]

    # Provide context for the agent to adjust
    if "DAILY_LIMIT" in error:
        return (
            f"BLOCKED: Daily spending limit reached. "
            f"Cannot send more today. Current limit: $1,000/day. "
            f"Do not attempt more transactions until tomorrow."
        )

    if "PER_TX_LIMIT" in error:
        return (
            f"BLOCKED: Amount exceeds per-transaction limit of $100. "
            f"Consider splitting into smaller transactions or "
            f"requesting limit increase from administrator."
        )

    if "RECIPIENT_NOT_WHITELISTED" in error:
        return (
            f"BLOCKED: Recipient address not in approved whitelist. "
            f"Only pre-approved addresses can receive funds. "
            f"Do not attempt to send to this address."
        )

    return f"BLOCKED: {error}"
Enter fullscreen mode Exit fullscreen mode

Memory and Retry Prevention

AutoGPT's memory can cause it to retry blocked transactions. Add explicit instructions:

# In your AutoGPT system prompt or commands.py

CRYPTO_SAFETY_PROMPT = """
When using the send_crypto command:
1. If a transaction is BLOCKED, do not retry with the same parameters
2. If you hit a daily limit, stop all crypto operations until the next day
3. Never attempt to split transactions to circumvent limits
4. Always verify the recipient address is correct before sending
5. Log all blocked transactions for human review
"""
Enter fullscreen mode Exit fullscreen mode

Monitoring AutoGPT Spending

Track your agent's financial activity:

import logging

logger = logging.getLogger("autogpt.spending")

@command("send_crypto", "Send cryptocurrency with spending limits")
def send_crypto(recipient: str, amount: str, currency: str) -> str:
    # Log the attempt
    logger.info(f"Crypto send attempt: {amount} {currency} to {recipient}")

    result = policy_wallet.send(recipient, amount, currency)

    if result["success"]:
        logger.info(f"Transaction succeeded: {result['tx_hash']}")
    else:
        logger.warning(f"Transaction blocked: {result['error']}")

    return format_result(result)
Enter fullscreen mode Exit fullscreen mode

Emergency Kill Switch

If your AutoGPT agent goes rogue, kill it instantly:

# kill_agent.py - Run this to stop all transactions
import requests

def activate_kill_switch():
    response = requests.post(
        "https://api.policylayer.com/kill-switch",
        headers={"Authorization": f"Bearer {ADMIN_API_KEY}"},
        json={
            "scope": "agent",
            "agentId": "autogpt-main",
            "reason": "Emergency shutdown"
        }
    )

    if response.status_code == 200:
        print("Kill switch activated. All transactions blocked.")
    else:
        print(f"Failed to activate kill switch: {response.text}")

if __name__ == "__main__":
    activate_kill_switch()
Enter fullscreen mode Exit fullscreen mode

Once activated, all transactions from that agent are blocked until you manually re-enable.

Testing Before Production

Validate your setup with dry-run mode:

# Test policy without executing real transactions
@command("test_crypto_policy", "Test if a transaction would be allowed")
def test_crypto_policy(recipient: str, amount: str, currency: str) -> str:
    # Only runs Gate 1 - doesn't execute
    validation = policy_wallet._validate_intent(recipient, amount, currency)

    if validation["decision"] == "allow":
        return f"WOULD ALLOW: Transaction within policy limits"
    else:
        return f"WOULD BLOCK: {validation.get('reason')}"
Enter fullscreen mode Exit fullscreen mode

Related Reading

Ready to secure your AutoGPT agents?

Top comments (0)