How to Add Spending Limits to CrewAI Agents

CrewAI excels at orchestrating multiple AI agents working together. But when your crew handles money, coordination isn’t enough. You need controls that prevent any single agent—or the entire crew—from draining your wallet.

This guide shows you how to integrate PolicyLayer with CrewAI for secure financial operations.

The Problem: Crews Without Limits

A typical CrewAI setup for financial operations might look like this:

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Market Researcher",
    goal="Find arbitrage opportunities"
)

trader = Agent(
    role="Trader",
    goal="Execute profitable trades",
    tools=[execute_trade_tool]  # ⚠️ Direct wallet access
)

crew = Crew(
    agents=[researcher, trader],
    tasks=[research_task, trading_task]
)

The problem: execute_trade_tool has unrestricted wallet access. The trader agent can execute any transaction the researcher suggests—including hallucinated opportunities that drain your funds.

The Solution: Policy-Wrapped Tools

Instead of giving agents direct wallet access, we create tools that route through PolicyLayer. The agent interacts with the tool normally, but every transaction is validated against your spending policy.

┌──────────┐     ┌──────────┐     ┌──────────┐     ┌─────────────┐
│ Research │────▶│  Trader  │────▶│  Policy  │────▶│ Transaction │
│  Agent   │     │  Agent   │     │  Layer   │     │  Executed   │
└──────────┘     └──────────┘     └──────────┘     └─────────────┘
                                        │
                                        ▼
                              [Validated against limits]

Step 1: Install Dependencies

pip install crewai policylayer-sdk
npm install @policylayer/sdk  # For the TypeScript SDK

Or if using Python throughout:

pip install crewai requests  # requests for PolicyLayer API calls

Step 2: Create the Policy-Aware Tool

import os
import json
import requests
from crewai import Tool

class PolicyAwareTradeTool(Tool):
    name = "secure_trade"
    description = """
    Execute a trade with spending limits enforced.
    Input: JSON with 'action' (buy/sell), 'asset', 'amount', 'price'
    """

    def __init__(self):
        self.api_key = os.environ["POLICYLAYER_API_KEY"]
        self.api_url = "https://api.policylayer.com"
        self.wallet_address = os.environ["WALLET_ADDRESS"]

    def _run(self, input_str: str) -> str:
        try:
            params = json.loads(input_str)
        except json.JSONDecodeError:
            return "ERROR: Invalid JSON input"

        action = params.get("action")
        asset = params.get("asset", "usdc")
        amount = params.get("amount")

        if not all([action, amount]):
            return "ERROR: Missing required fields"

        # Convert to smallest unit (assuming 6 decimals for USDC)
        amount_raw = str(int(float(amount) * 1_000_000))

        # Gate 1: Validate intent against policy
        validation = self._validate_intent(asset, amount_raw)

        if validation["decision"] != "allow":
            return f"BLOCKED: {validation.get('reason', 'Policy violation')}"

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

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

        # Execute the actual trade (your trading logic here)
        result = self._execute_trade(action, asset, amount)

        return f"SUCCESS: {action} {amount} {asset} - {result}"

    def _validate_intent(self, asset: str, amount: 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,
                    "to": "0x...exchange-address",
                    "amount": amount,
                    "nonce": str(uuid.uuid4()),
                    "expWindowSeconds": 60
                },
                "idempotencyKey": str(uuid.uuid4()),
                "metadata": {
                    "agentId": "crewai-trader"
                }
            }
        )
        return response.json()

    def _verify_authorisation(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_trade(self, action: str, asset: str, amount: str) -> str:
        # Your actual trading logic here
        # This is where you'd interact with an exchange API
        return f"tx_hash_abc123"

Step 3: Configure Your Policy

In your PolicyLayer dashboard, create a policy for your trading agent:

{
  "name": "crewai-trading-policy",
  "asset": "usdc",
  "limits": {
    "perTransactionLimit": "10000000000",
    "dailyLimit": "100000000000",
    "hourlyLimit": "25000000000",
    "maxTxPerHour": 50
  },
  "recipientWhitelist": [
    "0x...exchange-address-1",
    "0x...exchange-address-2"
  ]
}

This policy:

• Limits each trade to $10,000
• Caps daily trading volume at $100,000
• Restricts hourly volume to $25,000
• Allows maximum 50 trades per hour
• Only allows transactions to whitelisted exchanges

Step 4: Build the Crew

from crewai import Agent, Task, Crew

# Create the policy-aware tool
trade_tool = PolicyAwareTradeTool()

# Research agent (no wallet access)
researcher = Agent(
    role="Market Analyst",
    goal="Identify profitable trading opportunities",
    backstory="Expert in crypto market analysis",
    verbose=True
)

# Trading agent (policy-enforced wallet access)
trader = Agent(
    role="Trade Executor",
    goal="Execute trades within risk parameters",
    backstory="Disciplined trader focused on risk management",
    tools=[trade_tool],
    verbose=True
)

# Define tasks
research_task = Task(
    description="Analyse current market conditions and identify one trading opportunity",
    agent=researcher,
    expected_output="A specific trade recommendation with asset, direction, and size"
)

trading_task = Task(
    description="Execute the recommended trade using the secure_trade tool",
    agent=trader,
    expected_output="Trade execution confirmation or rejection reason",
    context=[research_task]
)

# Create and run crew
crew = Crew(
    agents=[researcher, trader],
    tasks=[research_task, trading_task],
    verbose=True
)

result = crew.kickoff()

Step 5: Handle Rejections Gracefully

When a trade exceeds policy limits, the tool returns a BLOCKED response. Your agent should handle this:

trader = Agent(
    role="Trade Executor",
    goal="Execute trades within risk parameters",
    backstory="""
    You are a disciplined trader. If a trade is blocked by policy,
    you should:
    1. Report why it was blocked
    2. Suggest a smaller trade that might be approved
    3. Never attempt to circumvent the policy
    """,
    tools=[trade_tool],
    verbose=True
)

Multi-Agent Spending Coordination

For crews where multiple agents can spend, use separate policies per agent:

# Conservative researcher with small budget for data purchases
researcher_tool = PolicyAwareTool(
    api_key=os.environ["RESEARCHER_API_KEY"],
    agent_id="researcher",
    daily_limit=100  # $100/day for data
)

# Trader with larger budget
trader_tool = PolicyAwareTool(
    api_key=os.environ["TRADER_API_KEY"],
    agent_id="trader",
    daily_limit=100000  # $100k/day for trading
)

# Ops agent for fees and gas
ops_tool = PolicyAwareTool(
    api_key=os.environ["OPS_API_KEY"],
    agent_id="ops",
    daily_limit=1000  # $1k/day for gas and fees
)

Each agent has isolated limits. Even if all agents are compromised, total daily exposure is bounded.

Monitoring Your Crew

PolicyLayer provides real-time visibility into crew spending:

# Fetch spending stats via API
response = requests.get(
    f"{api_url}/api/dashboard/agents",
    headers={"Authorization": f"Bearer {dashboard_token}"}
)

for agent in response.json():
    print(f"{agent['agentId']}: ${agent['counters']['todaySpent']} spent today")

Or use the PolicyLayer dashboard for real-time monitoring of all agent activity.

---

Related Reading

• Securing LangChain Agents — Similar integration for LangChain
• Two-Gate Enforcement — How policy validation works
• The Kill Switch — Emergency controls for crews

---

Ready to secure your CrewAI agents?

• Quick Start Guide — Production-ready in 5 minutes
• GitHub — Open source SDK