CrewAI Integration

pip install requests crewai

"""
Valta Wallet Tool for CrewAI
"""
import requests
from typing import Optional
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

VALTA_BASE_URL = "https://www.valta.co"

class CheckBalanceInput(BaseModel):
    agent_id: str = Field(description="The Valta wallet ID or name to check balance for")

class SpendInput(BaseModel):
    agent_id: str = Field(description="The Valta wallet ID or name making the spend")
    amount: float = Field(description="Amount in USDC to spend")
    description: str = Field(description="What the spend is for")
    category: Optional[str] = Field(default="general", description="Spend category")

class TransferInput(BaseModel):
    from_agent_id: str = Field(description="Wallet ID or name sending funds")
    to_agent_id: str = Field(description="Wallet ID or name receiving funds")
    amount: float = Field(description="Amount in USDC to transfer")
    description: Optional[str] = Field(default="")

class ValtaCheckBalanceTool(BaseTool):
    name: str = "valta_check_balance"
    description: str = (
        "Check how much USDC an AI agent has in its Valta wallet. "
        "Use before spending to confirm funds are available."
    )
    args_schema: type[BaseModel] = CheckBalanceInput
    api_key: str = ""

    def _run(self, agent_id: str) -> str:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        r = requests.get(
            f"{VALTA_BASE_URL}/api/v1/agents/{agent_id}/wallet",
            headers=headers, timeout=10,
        )
        if r.status_code == 200:
            data = r.json().get("wallet", {})
            return (
                f"Agent '{data.get('name', agent_id)}' wallet: "
                f"{data.get('balance', 0)} USDC available | "
                f"Status: {data.get('status', 'unknown')}"
            )
        return f"Error fetching balance: {r.status_code} — {r.text}"

class ValtaSpendTool(BaseTool):
    name: str = "valta_spend"
    description: str = (
        "Request a spend from an agent's Valta wallet. "
        "Approved or blocked based on the agent's spending policy."
    )
    args_schema: type[BaseModel] = SpendInput
    api_key: str = ""

    def _run(self, agent_id: str, amount: float, description: str, category: str = "general") -> str:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        r = requests.post(
            f"{VALTA_BASE_URL}/api/v1/agents/{agent_id}/wallet/spend",
            json={"amount": amount, "description": description, "category": category},
            headers=headers, timeout=10,
        )
        if r.status_code == 200:
            data = r.json()
            if data.get("approved"):
                return f"Spend approved: {amount} USDC for '{description}'. New balance: {data.get('newBalance', 'N/A')} USDC"
            return f"Spend blocked by policy: {data.get('reason', 'limit exceeded')}"
        return f"Spend request failed: {r.status_code} — {r.text}"

class ValtaTransferTool(BaseTool):
    name: str = "valta_transfer"
    description: str = "Transfer USDC between two agent wallets on Valta."
    args_schema: type[BaseModel] = TransferInput
    api_key: str = ""

    def _run(self, from_agent_id: str, to_agent_id: str, amount: float, description: str = "") -> str:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        r = requests.post(
            f"{VALTA_BASE_URL}/api/v1/wallet/transfer",
            json={"fromAgentId": from_agent_id, "toAgentId": to_agent_id, "amount": amount, "description": description},
            headers=headers, timeout=10,
        )
        if r.status_code == 200:
            data = r.json()
            return f"Transfer complete: {amount} USDC from {from_agent_id} → {to_agent_id}. TX: {data.get('txId', 'N/A')}"
        return f"Transfer failed: {r.status_code} — {r.text}"

def get_valta_tools(api_key: str) -> list:
    return [
        ValtaCheckBalanceTool(api_key=api_key),
        ValtaSpendTool(api_key=api_key),
        ValtaTransferTool(api_key=api_key),
    ]

from crewai import Agent, Task, Crew
from valta_tool import get_valta_tools

tools = get_valta_tools("sk_valta_your_api_key")

financial_agent = Agent(
    role="Autonomous Financial Agent",
    goal="Manage purchases within budget",
    backstory="An agent that handles payments safely.",
    tools=tools,
    verbose=True,
)

task = Task(
    description="Check your wallet balance, then spend 5 USDC on API credits if funds allow.",
    agent=financial_agent,
)

crew = Crew(agents=[financial_agent], tasks=[task])
crew.kickoff()

curl -X POST "https://www.valta.co/api/v1/wallets" \
  -H "x-api-key: sk_valta_your_key" \
  -H "Content-Type: application/json" \
  -d '{"name": "my-crew-agent", "dailyLimit": 50, "perTxLimit": 10}'