LangChain Integration

pip install requests langchain-core

"""
Valta Wallet Tool for LangChain
"""
import requests
from typing import Optional, Type
from langchain_core.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")
    description: str = Field(description="Purpose of the spend")
    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")
    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"Balance: {data.get('balance', 0)} USDC | "
                f"Status: {data.get('status', 'unknown')} | "
                f"Daily limit: {data.get('limits', {}).get('daily', 'none')} USDC"
            )
        return f"Error: {r.status_code}"

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"Approved: {amount} USDC. New balance: {data.get('newBalance')} USDC"
            return f"Blocked by policy: {data.get('reason', 'limit exceeded')}"
        return f"Error: {r.status_code}"

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:
            return f"Transfer complete. TX: {r.json().get('txId', 'N/A')}"
        return f"Error: {r.status_code}"

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 langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from valta_tool import get_valta_tools

tools = get_valta_tools("sk_valta_your_api_key")

llm = ChatOpenAI(model="gpt-4o")
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

executor.invoke({
    "input": "Check my wallet balance, then spend 5 USDC on API credits if I have enough."
})

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-langchain-agent", "dailyLimit": 50, "perTxLimit": 10}'