"""Thin OpenClaw-facing adapter for agent wallet backends.""" from __future__ import annotations import hashlib import json import time from typing import Any from agent_wallet.approval import inspect_approval_token, verify_approval_token from agent_wallet.autonomous_policy import OperationRequest from agent_wallet.exceptions import ProviderError from agent_wallet.models import AgentToolResult, AgentToolSpec from agent_wallet.providers import x402 from agent_wallet.wallet_layer.base import AgentWalletBackend, WalletBackendError def _canonical_json_text(payload: dict[str, Any]) -> str: return json.dumps(payload, sort_keys=True, separators=(",", ":")) def preview_payload_digest(preview: dict[str, Any]) -> str: return hashlib.sha256(_canonical_json_text(preview).encode("utf-8")).hexdigest() WALLET_RUNTIME_INSTRUCTIONS = """ Use wallet tools only when the user explicitly asks for wallet-related actions. Treat any signing request as sensitive. Before signing a message, make sure the user intent is explicit and the purpose is clear. Never claim that funds were moved unless a transfer tool explicitly returns a confirmed transaction result. If the wallet backend is sign-only, do not describe the action as broadcast on-chain. If the backend supports signing but should not broadcast, prefer preview and host-mediated execution planning instead of returning signed transactions to the agent. For transfers, prefer preview mode first. Only use execute mode after explicit user approval. Prepare mode must never expose signed transaction bytes to the agent. Execute mode requires a host-issued approval token bound to the exact operation being performed. On mainnet, execute mode requires an approval token that includes an explicit mainnet confirmation. Before any mainnet execute, restate the network, operation type, asset, amount, and destination, validator, or stake account. If the preview result includes a confirmation_summary or mainnet_warning, surface it before asking for confirmation. Never bypass the approval token requirement for wallet writes. In OpenClaw, switch between Solana, EVM, and Bitcoin wallets with set_wallet_backend. The plugin config is the startup default, not something to edit during a normal conversation. For EVM wallets, switch between Ethereum, Base, and Robinhood with set_evm_network or by passing the network argument to EVM tools. Do not edit code, plugin config, or environment variables just to switch the active EVM network. """.strip() EVM_NATIVE_TOKEN_ADDRESS = "0x0000000000000000000000000000000000000000" VELORA_NATIVE_TOKEN_ADDRESS = "0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee" SOLANA_NATIVE_TOKEN_ADDRESS = "11111111111111111111111111111111" LIFI_CHAIN_ALIASES = { "eth": "1", "ethereum": "1", "mainnet": "1", "eth-mainnet": "1", "1": "1", "base": "8453", "base-mainnet": "8453", "8453": "8453", "sol": "1151111081099710", "solana": "1151111081099710", "1151111081099710": "1151111081099710", } class OpenClawWalletAdapter: """Expose wallet backend primitives as safe agent-facing tools.""" def __init__(self, backend: AgentWalletBackend): self.backend = backend def _is_mainnet_network(self, network: Any) -> bool: chain = str(getattr(self.backend, "chain", "")).strip().lower() normalized = str(network or "").strip().lower() if chain == "bitcoin": return normalized == "bitcoin" if chain == "evm": return normalized in {"ethereum", "base", "robinhood", "eip155:1", "eip155:8453", "eip155:4663"} if chain == "solana": return normalized in {"mainnet", "solana:5eykt4usfv8p8njdtrepy1vzkqzkvdp"} return normalized == "mainnet" def _is_mainnet(self) -> bool: return self._is_mainnet_network(getattr(self.backend, "network", "")) def _is_mainnet_for_backend(self, backend: AgentWalletBackend) -> bool: return self._is_mainnet_network(getattr(backend, "network", "")) def _supports_evm_velora(self) -> bool: return str(getattr(self.backend, "chain", "")).strip().lower() == "evm" and self._is_mainnet() def _supports_evm_velora_for_backend(self, backend: AgentWalletBackend) -> bool: return str(getattr(backend, "chain", "")).strip().lower() == "evm" and self._is_mainnet_for_backend(backend) def _normalize_evm_tool_network(self, value: Any) -> str: network = str(value or "").strip().lower() aliases = { "mainnet": "ethereum", "eth": "ethereum", "eth-mainnet": "ethereum", "base-mainnet": "base", } network = aliases.get(network, network) if network in {"sepolia", "base-sepolia", "base_sepolia"}: raise WalletBackendError("EVM testnets are no longer supported. Use ethereum, base, or robinhood.") if network not in {"ethereum", "base", "robinhood"}: raise WalletBackendError("EVM network must be 'ethereum', 'base', or 'robinhood'.") return network def _resolve_backend_for_args(self, args: dict[str, Any]) -> AgentWalletBackend: if str(getattr(self.backend, "chain", "")).strip().lower() != "evm": return self.backend requested_network = args.get("network") if requested_network is None: return self.backend if not isinstance(requested_network, str) or not requested_network.strip(): raise WalletBackendError("network must be a non-empty string when provided.") return self.backend.with_network(self._normalize_evm_tool_network(requested_network)) def _normalize_optional_non_negative_number( self, value: Any, *, field_name: str ) -> float | int | None: if value is None: return None if isinstance(value, bool) or not isinstance(value, (int, float)): raise WalletBackendError(f"{field_name} must be a number when provided.") if value < 0: raise WalletBackendError(f"{field_name} must not be negative.") return value def _normalize_positive_limit(self, value: Any, *, field_name: str, default: int, maximum: int) -> int: if value is None: return default if not isinstance(value, int) or value <= 0: raise WalletBackendError(f"{field_name} must be a positive integer.") return min(value, maximum) def _normalize_lifi_slippage(self, value: Any) -> float | int | None: if value is None: return None if isinstance(value, bool) or not isinstance(value, (int, float)): raise WalletBackendError("slippage must be a number when provided.") if value < 0 or value > 1: raise WalletBackendError("slippage must be between 0 and 1, for example 0.01 for 1%.") return value def _normalize_optional_string_list(self, value: Any, *, field_name: str) -> list[str] | None: if value is None: return None if isinstance(value, str): items = [item.strip() for item in value.split(",") if item.strip()] return items or None if not isinstance(value, list): raise WalletBackendError(f"{field_name} must be an array of strings when provided.") items: list[str] = [] for item in value: if not isinstance(item, str) or not item.strip(): raise WalletBackendError(f"{field_name} must contain only non-empty strings.") items.append(item.strip()) return items def _canonicalize_lifi_chain_identifier(self, value: Any) -> str: text = str(value or "").strip().lower() return LIFI_CHAIN_ALIASES.get(text, text) def _canonicalize_lifi_token_identifier(self, value: Any, *, chain_id: str) -> str: text = str(value or "").strip() alias = text.lower() if chain_id in {"1", "8453"}: if alias in {"native", "eth", "ethereum"}: return EVM_NATIVE_TOKEN_ADDRESS if alias.startswith("0x") and len(alias) == 42: return alias return text if chain_id == "1151111081099710" and alias in {"native", "sol", "solana"}: return SOLANA_NATIVE_TOKEN_ADDRESS return text def _canonicalize_velora_token_identifier(self, value: Any) -> str: text = str(value or "").strip() alias = text.lower() if alias in {"native", "eth", "ethereum"}: return VELORA_NATIVE_TOKEN_ADDRESS if alias == EVM_NATIVE_TOKEN_ADDRESS: return VELORA_NATIVE_TOKEN_ADDRESS if alias == VELORA_NATIVE_TOKEN_ADDRESS: return VELORA_NATIVE_TOKEN_ADDRESS if alias.startswith("0x") and len(alias) == 42: return alias return text def _require_prepare_intent(self, user_intent: Any) -> None: if user_intent is not True: raise WalletBackendError( "Prepare mode requires explicit user intent confirmation." ) def _x402_tool_specs(self) -> list[AgentToolSpec]: return [ AgentToolSpec( name="x402_search_services", description=( "Search x402-paid services through CDP Bazaar or Agentic Market. " "This is read-only discovery and does not spend funds." ), input_schema={ "type": "object", "properties": { "query": {"type": "string"}, "discovery_provider": { "type": "string", "enum": ["auto", "cdp_bazaar", "agentic_market"], }, "network": {"type": "string"}, "asset": {"type": "string"}, "scheme": {"type": "string"}, "max_usd_price": {"type": "string"}, "limit": {"type": "integer"}, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="x402_get_service_details", description=( "Resolve one x402 service or resource into a normalized details payload. " "Use a resource URL for CDP Bazaar or a domain/service id for Agentic Market." ), input_schema={ "type": "object", "properties": { "reference": {"type": "string"}, "discovery_provider": { "type": "string", "enum": ["auto", "cdp_bazaar", "agentic_market"], }, }, "required": ["reference"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="x402_preview_request", description=( "Make an unpaid HTTP request to an x402 endpoint, detect HTTP 402, parse " "PAYMENT-REQUIRED, and summarize the payment options. This does not pay or execute." ), input_schema={ "type": "object", "properties": { "url": {"type": "string"}, "method": {"type": "string"}, "headers": {"type": "object", "additionalProperties": {"type": "string"}}, "query": {"type": "object", "additionalProperties": True}, "json_body": {}, "text_body": {"type": "string"}, }, "required": ["url"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="x402_pay_request", description=( "Pay for and call an x402 endpoint using the active wallet backend. " "The tool probes the endpoint, validates compatibility, signs the payment, " "and returns the service response in one call." ), input_schema={ "type": "object", "properties": { "url": {"type": "string"}, "method": {"type": "string"}, "headers": {"type": "object", "additionalProperties": {"type": "string"}}, "query": {"type": "object", "additionalProperties": True}, "json_body": {}, "text_body": {"type": "string"}, "purpose": {"type": "string"}, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode when the payment is mainnet/real-value and no autonomous session or permission grant covers it.", }, }, "required": ["url", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=False, risk_level="high", ), ] @staticmethod def _autonomous_signals(summary: dict[str, Any]) -> tuple[str | None, int | None]: """Best-effort extraction of (recipient, smallest-unit spend) from a summary. Only raw/base-unit amount fields are trusted for spend accounting; a UI-denominated amount is ignored so caps are never applied at the wrong scale. When no raw amount is present the spend is reported as ``None`` (unknown), which the session layer treats as fail-closed when caps are configured. """ recipient: str | None = None for key in ("to_address", "to", "recipient", "destination", "destination_address", "spender"): value = summary.get(key) if isinstance(value, str) and value.strip(): recipient = value.strip() break spend: int | None = None for key in ("input_amount_raw", "amount_raw", "amount_wei", "amount_lamports", "amount_base_units"): value = summary.get(key) if value is None: continue try: spend = int(str(value)) except (TypeError, ValueError): continue break return recipient, spend def _build_session_config_from_args(self, args: dict[str, Any]): """Construct an AutonomousSessionConfig from start_autonomous_session args.""" from agent_wallet.autonomous_policy import AutonomousSessionConfig from agent_wallet.spending_limits import SpendingConfig def _str_list(value: Any, field: str) -> list[str]: if value is None: return [] if not isinstance(value, list) or any(not isinstance(v, str) or not v.strip() for v in value): raise WalletBackendError(f"{field} must be an array of non-empty strings.") return [v.strip() for v in value] def _int(value: Any, field: str) -> int: if value is None: return 0 if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise WalletBackendError(f"{field} must be a non-negative integer.") return value return AutonomousSessionConfig( enabled=True, allowed_tools=frozenset(_str_list(args.get("allowed_tools"), "allowed_tools")), allowed_networks=frozenset(_str_list(args.get("allowed_networks"), "allowed_networks")), allow_mainnet=bool(args.get("allow_mainnet", False)), allowed_recipients=frozenset(_str_list(args.get("allowed_recipients"), "allowed_recipients")), allow_any_recipient=bool(args.get("allow_any_recipient", False)), require_simulation=bool(args.get("require_simulation", True)), spending=SpendingConfig( max_per_tx_lamports=_int(args.get("max_per_tx_lamports"), "max_per_tx_lamports"), max_hourly_lamports=_int(args.get("max_hourly_lamports"), "max_hourly_lamports"), max_daily_lamports=_int(args.get("max_daily_lamports"), "max_daily_lamports"), max_txs_per_minute=_int(args.get("max_txs_per_minute"), "max_txs_per_minute"), ), max_operations=_int(args.get("max_operations"), "max_operations"), session_ttl_seconds=_int(args.get("session_ttl_seconds"), "session_ttl_seconds"), approval_ttl_seconds=_int(args.get("approval_ttl_seconds"), "approval_ttl_seconds") or 120, ) async def _authorize_autonomous_permission( self, *, active_backend: AgentWalletBackend, tool_name: str, action_label: str, preview_kwargs: dict[str, Any], preview_method: Any, scope: str, ) -> tuple[str, dict[str, Any]]: """Authorize one operation through a high-trust permission toggle.""" from agent_wallet import autonomous_permissions network = str(getattr(active_backend, "network", "unknown")).strip().lower() if scope == autonomous_permissions.BASE_SWAP_SCOPE: if network != autonomous_permissions.BASE_SWAP_NETWORK: raise WalletBackendError( "Autonomous Base swap permission only applies on network=base. " "Use set_evm_network or the network parameter to select Base." ) if not autonomous_permissions.is_base_swap_approved(): raise WalletBackendError( "Autonomous execution is not enabled. Ask the user to run " "agentlayer_autonomous_approve first." ) elif scope == autonomous_permissions.DEFI_TOOLS_SCOPE: if network not in autonomous_permissions.DEFI_TOOLS_NETWORKS: raise WalletBackendError( "Autonomous DeFi permission only applies on ethereum, base, or robinhood. " "Use set_evm_network or the network parameter to select a supported network." ) if not autonomous_permissions.is_defi_tools_approved(): raise WalletBackendError( "Autonomous execution is not enabled. Ask the user to run " "agentlayer_autonomous_approve first." ) else: raise WalletBackendError("Unsupported autonomous permission scope.") preview = await preview_method(**preview_kwargs) annotated_preview = self._annotate_sensitive_payload( preview, action_label=action_label, mode="preview", ) summary = dict(annotated_preview.get("confirmation_summary") or {}) if not summary: raise WalletBackendError("Autonomous preview did not produce a confirmation_summary.") if scope == autonomous_permissions.BASE_SWAP_SCOPE: token = autonomous_permissions.authorize_base_swap( tool_name=tool_name, network=network, summary=summary, ) else: token = autonomous_permissions.authorize_defi_tool( tool_name=tool_name, network=network, summary=summary, ) return token, summary async def _authorize_base_swap_permission( self, *, active_backend: AgentWalletBackend, tool_name: str, action_label: str, preview_kwargs: dict[str, Any], preview_method: Any, ) -> tuple[str, dict[str, Any]]: """Authorize one Base swap through the high-trust permission toggle.""" from agent_wallet import autonomous_permissions return await self._authorize_autonomous_permission( active_backend=active_backend, tool_name=tool_name, action_label=action_label, preview_kwargs=preview_kwargs, preview_method=preview_method, scope=autonomous_permissions.BASE_SWAP_SCOPE, ) async def _authorize_defi_permission( self, *, active_backend: AgentWalletBackend, tool_name: str, action_label: str, preview_kwargs: dict[str, Any], preview_method: Any, ) -> tuple[str, dict[str, Any]]: """Authorize one EVM DeFi operation through the high-trust permission toggle.""" from agent_wallet import autonomous_permissions return await self._authorize_autonomous_permission( active_backend=active_backend, tool_name=tool_name, action_label=action_label, preview_kwargs=preview_kwargs, preview_method=preview_method, scope=autonomous_permissions.DEFI_TOOLS_SCOPE, ) def _require_execute_approval( self, *, approval_token: Any, tool_name: str, summary: dict[str, Any], action_label: str, backend: AgentWalletBackend | None = None, ) -> None: active_backend = backend or self.backend network = str(getattr(active_backend, "network", "unknown")) if not isinstance(approval_token, str) or not approval_token.strip(): # No host token: fall back to the autonomous policy engine if (and # only if) a human-authorized session envelope is active. The engine # mints the same signed token the host would, so verification below # is identical for both paths. from agent_wallet import autonomous_session if autonomous_session.is_active(): recipient, spend = self._autonomous_signals(summary) approval_token = autonomous_session.authorize_operation( OperationRequest( tool_name=tool_name, network=network, summary=summary, spend_amount=spend, recipient=recipient, simulated=True, ) ) else: # No budgeted session either: fall back to the standing # high-trust permission toggle (agentlayer_autonomous_approve). # This is the single choke point every execute tool funnels # through, so enabling that toggle covers every tool here -- # not just the Base-swap/EVM-DeFi tools that also have their # own dedicated pre-authorization step before reaching here. from agent_wallet import autonomous_permissions if autonomous_permissions.is_autonomous_approved(): approval_token = autonomous_permissions.authorize_operation( tool_name=tool_name, network=network, summary=summary, ) else: raise WalletBackendError( f"{action_label} execution requires a host-issued approval_token " "(or an active autonomous session/permission grant authorized by the host)." ) verify_approval_token( approval_token.strip(), tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), summary=summary, require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) # Enforce single-use: reject replayed approval tokens. from agent_wallet.nonce_registry import require_single_use require_single_use(approval_token.strip()) def _build_confirmation_summary( self, *, action_label: str, payload: dict[str, Any], ) -> dict[str, Any]: asset_type = str(payload.get("asset_type") or "").strip().lower() if asset_type == "swap": swap_quote_binding = { "swap_provider": payload.get("swap_provider"), "estimated_output_amount_ui": payload.get("estimated_output_amount_ui"), "minimum_output_amount_ui": payload.get("minimum_output_amount_ui"), "price_impact_pct": payload.get("price_impact_pct"), "fee_summary": payload.get("fee_summary"), "route_plan": payload.get("route_plan"), } quote_fingerprint = hashlib.sha256( json.dumps( swap_quote_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() summary: dict[str, Any] = { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "swap_provider": payload.get("swap_provider"), "estimated_output_amount_ui": payload.get("estimated_output_amount_ui"), "minimum_output_amount_ui": payload.get("minimum_output_amount_ui"), "price_impact_pct": payload.get("price_impact_pct"), "quote_fingerprint": quote_fingerprint, } for key in ( "owner", "input_mint", "output_mint", "input_amount_ui", "input_amount_raw", "slippage_bps", ): value = payload.get(key) if value is not None: summary[key] = value return summary if asset_type == "solana-swap-intent": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "owner": payload.get("owner"), "input_mint": payload.get("input_mint"), "output_mint": payload.get("output_mint"), "input_amount_ui": payload.get("input_amount_ui"), "input_amount_raw": payload.get("input_amount_raw"), "minimum_output_amount_raw": payload.get("minimum_output_amount_raw"), "minimum_output_amount_ui": payload.get("minimum_output_amount_ui"), "max_slippage_bps": payload.get("max_slippage_bps"), "slippage_bps": payload.get("slippage_bps"), "max_fee_lamports": payload.get("max_fee_lamports"), "valid_until_epoch_seconds": payload.get("valid_until_epoch_seconds"), "valid_for_seconds": payload.get("valid_for_seconds"), "max_attempts": payload.get("max_attempts"), "allowed_providers": payload.get("allowed_providers"), "recipient_policy": payload.get("recipient_policy"), "spend_policy": payload.get("spend_policy"), } if asset_type == "kamino-lend-intent": summary = { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "owner": payload.get("owner"), "kamino_operation": payload.get("kamino_operation"), "market": payload.get("market"), "reserve": payload.get("reserve"), "amount_ui": payload.get("amount_ui"), "recipient_policy": payload.get("recipient_policy"), "spend_policy": payload.get("spend_policy"), "valid_until_epoch_seconds": payload.get("valid_until_epoch_seconds"), } obligation_address = payload.get("obligation_address") if obligation_address is not None: summary["obligation_address"] = obligation_address return summary if asset_type == "kamino-earn-intent": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "owner": payload.get("owner"), "kamino_operation": payload.get("kamino_operation"), "kvault": payload.get("kvault"), "amount_ui": payload.get("amount_ui"), "recipient_policy": payload.get("recipient_policy"), "spend_policy": payload.get("spend_policy"), "valid_until_epoch_seconds": payload.get("valid_until_epoch_seconds"), } if asset_type == "solana-lifi-cross-chain-swap": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "swap_provider": payload.get("swap_provider"), "source_chain": payload.get("source_chain"), "source_chain_id": payload.get("source_chain_id"), "destination_chain": payload.get("destination_chain"), "destination_chain_id": payload.get("destination_chain_id"), "owner": payload.get("owner"), "input_token": payload.get("input_token"), "input_mint": payload.get("input_mint"), "output_token": payload.get("output_token"), "destination_address": payload.get("destination_address"), "input_amount_raw": payload.get("input_amount_raw"), "input_amount_ui": payload.get("input_amount_ui"), "estimated_output_amount_raw": payload.get("estimated_output_amount_raw"), "minimum_output_amount_raw": payload.get("minimum_output_amount_raw"), "slippage": payload.get("slippage"), "quote_type": payload.get("quote_type"), "quote_id": payload.get("quote_id"), "transaction_id": payload.get("transaction_id"), "tool": payload.get("tool"), "transaction_data_hash": payload.get("transaction_data_hash"), } if asset_type == "evm-lifi-cross-chain-swap": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "swap_provider": payload.get("swap_provider"), "source_chain": payload.get("source_chain"), "destination_chain": payload.get("destination_chain"), "token_in": payload.get("token_in"), "output_token": payload.get("output_token"), "destination_address": payload.get("destination_address"), "input_amount_raw": payload.get("input_amount_raw"), "estimated_output_amount_raw": payload.get("estimated_output_amount_raw"), "minimum_output_amount_raw": payload.get("minimum_output_amount_raw"), "slippage_bps": payload.get("slippage_bps"), "slippage": payload.get("slippage"), "gas_drop": payload.get("gas_drop"), "quote_type": payload.get("quote_type"), "quote_id": payload.get("quote_id"), "tool": payload.get("tool"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_swap_fee_wei": payload.get("estimated_swap_fee_wei"), "estimated_approval_fee_wei": payload.get("estimated_approval_fee_wei"), "router": payload.get("router"), } if asset_type == "evm-swap": output_amount_raw = ( payload.get("estimated_output_amount_raw") if payload.get("estimated_output_amount_raw") is not None else payload.get("output_amount_raw") ) provided_fingerprint = payload.get("quote_fingerprint") evm_swap_binding = { "swap_provider": payload.get("swap_provider"), "token_in": payload.get("token_in"), "token_out": payload.get("token_out"), "input_amount_raw": payload.get("input_amount_raw"), "output_amount_raw": output_amount_raw, "minimum_output_amount_raw": payload.get("minimum_output_amount_raw"), "slippage_bps": payload.get("slippage_bps"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "router": payload.get("router"), "swap_transaction": payload.get("swap_transaction"), "quote_fingerprint": provided_fingerprint, } evm_swap_fingerprint = ( str(provided_fingerprint).strip() if isinstance(provided_fingerprint, str) and str(provided_fingerprint).strip() else hashlib.sha256( json.dumps( evm_swap_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() ) return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "swap_provider": payload.get("swap_provider"), "token_in": payload.get("token_in"), "token_out": payload.get("token_out"), "input_amount_raw": payload.get("input_amount_raw"), "output_amount_raw": output_amount_raw, "minimum_output_amount_raw": payload.get("minimum_output_amount_raw"), "slippage_bps": payload.get("slippage_bps"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_swap_fee_wei": payload.get("estimated_swap_fee_wei"), "estimated_approval_fee_wei": payload.get("estimated_approval_fee_wei"), "quote_fingerprint": provided_fingerprint, "router": payload.get("router"), "swap_transaction": payload.get("swap_transaction"), "evm_swap_fingerprint": evm_swap_fingerprint, } if asset_type == "evm-uniswap-swap": provided_fingerprint = payload.get("quote_fingerprint") return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "swap_provider": payload.get("swap_provider"), "routing": payload.get("routing"), "token_in": payload.get("token_in"), "token_out": payload.get("token_out"), "input_amount_raw": payload.get("input_amount_raw"), "estimated_output_amount_raw": payload.get("estimated_output_amount_raw"), "minimum_output_amount_raw": payload.get("minimum_output_amount_raw"), "slippage_bps": payload.get("slippage_bps"), "permit_required": payload.get("permit_required"), "router": payload.get("router"), "quote_fingerprint": provided_fingerprint, } if asset_type == "evm-morpho-vault": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "protocol": payload.get("protocol"), "morpho_surface": payload.get("surface"), "morpho_operation": payload.get("operation"), "target": payload.get("target"), "token_address": payload.get("token_address"), "amount_raw": payload.get("amount_raw"), "native_amount_raw": payload.get("native_amount_raw"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_operation_fee_wei": payload.get("estimated_operation_fee_wei"), "estimated_requirements_fee_wei": payload.get("estimated_requirements_fee_wei"), "quote_fingerprint": payload.get("quote_fingerprint"), "requirements": payload.get("requirements"), } if asset_type == "evm-morpho-market": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "protocol": payload.get("protocol"), "morpho_surface": payload.get("surface"), "morpho_operation": payload.get("operation"), "target": payload.get("target"), "token_address": payload.get("token_address"), "amount_raw": payload.get("amount_raw"), "native_amount_raw": payload.get("native_amount_raw"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_operation_fee_wei": payload.get("estimated_operation_fee_wei"), "estimated_requirements_fee_wei": payload.get("estimated_requirements_fee_wei"), "quote_fingerprint": payload.get("quote_fingerprint"), "requirements": payload.get("requirements"), } if asset_type == "evm-aave-v3": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "protocol": payload.get("protocol"), "aave_operation": payload.get("operation"), "token_address": payload.get("token_address"), "amount_raw": payload.get("amount_raw"), "amount_ui": payload.get("amount_ui"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_operation_fee_wei": payload.get("estimated_operation_fee_wei"), "estimated_approval_fee_wei": payload.get("estimated_approval_fee_wei"), "quote_fingerprint": payload.get("quote_fingerprint"), "allowance": payload.get("allowance"), } if asset_type == "evm-lido-staking": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "protocol": payload.get("protocol"), "lido_operation": payload.get("operation"), "amount_raw": payload.get("amount_raw"), "amount_ui": payload.get("amount_ui"), "expected_output_amount_raw": payload.get("expected_output_amount_raw"), "expected_output_amount_ui": payload.get("expected_output_amount_ui"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_operation_fee_wei": payload.get("estimated_operation_fee_wei"), "estimated_approval_fee_wei": payload.get("estimated_approval_fee_wei"), "quote_fingerprint": payload.get("quote_fingerprint"), "allowance": payload.get("allowance"), } if asset_type == "evm-lido-withdrawal-queue": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "protocol": payload.get("protocol"), "lido_withdrawal_operation": payload.get("operation"), "amount_raw": payload.get("amount_raw"), "amount_ui": payload.get("amount_ui"), "request_id": payload.get("request_id"), "queued_steth_amount_raw": payload.get("queued_steth_amount_raw"), "queued_steth_amount_ui": payload.get("queued_steth_amount_ui"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "estimated_operation_fee_wei": payload.get("estimated_operation_fee_wei"), "estimated_approval_fee_wei": payload.get("estimated_approval_fee_wei"), "quote_fingerprint": payload.get("quote_fingerprint"), "allowance": payload.get("allowance"), } if asset_type == "bags-token-launch": launch_binding = { "token_name": payload.get("token_name"), "token_symbol": payload.get("token_symbol"), "description": payload.get("description"), "image_url": payload.get("image_url"), "website": payload.get("website"), "twitter": payload.get("twitter"), "telegram": payload.get("telegram"), "discord": payload.get("discord"), "base_mint": payload.get("base_mint"), "claimers": payload.get("claimers"), "basis_points": payload.get("basis_points"), "initial_buy_lamports": payload.get("initial_buy_lamports"), "bags_config_type": payload.get("bags_config_type"), } launch_fingerprint = hashlib.sha256( json.dumps( launch_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "owner": payload.get("owner"), "wallet": payload.get("wallet"), "token_name": payload.get("token_name"), "token_symbol": payload.get("token_symbol"), "base_mint": payload.get("base_mint"), "claimers_count": payload.get("claimers_count"), "total_basis_points": payload.get("total_basis_points"), "initial_buy_sol": payload.get("initial_buy_sol"), "initial_buy_lamports": payload.get("initial_buy_lamports"), "launch_fingerprint": launch_fingerprint, } if asset_type in {"flash-trade-open-position", "flash-trade-close-position"}: flash_binding = { "pool_name": payload.get("pool_name"), "market_symbol": payload.get("market_symbol"), "collateral_symbol": payload.get("collateral_symbol"), "collateral_amount_raw": payload.get("collateral_amount_raw"), "leverage": payload.get("leverage"), "side": payload.get("side"), "estimated_size_usd": payload.get("estimated_size_usd"), "estimated_entry_price": payload.get("estimated_entry_price"), "estimated_liquidation_price": payload.get("estimated_liquidation_price"), "position_size_usd": payload.get("position_size_usd"), "close_amount_raw": payload.get("close_amount_raw"), } flash_fingerprint = hashlib.sha256( json.dumps( flash_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "owner": payload.get("owner"), "pool_name": payload.get("pool_name"), "market_symbol": payload.get("market_symbol"), "collateral_symbol": payload.get("collateral_symbol"), "collateral_amount_raw": payload.get("collateral_amount_raw"), "leverage": payload.get("leverage"), "side": payload.get("side"), "estimated_size_usd": payload.get("estimated_size_usd"), "estimated_entry_price": payload.get("estimated_entry_price"), "estimated_liquidation_price": payload.get("estimated_liquidation_price"), "position_size_usd": payload.get("position_size_usd"), "close_amount_raw": payload.get("close_amount_raw"), "flash_preview_fingerprint": flash_fingerprint, } if asset_type == "btc-transfer": btc_binding = { "recipient": payload.get("recipient"), "amount_sats": payload.get("amount_sats"), "fee_rate": payload.get("fee_rate"), "confirmation_target": payload.get("confirmation_target"), "estimated_fee_sats": payload.get("estimated_fee_sats"), } btc_fingerprint = hashlib.sha256( json.dumps( btc_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "recipient": payload.get("recipient"), "amount_sats": payload.get("amount_sats"), "estimated_fee_sats": payload.get("estimated_fee_sats"), "fee_rate": payload.get("fee_rate"), "confirmation_target": payload.get("confirmation_target"), "btc_transfer_fingerprint": btc_fingerprint, } if asset_type == "x402-request": return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "request_url": payload.get("request_url"), "method": payload.get("method"), "request_fingerprint": payload.get("request_fingerprint"), "body_hash": payload.get("body_hash"), "x402_network": payload.get("x402_network"), "x402_scheme": payload.get("x402_scheme"), "x402_asset": payload.get("x402_asset"), "x402_amount": payload.get("x402_amount"), "x402_amount_display": payload.get("x402_amount_display"), "x402_pay_to": payload.get("x402_pay_to"), } if asset_type == "evm-native-transfer": evm_binding = { "recipient": payload.get("recipient"), "amount_wei": payload.get("amount_wei"), "estimated_fee_wei": payload.get("estimated_fee_wei"), } evm_fingerprint = hashlib.sha256( json.dumps( evm_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "recipient": payload.get("recipient"), "amount_wei": payload.get("amount_wei"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "evm_transfer_fingerprint": evm_fingerprint, } if asset_type == "evm-token-transfer": evm_token_binding = { "recipient": payload.get("recipient"), "token_address": payload.get("token_address"), "amount_raw": payload.get("amount_raw"), "estimated_fee_wei": payload.get("estimated_fee_wei"), } evm_token_fingerprint = hashlib.sha256( json.dumps( evm_token_binding, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() return { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), "wallet": payload.get("wallet"), "from_address": payload.get("from_address"), "recipient": payload.get("recipient"), "token_address": payload.get("token_address"), "amount_raw": payload.get("amount_raw"), "estimated_fee_wei": payload.get("estimated_fee_wei"), "evm_token_transfer_fingerprint": evm_token_fingerprint, } summary: dict[str, Any] = { "operation": action_label, "network": str(payload.get("network") or getattr(self.backend, "network", "unknown")), } for key in ( "owner", "authority", "address", "kvault", "obligation_address", "market", "reserve", "amount_native", "amount_ui", "input_amount_ui", "estimated_output_amount_ui", "amount_raw", "mint", "asset", "input_mint", "output_mint", "from_address", "to_address", "recipient", "vote_account", "stake_account", "stake_account_address", "wallet", "fee_claimer", "token_mint", "token_name", "token_symbol", "base_mint", "claimable_position_count", "claimers_count", "total_basis_points", "initial_buy_sol", "initial_buy_lamports", "amount_sats", "estimated_fee_sats", "fee_rate", "confirmation_target", "amount_wei", "estimated_fee_wei", "token_address", "amount_raw", "token_in", "token_out", "input_amount_raw", "estimated_output_amount_raw", "output_amount_raw", "swap_provider", ): value = payload.get(key) if value is not None: summary[key] = value return summary def _build_prepare_plan( self, *, preview_payload: dict[str, Any], action_label: str, ) -> dict[str, Any]: plan = dict(preview_payload) plan["mode"] = "prepare" plan["prepared"] = False plan["signed"] = False plan["broadcasted"] = False plan["confirmed"] = False plan["execution_plan_only"] = True plan["prepare_note"] = ( "Signed transaction bytes are intentionally not returned by prepare mode. " "Use this as an execution plan and perform final signing or broadcast only through execute or a host-controlled path." ) for key in ( "transaction_base64", "transaction_encoding", "transaction_format", "signature", "last_valid_block_height", "latest_blockhash", "request_id", "verification", ): plan.pop(key, None) return plan def _annotate_sensitive_payload( self, payload: dict[str, Any], *, action_label: str, mode: str, ) -> dict[str, Any]: annotated = dict(payload) network = str(annotated.get("network") or getattr(self.backend, "network", "unknown")).strip().lower() is_mainnet = self._is_mainnet_network(network) annotated["network"] = network annotated["is_mainnet"] = is_mainnet # Stamp the mode so the host bridge can cache preview/prepare responses for # auto-approval at execute. The bridge already treats "preview" as cacheable # (alongside "prepare"/"intent_preview"); without this field a preview -> execute # flow never caches and execute fails with "confirmation context expired", # regardless of timing. prepare carries its own mode via _build_prepare_plan; # "execute" responses stay uncached (not in the bridge's cacheable set). annotated["mode"] = mode annotated["confirmation_summary"] = self._build_confirmation_summary( action_label=action_label, payload=annotated, ) annotated["confirmation_requirements"] = { "prepare_requires_user_intent": mode == "prepare", "execute_requires_approval_token": True, "execute_requires_mainnet_confirmed_in_token": is_mainnet, } if mode == "preview": annotated["approval_hint"] = { "host_must_issue_token_for": annotated["confirmation_summary"], "tool_name": None, } if is_mainnet and mode in {"preview", "prepare", "execute"}: annotated["mainnet_warning"] = ( "Mainnet operation. Confirm the network, asset, amount, and destination, validator, or stake account " "before execute. Execute requires a host-issued approval token with mainnet confirmation." ) return annotated def _annotate_x402_payload( self, payload: dict[str, Any], *, mode: str, ) -> dict[str, Any]: if not payload.get("payment_required"): return dict(payload) annotated = self._annotate_sensitive_payload( payload, action_label="x402 paid request", mode=mode, ) summary = dict(annotated.get("confirmation_summary") or {}) if summary: annotated["payment_summary"] = summary requirements = dict(annotated.get("confirmation_requirements") or {}) requirements["prepare_requires_user_intent"] = False de_minimis = x402.is_de_minimis_payment(payload) if de_minimis: # Below x402.DE_MINIMIS_USD_THRESHOLD, this payment never needs # approval -- not even on mainnet -- so keep the advertised # requirement honest in both preview and execute responses. requirements["execute_requires_approval_token"] = False requirements["execute_requires_mainnet_confirmed_in_token"] = False annotated["de_minimis_payment"] = { "applied": True, "threshold_usd": x402.DE_MINIMIS_USD_THRESHOLD, "amount_usd": x402.de_minimis_usd_amount(payload), } annotated.pop("approval_hint", None) if mode == "preview": annotated.pop("confirmation_summary", None) annotated["confirmation_requirements"] = requirements if annotated.get("is_mainnet") and not de_minimis: annotated["preview_note"] = ( "This is a paid mainnet endpoint preview only. Review the service URL, network, asset, amount, and payment destination before calling x402_pay_request." ) return annotated annotated["confirmation_requirements"] = requirements if annotated.get("is_mainnet") and not de_minimis: annotated["mainnet_warning"] = ( "Mainnet x402 payment. Confirm the service URL, network, asset, amount, and payment destination before paying." ) return annotated def list_tools(self) -> list[AgentToolSpec]: """Return wallet tools suitable for agent registration.""" capabilities = self.backend.get_capabilities() if capabilities.chain == "evm": tools = [ AgentToolSpec( name="get_wallet_capabilities", description="Describe the connected wallet backend, chain, and safety limits.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_address", description="Return the configured wallet address for the connected backend.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_balance", description=( "Get the EVM wallet overview: native asset, discovered ERC-20 balances, " "per-asset USD values, assets, balance_usd, and total_value_usd when available. " "Prices come from aggregator APIs, not RPC." ), input_schema={ "type": "object", "properties": { "address": { "type": "string", "description": "Optional wallet address override.", }, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_lifi_supported_chains", description="List the LI.FI chains currently allowed for OpenClaw cross-chain routing.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_lifi_quote", description="Get a read-only LI.FI cross-chain quote for Ethereum/Base/Solana routes. Execution is not enabled by this tool.", input_schema={ "type": "object", "properties": { "from_chain": {"type": "string", "description": "Source chain: ethereum, base, solana, or the LI.FI chain id."}, "to_chain": {"type": "string", "description": "Destination chain: ethereum, base, solana, or the LI.FI chain id."}, "from_token": {"type": "string", "description": "Source token address. Use native/eth/sol for native tokens."}, "to_token": {"type": "string", "description": "Destination token address. Use native/eth/sol for native tokens."}, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "from_address": { "type": "string", "description": "Optional source wallet address. Defaults to the active wallet when the source chain matches it.", }, "to_address": { "type": "string", "description": "Optional destination wallet address. Defaults to the active wallet when the destination chain matches it.", }, "slippage": { "type": "number", "description": "Optional decimal fraction, for example 0.01 for 1%.", }, "allow_bridges": {"type": "array", "items": {"type": "string"}}, "deny_bridges": {"type": "array", "items": {"type": "string"}}, "prefer_bridges": {"type": "array", "items": {"type": "string"}}, }, "required": ["from_chain", "to_chain", "from_token", "to_token", "amount_in_raw"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_lifi_transfer_status", description="Get LI.FI cross-chain transfer status using a source/destination transaction hash or LI.FI step id.", input_schema={ "type": "object", "properties": { "tx_hash": {"type": "string"}, "bridge": {"type": "string"}, "from_chain": {"type": "string"}, "to_chain": {"type": "string"}, }, "required": ["tx_hash"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_evm_network", description="Show the effective EVM network context, available networks, and swap-supported networks.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="set_evm_network", description=( "Select the active EVM network for subsequent wallet tool calls in this " "runtime session. Use this to switch between ethereum, base, and robinhood instead " "of editing code or plugin configuration." ), input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "EVM network to make active for subsequent calls.", }, }, "required": ["network"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_evm_token_balance", description="Get the ERC-20 token balance for the configured EVM wallet account.", input_schema={ "type": "object", "properties": { "token_address": { "type": "string", "description": "ERC-20 token contract address.", }, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "required": ["token_address"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_evm_token_metadata", description="Get ERC-20 token metadata for a contract address on the active EVM network.", input_schema={ "type": "object", "properties": { "token_address": { "type": "string", "description": "ERC-20 token contract address.", }, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "required": ["token_address"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_evm_fee_rates", description="Get current EVM fee-rate suggestions for the active network.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_evm_transaction_receipt", description="Get the transaction receipt for a broadcast EVM transaction hash.", input_schema={ "type": "object", "properties": { "tx_hash": { "type": "string", "description": "0x-prefixed EVM transaction hash.", }, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "required": ["tx_hash"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="transfer_evm_native", description=( "Preview, prepare, or execute a native EVM transfer using an amount in wei. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "recipient": {"type": "string"}, "amount_wei": { "type": "string", "description": "Transfer amount in wei as a base-10 integer string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "required": ["recipient", "amount_wei", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), AgentToolSpec( name="transfer_evm_token", description=( "Preview, prepare, or execute an ERC-20 transfer using a raw base-unit amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "token_address": {"type": "string"}, "recipient": {"type": "string"}, "amount_raw": { "type": "string", "description": "Transfer amount in token base units as a base-10 integer string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "Optional EVM network override for this request.", }, }, "required": ["token_address", "recipient", "amount_raw", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ] if self._supports_evm_velora(): tools.insert( 6, AgentToolSpec( name="get_evm_aave_account", description="Get read-only Aave V3 account data for the configured EVM wallet on supported mainnet networks.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 7, AgentToolSpec( name="get_evm_aave_reserves", description="Get the read-only Aave V3 reserve catalog for the configured EVM network, including reserve flags, pricing, and liquidity metadata.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 8, AgentToolSpec( name="get_evm_aave_positions", description="Get read-only Aave V3 per-reserve positions for the configured EVM wallet, including supplied and borrowed balances on supported mainnet networks.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 9, AgentToolSpec( name="manage_evm_aave_position", description=( "Preview, prepare, or execute a narrow Aave V3 lending operation on supported EVM mainnet networks. " "Supported operations are supply, withdraw, borrow, and repay. Prepare returns an execution plan only, " "and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["supply", "withdraw", "borrow", "repay"], }, "token_address": { "type": "string", "description": "Underlying ERC-20 reserve token address.", }, "amount_raw": { "type": "string", "description": "Amount in token base units as a base-10 integer string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "required": ["operation", "token_address", "amount_raw", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 10, AgentToolSpec( name="get_evm_lido_overview", description="Get the read-only Lido staking overview for the configured EVM wallet on supported networks, including contract addresses and sample wrap rates.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 11, AgentToolSpec( name="get_evm_lido_positions", description="Get read-only Lido positions for the configured EVM wallet, including stETH, wstETH, and stETH-equivalent balances.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 12, AgentToolSpec( name="manage_evm_lido_position", description=( "Preview, prepare, or execute a narrow Lido staking operation on Ethereum mainnet. " "Supported operations are stake_eth_for_wsteth, wrap_steth, and unwrap_wsteth. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["stake_eth_for_wsteth", "wrap_steth", "unwrap_wsteth"], }, "amount_raw": { "type": "string", "description": "Amount in base units as a base-10 integer string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum"], "description": "Optional EVM network override for this request.", }, }, "required": ["operation", "amount_raw", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 13, AgentToolSpec( name="get_evm_lido_withdrawal_requests", description="Get read-only Lido withdrawal queue requests for the configured EVM wallet, including finalized and claimable request statuses.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 14, AgentToolSpec( name="manage_evm_lido_withdrawal", description=( "Preview, prepare, or execute a narrow Lido withdrawal queue operation on Ethereum mainnet. " "Supported operations are request_withdrawal_steth, request_withdrawal_wsteth, and claim_withdrawal. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "operation": { "type": "string", "enum": [ "request_withdrawal_steth", "request_withdrawal_wsteth", "claim_withdrawal", ], }, "amount_raw": { "type": "string", "description": "Amount in base units as a base-10 integer string. Required for request operations.", }, "request_id": { "type": "string", "description": "Withdrawal request id as a base-10 integer string. Required for claim_withdrawal.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum"], "description": "Optional EVM network override for this request.", }, }, "required": ["operation", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 15, AgentToolSpec( name="get_evm_morpho_vaults", description="Get read-only Morpho vault discovery and detail data for the configured EVM network on supported mainnet chains. When listing, results are ordered (default: largest TVL first) and can be filtered by underlying asset.", input_schema={ "type": "object", "properties": { "vault_address": { "type": "string", "description": "Optional explicit Morpho vault address for a single-vault lookup.", }, "limit": { "type": "integer", "minimum": 1, "maximum": 500, "description": "Optional max number of vaults to return when listing.", }, "listed_only": { "type": "boolean", "description": "Filter to listed vaults only. Defaults to true.", }, "asset_address": { "type": "string", "description": "Optional underlying asset address to filter vaults (e.g. only USDC vaults).", }, "min_tvl_usd": { "type": "number", "minimum": 0, "description": "Optional minimum vault TVL in USD. Recommended when sorting by APY, so dust vaults with inflated yields are excluded (e.g. 1000000).", }, "min_net_apy": { "type": "number", "minimum": 0, "description": "Optional minimum net APY as a fraction (e.g. 0.03 for 3%).", }, "order_by": { "type": "string", "enum": [ "TotalAssetsUsd", "TotalAssets", "TotalSupply", "Liquidity", "LiquidityUsd", "Apy", "NetApy", "Address", ], "description": "Optional sort field when listing. Defaults to TotalAssetsUsd (largest TVL first). When sorting by Apy/NetApy, pair with min_tvl_usd to avoid dust vaults.", }, "order_direction": { "type": "string", "enum": ["asc", "desc"], "description": "Optional sort direction. Defaults to desc.", }, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 16, AgentToolSpec( name="get_evm_morpho_markets", description="Get read-only Morpho market discovery and detail data for the configured EVM network on supported mainnet chains. When listing, results are ordered (default: largest supply first) and can be filtered by a free-text search or by collateral/loan asset.", input_schema={ "type": "object", "properties": { "market_id": { "type": "string", "description": "Optional explicit Morpho market id (32-byte hex) for a single-market lookup.", }, "limit": { "type": "integer", "minimum": 1, "maximum": 500, "description": "Optional max number of markets to return when listing.", }, "listed_only": { "type": "boolean", "description": "Filter to listed markets only. Defaults to true.", }, "search": { "type": "string", "description": "Optional free-text search over market/asset symbols (e.g. 'wstETH').", }, "collateral_asset_address": { "type": "string", "description": "Optional collateral asset address to filter markets.", }, "loan_asset_address": { "type": "string", "description": "Optional loan asset address to filter markets.", }, "min_supply_usd": { "type": "number", "minimum": 0, "description": "Optional minimum market supply in USD. Recommended when sorting by APY, so dust markets with inflated yields are excluded (e.g. 1000000).", }, "min_net_supply_apy": { "type": "number", "minimum": 0, "description": "Optional minimum net supply APY as a fraction (e.g. 0.03 for 3%).", }, "order_by": { "type": "string", "enum": [ "SupplyAssetsUsd", "BorrowAssetsUsd", "SupplyApy", "NetSupplyApy", "BorrowApy", "NetBorrowApy", "Utilization", "TotalLiquidityUsd", "Lltv", ], "description": "Optional sort field when listing. Defaults to SupplyAssetsUsd (largest supply first). When sorting by APY, pair with min_supply_usd to avoid dust markets.", }, "order_direction": { "type": "string", "enum": ["asc", "desc"], "description": "Optional sort direction. Defaults to desc.", }, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 17, AgentToolSpec( name="get_evm_morpho_positions", description="Get read-only Morpho vault and market positions for the configured EVM wallet on supported mainnet chains.", input_schema={ "type": "object", "properties": { "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 18, AgentToolSpec( name="manage_evm_morpho_vault_position", description=( "Preview, prepare, or execute a narrow Morpho vault operation on supported EVM mainnet networks. " "Supported operations are supply and withdraw. Prepare returns an execution plan only, " "and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["supply", "withdraw"], }, "token_address": { "type": "string", "description": "Vault asset token address.", }, "vault_address": { "type": "string", "description": "Optional explicit Morpho vault address.", }, "vault_preset": { "type": "string", "description": "Optional Morpho vault preset name.", }, "amount_raw": { "type": "string", "description": "Asset amount in base units. Required for withdraw and optional for supply when native_amount_raw is used.", }, "native_amount_raw": { "type": "string", "description": "Optional native-wrap amount in base units for supply.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "required": ["operation", "token_address", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 19, AgentToolSpec( name="manage_evm_morpho_market_position", description=( "Preview, prepare, or execute a narrow Morpho market operation on supported EVM mainnet networks. " "Supported operations are supply_collateral, borrow, repay, and withdraw_collateral. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["supply_collateral", "borrow", "repay", "withdraw_collateral"], }, "token_address": { "type": "string", "description": "Loan token or collateral token address for the selected operation.", }, "market_id": { "type": "string", "description": "Optional explicit Morpho market id.", }, "market_preset": { "type": "string", "description": "Optional Morpho market preset name.", }, "amount_raw": { "type": "string", "description": "Operation amount in base units. For repay, use a positive integer string or the literal max.", }, "native_amount_raw": { "type": "string", "description": "Optional native-wrap amount in base units for supply_collateral.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "required": ["operation", "token_address", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 8, AgentToolSpec( name="get_evm_swap_quote", description=( "Get a read-only Velora quote for an ERC-20 or native ETH swap on supported EVM mainnet networks. " "This does not approve or execute a swap." ), input_schema={ "type": "object", "properties": { "token_in": { "type": "string", "description": "ERC-20 contract address for the input token, or native/eth for native ETH.", }, "token_out": { "type": "string", "description": "ERC-20 contract address for the output token, or native/eth for native ETH.", }, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "required": ["token_in", "token_out", "amount_in_raw"], "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 9, AgentToolSpec( name="swap_evm_tokens", description=( "Preview, prepare, or execute an ERC-20 or native ETH swap through Velora on supported EVM mainnet networks. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation " "unless the high-trust autonomous permission group is enabled." ), input_schema={ "type": "object", "properties": { "token_in": {"type": "string"}, "token_out": {"type": "string"}, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "required": ["token_in", "token_out", "amount_in_raw", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 10, AgentToolSpec( name="swap_evm_lifi_cross_chain_tokens", description=( "Preview, prepare, or execute an EVM-origin cross-chain swap through LI.FI. " "This currently supports ethereum/base as the source network and ethereum/base/solana as the destination chain. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "token_in": { "type": "string", "description": "Source EVM token contract address, native, eth, or the zero address for native ETH.", }, "destination_chain": { "type": "string", "enum": [ "ethereum", "base", "solana", "1", "8453", "1151111081099710", ], }, "output_token": { "type": "string", "description": "Destination token identifier, for example the Solana USDC mint.", }, "destination_address": { "type": "string", "description": "Destination wallet address on the target chain.", }, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "slippage": { "type": "number", "description": "Optional decimal fraction, for example 0.01 for 1%.", }, "allow_bridges": {"type": "array", "items": {"type": "string"}}, "deny_bridges": {"type": "array", "items": {"type": "string"}}, "prefer_bridges": {"type": "array", "items": {"type": "string"}}, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base"], "description": "Optional EVM network override for this request.", }, }, "required": [ "token_in", "destination_chain", "output_token", "destination_address", "amount_in_raw", "mode", "purpose", ], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 11, AgentToolSpec( name="get_uniswap_swap_quote", description=( "Get a read-only Uniswap execution preview for an ERC-20 or native ETH swap " "on ethereum, base, or robinhood. Supported paths are CLASSIC, UniswapX orders, " "and canonical ETH↔WETH wrap/unwrap. Supports Permit2 where required. " "This does not approve, sign, or execute a swap." ), input_schema={ "type": "object", "properties": { "token_in": { "type": "string", "description": "ERC-20 contract address for the input token, or native/eth for native ETH.", }, "token_out": { "type": "string", "description": "ERC-20 contract address for the output token, or native/eth for native ETH.", }, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "slippage_bps": { "type": "integer", "description": "Optional slippage tolerance in basis points (e.g. 50 = 0.5%, 300 = 3%). Defaults to 300.", }, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "EVM network. Uniswap Trading API supports ethereum, base, and robinhood.", }, }, "required": ["token_in", "token_out", "amount_in_raw"], "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.append( AgentToolSpec( name="search_uniswap_pairs", description=( "Search for onchain Uniswap-tradeable token pairs by ticker/name or by exact " "ERC-20 address, with live price, 5m/1h/6h/24h volume, liquidity, FDV, market " "cap, and buy/sell counts. Backed by DexScreener's public index (covers " "Uniswap v2/v3/v4 pools and other DEXes on the same chain), since the Uniswap " "Trading API itself only quotes a pair you already know the addresses for, not " "search-by-name. Defaults to the currently active EVM network (e.g. robinhood); " "set all_chains to search across every chain DexScreener indexes. Read-only " "market data, not a swap quote. Free-text search can surface impersonator " "tokens reusing a legitimate ticker with fabricated liquidity/FDV — before " "quoting or swapping, verify the token_address independently (e.g. via " "get_evm_token_metadata) rather than trusting the top result by symbol match. " "Do not use liquidity/FDV size to pick a winner: observed impersonators on " "robinhood fabricate liquidity/FDV an order of magnitude above the genuine " "pool specifically to look most trustworthy, and a Uniswap quote succeeding " "for a token proves nothing either (Trading API routes any pool with reserves, " "impersonators included). This only matters for tickers claiming to represent " "a real-world asset (stocks/ETFs) - unrelated meme/community tokens on the " "same chain aren't impersonators just for lacking a canonical issuer. When the " "query is a real-world-asset ticker, cross-check the returned token_address " "against Robinhood's own canonical list at docs.robinhood.com/chain/contracts " "(or the equivalent official source for other chains) before relying on the result." ), input_schema={ "type": "object", "properties": { "query": { "type": "string", "description": "Free-text search, e.g. a ticker or name (\"AAPL\", \"Apple Robinhood\"). Required unless token_address is given.", }, "token_address": { "type": "string", "description": "Exact ERC-20 contract address to look up instead of a free-text query.", }, "chain": { "type": "string", "description": "DexScreener chain slug to filter to (e.g. \"robinhood\", \"ethereum\", \"base\"). Defaults to the active EVM network.", }, "dex_id": { "type": "string", "description": "Optional exact DEX filter, e.g. \"uniswap\".", }, "all_chains": { "type": "boolean", "description": "If true, do not filter query results to a single chain. Ignored when token_address is set.", }, "limit": { "type": "integer", "description": "Max number of pairs to return (1-30). Defaults to 10.", }, }, "required": [], "additionalProperties": False, }, read_only=True, risk_level="low", ), ) tools.insert( 12, AgentToolSpec( name="swap_evm_uniswap_tokens", description=( "Preview, prepare, or execute a supported Uniswap path on ethereum, base, or robinhood: " "CLASSIC, UniswapX orders, or canonical ETH↔WETH wrap/unwrap. ERC-20 paths may use Permit2 " "EIP-712 or an UniswapX order signature. " "Prepare returns an execution plan only. Execute requires a host-issued approval token bound to the previewed operation " "unless the high-trust autonomous permission group is enabled." ), input_schema={ "type": "object", "properties": { "token_in": { "type": "string", "description": "ERC-20 contract address or native/eth for native ETH.", }, "token_out": { "type": "string", "description": "ERC-20 contract address or native/eth for native ETH.", }, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "slippage_bps": { "type": "integer", "description": "Optional slippage tolerance in basis points (e.g. 50 = 0.5%, 300 = 3%). Defaults to 300.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, "network": { "type": "string", "enum": ["ethereum", "base", "robinhood"], "description": "EVM network. Uniswap Trading API supports ethereum, base, and robinhood.", }, }, "required": ["token_in", "token_out", "amount_in_raw", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.insert( 13, AgentToolSpec( name="issue_wallet_approval", description=( "Issue a host approval token bound to an exact wallet operation. " "Call this after prepare mode to get the token required for execute mode. " "Pass tool_name and the confirmation_summary exactly as returned in the prepare response. " "In Claude Code this is the bridge step that Codex performs automatically via its UI dialog." ), input_schema={ "type": "object", "properties": { "tool_name": { "type": "string", "description": "Name of the tool being approved, e.g. swap_evm_uniswap_tokens or swap_evm_tokens.", }, "summary": { "type": "object", "description": "The confirmation_summary dict returned verbatim from the prepare response.", "additionalProperties": True, }, "mainnet_confirmed": { "type": "boolean", "description": "Set to true to confirm this is a mainnet operation and you accept the risk.", }, }, "required": ["tool_name", "summary", "mainnet_confirmed"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ) tools.extend(self._x402_tool_specs()) tools.extend(self._autonomous_permission_tool_specs()) return tools if capabilities.chain == "bitcoin": return [ AgentToolSpec( name="get_wallet_capabilities", description="Describe the connected wallet backend, chain, and safety limits.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_address", description="Return the configured wallet address for the connected backend.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_balance", description="Get the native token balance for the configured wallet address.", input_schema={ "type": "object", "properties": { "address": { "type": "string", "description": "Optional wallet address override.", } }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_btc_transfer_history", description="Get BTC transfer history for the configured wallet account.", input_schema={ "type": "object", "properties": { "direction": { "type": "string", "enum": ["incoming", "outgoing", "all"], "description": "Optional transfer direction filter.", }, "limit": { "type": "integer", "description": "Maximum number of transfers to return. Defaults to 10.", }, "skip": { "type": "integer", "description": "Optional offset for paginated history queries.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_btc_fee_rates", description="Get current BTC fee-rate suggestions from the connected wallet service.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_btc_max_spendable", description="Estimate the maximum BTC amount spendable after fees for the configured wallet account.", input_schema={ "type": "object", "properties": { "fee_rate": { "type": "integer", "description": "Optional fee rate in sats/vB to price the estimate.", } }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="transfer_btc", description=( "Preview, prepare, or execute a BTC transfer in satoshis. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "recipient": {"type": "string"}, "amount_sats": { "type": "integer", "description": "Transfer amount in satoshis.", }, "fee_rate": { "type": "integer", "description": "Optional fee rate in sats/vB.", }, "confirmation_target": { "type": "integer", "description": "Optional target confirmation blocks for fee estimation.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, }, "required": ["recipient", "amount_sats", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), ] tools = [ AgentToolSpec( name="get_wallet_capabilities", description="Describe the connected wallet backend, chain, and safety limits.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_address", description="Return the configured wallet address for the connected backend.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_balance", description=( "Get the wallet overview for the configured Solana address: native SOL, " "non-zero SPL token accounts, per-asset USD values when available, and total_value_usd. " "Prices come from Jupiter, not Solana RPC." ), input_schema={ "type": "object", "properties": { "address": { "type": "string", "description": "Optional wallet address override. If omitted, use the configured wallet address.", } }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_lifi_supported_chains", description="List the LI.FI chains currently allowed for OpenClaw cross-chain routing.", input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_lifi_quote", description="Get a read-only LI.FI cross-chain quote for Ethereum/Base/Solana routes. Execution is not enabled by this tool.", input_schema={ "type": "object", "properties": { "from_chain": {"type": "string", "description": "Source chain: ethereum, base, solana, or the LI.FI chain id."}, "to_chain": {"type": "string", "description": "Destination chain: ethereum, base, solana, or the LI.FI chain id."}, "from_token": {"type": "string", "description": "Source token address. Use native/eth/sol for native tokens."}, "to_token": {"type": "string", "description": "Destination token address. Use native/eth/sol for native tokens."}, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "from_address": { "type": "string", "description": "Optional source wallet address. Defaults to the active wallet when the source chain matches it.", }, "to_address": { "type": "string", "description": "Optional destination wallet address. Defaults to the active wallet when the destination chain matches it.", }, "slippage": { "type": "number", "description": "Optional decimal fraction, for example 0.01 for 1%.", }, "allow_bridges": {"type": "array", "items": {"type": "string"}}, "deny_bridges": {"type": "array", "items": {"type": "string"}}, "prefer_bridges": {"type": "array", "items": {"type": "string"}}, }, "required": ["from_chain", "to_chain", "from_token", "to_token", "amount_in_raw"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_lifi_transfer_status", description="Get LI.FI cross-chain transfer status using a source/destination transaction hash or LI.FI step id.", input_schema={ "type": "object", "properties": { "tx_hash": {"type": "string"}, "bridge": {"type": "string"}, "from_chain": {"type": "string"}, "to_chain": {"type": "string"}, }, "required": ["tx_hash"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_wallet_portfolio", description=( "Get the Solana wallet portfolio. This is the detailed equivalent of get_wallet_balance " "and includes native SOL, non-zero SPL token accounts, USD pricing when available, and total_value_usd." ), input_schema={ "type": "object", "properties": { "address": { "type": "string", "description": "Optional wallet address override. If omitted, use the configured wallet address.", } }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_solana_token_prices", description="Get current token prices for one or more Solana mint addresses via Jupiter.", input_schema={ "type": "object", "properties": { "mints": { "type": "array", "items": {"type": "string"}, "description": "List of Solana token mint addresses.", } }, "required": ["mints"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_solana_staking_validators", description="List native Solana staking validators by vote account, commission, and activated stake.", input_schema={ "type": "object", "properties": { "limit": { "type": "integer", "description": "Maximum number of validators to return. Defaults to 20.", }, "include_delinquent": { "type": "boolean", "description": "If true, include delinquent validators after current ones.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_solana_stake_account", description="Inspect a native Solana stake account and its activation status.", input_schema={ "type": "object", "properties": { "stake_account": { "type": "string", "description": "Stake account address to inspect.", } }, "required": ["stake_account"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_flash_trade_markets", description="List Flash Trade perpetual markets currently available on Solana mainnet.", input_schema={ "type": "object", "properties": { "pool_name": { "type": "string", "description": "Optional Flash pool identifier such as Crypto.1.", } }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_flash_trade_positions", description="Get Flash Trade perpetual positions for a Solana wallet on mainnet.", input_schema={ "type": "object", "properties": { "owner": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, "pool_name": { "type": "string", "description": "Optional Flash pool identifier such as Crypto.1.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="flash_trade_open_position", description=( "Preview, prepare, or execute a Flash Trade perpetual open on Solana mainnet using a supported Flash collateral." ), input_schema={ "type": "object", "properties": { "pool_name": { "type": "string", "description": "Flash pool identifier such as Crypto.1.", }, "market_symbol": { "type": "string", "description": "Flash market symbol such as SOL or BTC.", }, "collateral_symbol": { "type": "string", "description": "Flash collateral symbol, for example SOL for SOL longs or USDC for SOL shorts.", }, "collateral_amount_raw": { "type": "string", "description": "Collateral amount in raw token units.", }, "leverage": { "type": "string", "description": "Requested leverage as a decimal string such as 5 or 7.5.", }, "side": { "type": "string", "enum": ["long", "short"], "description": "Position direction.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns trade details; prepare returns an execution plan; execute broadcasts after host approval.", }, "purpose": { "type": "string", "description": "Short explanation of why the position should be opened.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": [ "pool_name", "market_symbol", "collateral_symbol", "collateral_amount_raw", "leverage", "side", "mode", "purpose", ], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), AgentToolSpec( name="flash_trade_close_position", description=( "Preview, prepare, or execute a Flash Trade perpetual close on Solana mainnet." ), input_schema={ "type": "object", "properties": { "pool_name": { "type": "string", "description": "Flash pool identifier such as Crypto.1.", }, "market_symbol": { "type": "string", "description": "Flash market symbol such as SOL or BTC.", }, "side": { "type": "string", "enum": ["long", "short"], "description": "Position direction to close.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns close details; prepare returns an execution plan; execute broadcasts after host approval.", }, "purpose": { "type": "string", "description": "Short explanation of why the position should be closed.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": [ "pool_name", "market_symbol", "side", "mode", "purpose", ], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ), AgentToolSpec( name="get_kamino_portfolio", description="Get the unified Kamino portfolio view for a Solana wallet on mainnet across lending, multiply, leverage, liquidity, earn, and staking.", input_schema={ "type": "object", "properties": { "user": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_vaults", description=( "Discover Kamino Earn vaults on Solana mainnet. Returns compact vault " "summaries pre-ranked by AUM; pass include_metrics=true to fetch APY/TVL " "metrics for the top vaults (sorted by APY), token_mint to filter by " "deposit token, or vault_address for one vault's detail with metrics. " "For yield search, combine token_mint with include_metrics." ), input_schema={ "type": "object", "properties": { "vault_address": { "type": "string", "description": "Optional vault address for a single-vault lookup with APY/TVL metrics.", }, "token_mint": { "type": "string", "description": "Optional deposit token mint to filter vaults (e.g. EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v for USDC).", }, "include_metrics": { "type": "boolean", "description": "Fetch APY/TVL metrics for the top vaults and sort by APY. Defaults to false. Metrics are fetched for at most 20 vaults per call.", }, "limit": { "type": "integer", "minimum": 1, "maximum": 100, "description": "Optional max number of vaults to return. Defaults to 50 (capped at 20 when include_metrics is true).", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_earn_positions", description="Get Kamino Earn vault positions for a Solana wallet on mainnet.", input_schema={ "type": "object", "properties": { "user": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_liquidity_positions", description="Get Kamino Liquidity strategy positions for a Solana wallet on mainnet.", input_schema={ "type": "object", "properties": { "user": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_lend_markets", description=( "List Kamino lending markets currently available on Solana mainnet. " "The list has no yield data; markets flagged isPrimary (Main Market) hold " "most TVL — start there and call get_kamino_lend_market_reserves for APYs." ), input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_lend_market_reserves", description=( "Get reserve metrics for one Kamino lending market on Solana mainnet: " "supplyApy, borrowApy, TVL in USD, and maxLtv per token — the data to use " "for lend/borrow yield comparisons within a market." ), input_schema={ "type": "object", "properties": { "market": { "type": "string", "description": "Kamino market address.", } }, "required": ["market"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_lend_user_obligations", description="Get Kamino obligations for a wallet in a specific Kamino market on Solana mainnet.", input_schema={ "type": "object", "properties": { "market": { "type": "string", "description": "Kamino market address.", }, "user": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, }, "required": ["market"], "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_lend_user_rewards", description="Get Kamino rewards summary for a Solana wallet on mainnet.", input_schema={ "type": "object", "properties": { "user": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="get_kamino_open_positions", description=( "Get all open Kamino lending positions for a Solana wallet on mainnet, " "aggregated across markets with loan details, reserve APYs, and rewards." ), input_schema={ "type": "object", "properties": { "user": { "type": "string", "description": "Optional Solana wallet address override. If omitted, use the configured wallet.", }, }, "additionalProperties": False, }, read_only=True, risk_level="low", ), ] if capabilities.can_sign_message: tools.append( AgentToolSpec( name="sign_wallet_message", description=( "Sign an arbitrary message with the connected wallet. " "Only use after explicit user approval." ), input_schema={ "type": "object", "properties": { "message": { "type": "string", "description": "Exact message to sign.", }, "purpose": { "type": "string", "description": "Short explanation of why the signature is needed.", }, "user_confirmed": { "type": "boolean", "description": "Must be true if the user explicitly approved the signature request.", }, }, "required": ["message", "purpose", "user_confirmed"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="medium", ) ) if capabilities.chain == "solana": tools.append( AgentToolSpec( name="transfer_sol", description=( "Preview or execute a native SOL transfer. " "Use preview first, then execute only after explicit user approval." ), input_schema={ "type": "object", "properties": { "recipient": { "type": "string", "description": "Destination Solana wallet address.", }, "amount": { "type": "number", "description": "Amount of SOL to transfer.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns a transfer summary, prepare returns an execution plan without signed transaction bytes, execute attempts to send.", }, "purpose": { "type": "string", "description": "Short explanation of why the transfer is being made.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["recipient", "amount", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="stake_sol_native", description=( "Preview, prepare, or execute native SOL staking to a validator vote account " "through the Solana Stake Program." ), input_schema={ "type": "object", "properties": { "vote_account": { "type": "string", "description": "Validator vote account address.", }, "amount": { "type": "number", "description": "Amount of SOL to stake, excluding rent reserve.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns a staking summary, prepare returns an execution plan without signed transaction bytes, execute attempts to send.", }, "purpose": { "type": "string", "description": "Short explanation of why the staking action is being made.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["vote_account", "amount", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="transfer_spl_token", description=( "Preview or execute an SPL token transfer by mint address. " "Use preview first, then execute only after explicit user approval." ), input_schema={ "type": "object", "properties": { "recipient": { "type": "string", "description": "Destination Solana wallet address.", }, "mint": { "type": "string", "description": "SPL token mint address.", }, "amount": { "type": "number", "description": "Token amount in UI units.", }, "decimals": { "type": "integer", "description": "Optional token decimals override. If omitted, fetch from chain.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns a transfer summary, prepare returns an execution plan without signed transaction bytes, execute attempts to send.", }, "purpose": { "type": "string", "description": "Short explanation of why the transfer is being made.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["recipient", "mint", "amount", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="swap_solana_tokens", description=( "Preview or execute a Solana token swap through Jupiter routing. " "Prefer intent_preview, then intent_execute after explicit chat confirmation; " "intent_execute fetches a fresh quote and only sends if it remains inside the approved limits. " "OpenClaw should not use legacy execute for Solana swaps because exact Jupiter quote payloads expire quickly." ), input_schema={ "type": "object", "properties": { "input_mint": { "type": "string", "description": "Input token mint address. Use native SOL mint for SOL swaps.", }, "output_mint": { "type": "string", "description": "Output token mint address. Use native SOL mint for SOL swaps.", }, "amount": { "type": "number", "description": "Input token amount in UI units.", }, "slippage_bps": { "type": "integer", "description": "Optional slippage tolerance in basis points. Defaults to 300 (3%) for Solana swaps.", }, "minimum_output_amount_raw": { "type": "integer", "description": "Optional approved minimum output in raw output token units for intent_preview. For intent swaps, overly strict values are clamped to the slippage floor.", }, "max_fee_lamports": { "type": "integer", "description": "Optional maximum Solana network fee in lamports for intent_preview.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds. Intent swaps use at least 120 seconds, max 120.", }, "max_attempts": { "type": "integer", "description": "Optional number of fresh quote/simulate/execute attempts. Intent swaps use at least 3 attempts, max 5.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "intent_preview returns approved risk limits; intent_execute requotes and executes atomically inside those limits. Legacy preview/prepare/execute remains supported.", }, "purpose": { "type": "string", "description": "Short explanation of why the swap is being made.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["input_mint", "output_mint", "amount", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="swap_solana_lifi_cross_chain_tokens", description=( "Preview, prepare, or execute a Solana-origin cross-chain swap through LI.FI. " "This currently supports Solana as the source chain and ethereum/base as the destination chain. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "input_token": { "type": "string", "description": "Source Solana token mint, native, sol, or 11111111111111111111111111111111 for native SOL.", }, "destination_chain": { "type": "string", "enum": ["ethereum", "base", "1", "8453"], }, "output_token": { "type": "string", "description": "Destination EVM token contract address, native, eth, or the zero address for native ETH.", }, "destination_address": { "type": "string", "description": "Destination EVM wallet address on the target chain.", }, "amount_in_raw": { "type": "string", "description": "Input amount in token base units as a base-10 integer string.", }, "slippage": { "type": "number", "description": "Optional decimal fraction, for example 0.01 for 1%.", }, "allow_bridges": {"type": "array", "items": {"type": "string"}}, "deny_bridges": {"type": "array", "items": {"type": "string"}}, "prefer_bridges": {"type": "array", "items": {"type": "string"}}, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], }, "purpose": {"type": "string"}, "user_intent": {"type": "boolean"}, "approval_token": {"type": "string"}, }, "required": [ "input_token", "destination_chain", "output_token", "destination_address", "amount_in_raw", "mode", "purpose", ], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="launch_bags_token", description=( "Preview, prepare, or execute a Bags token launch with fee-share config on mainnet. " "Prepare returns an execution plan only, and execute requires a host-issued approval token." ), input_schema={ "type": "object", "properties": { "name": {"type": "string", "description": "Token name."}, "symbol": {"type": "string", "description": "Token ticker symbol."}, "description": {"type": "string", "description": "Token description."}, "image_url": { "type": "string", "description": "Optional hosted token image URL.", }, "website": {"type": "string", "description": "Optional project website URL."}, "twitter": {"type": "string", "description": "Optional project Twitter/X handle or URL."}, "telegram": {"type": "string", "description": "Optional Telegram URL or handle."}, "discord": {"type": "string", "description": "Optional Discord URL."}, "base_mint": { "type": "string", "description": "Base mint used for Bags fee share configuration.", }, "claimers": { "type": "array", "items": {"type": "string"}, "description": "Fee-share claimer wallet addresses.", }, "basis_points": { "type": "array", "items": {"type": "integer"}, "description": "Fee-share split in basis points. Must sum to 10000.", }, "initial_buy_sol": { "type": "number", "description": "Initial buy amount in SOL. Use 0 for no initial buy.", }, "bags_config_type": { "type": "integer", "description": "Optional Bags fee-share config type override.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns a launch summary, prepare returns an execution plan without signed transaction bytes, execute attempts to sign and broadcast the launch transaction.", }, "purpose": { "type": "string", "description": "Short explanation of why the token is being launched.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": [ "name", "symbol", "description", "base_mint", "claimers", "basis_points", "initial_buy_sol", "mode", "purpose", ], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="kamino_lend_deposit", description=( "Preview, prepare, or execute a Kamino lending deposit using a decimal token amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "market": {"type": "string", "description": "Kamino market address."}, "reserve": {"type": "string", "description": "Kamino reserve address."}, "amount_ui": { "type": "string", "description": "Decimal token amount to deposit, as a string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "Prefer intent_preview then intent_execute after explicit chat confirmation: intent_execute re-derives the Kamino transaction and executes within the approved parameters without round-tripping the preview payload. Legacy preview/prepare/execute remains supported.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds for intent_preview (1-300, default 120).", }, "purpose": {"type": "string", "description": "Short explanation of why the deposit is being made."}, "user_intent": {"type": "boolean", "description": "Must be true for prepare mode."}, "approval_token": {"type": "string", "description": "Host-issued approval token required for execute/intent_execute mode."}, }, "required": ["market", "reserve", "amount_ui", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="kamino_lend_withdraw", description=( "Preview, prepare, or execute a Kamino lending withdraw using a decimal token amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "market": {"type": "string", "description": "Kamino market address."}, "reserve": {"type": "string", "description": "Kamino reserve address."}, "amount_ui": { "type": "string", "description": "Decimal token amount to withdraw, as a string.", }, "obligation_address": { "type": "string", "description": "Optional Kamino obligation address. Required when preview shows multiple matching obligations.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "Prefer intent_preview then intent_execute after explicit chat confirmation: intent_execute re-derives the Kamino transaction and executes within the approved parameters without round-tripping the preview payload. Legacy preview/prepare/execute remains supported.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds for intent_preview (1-300, default 120).", }, "purpose": {"type": "string", "description": "Short explanation of why the withdraw is being made."}, "user_intent": {"type": "boolean", "description": "Must be true for prepare mode."}, "approval_token": {"type": "string", "description": "Host-issued approval token required for execute/intent_execute mode."}, }, "required": ["market", "reserve", "amount_ui", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="kamino_lend_borrow", description=( "Preview, prepare, or execute a Kamino lending borrow using a decimal token amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "market": {"type": "string", "description": "Kamino market address."}, "reserve": {"type": "string", "description": "Kamino reserve address."}, "amount_ui": { "type": "string", "description": "Decimal token amount to borrow, as a string.", }, "obligation_address": { "type": "string", "description": "Optional Kamino obligation address. Required when preview shows multiple obligations in the selected market.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "Prefer intent_preview then intent_execute after explicit chat confirmation: intent_execute re-derives the Kamino transaction and executes within the approved parameters without round-tripping the preview payload. Legacy preview/prepare/execute remains supported.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds for intent_preview (1-300, default 120).", }, "purpose": {"type": "string", "description": "Short explanation of why the borrow is being made."}, "user_intent": {"type": "boolean", "description": "Must be true for prepare mode."}, "approval_token": {"type": "string", "description": "Host-issued approval token required for execute/intent_execute mode."}, }, "required": ["market", "reserve", "amount_ui", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="kamino_lend_repay", description=( "Preview, prepare, or execute a Kamino lending repay using a decimal token amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "market": {"type": "string", "description": "Kamino market address."}, "reserve": {"type": "string", "description": "Kamino reserve address."}, "amount_ui": { "type": "string", "description": "Decimal token amount to repay, as a string.", }, "obligation_address": { "type": "string", "description": "Optional Kamino obligation address. Required when preview shows multiple matching debt obligations.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "Prefer intent_preview then intent_execute after explicit chat confirmation: intent_execute re-derives the Kamino transaction and executes within the approved parameters without round-tripping the preview payload. Legacy preview/prepare/execute remains supported.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds for intent_preview (1-300, default 120).", }, "purpose": {"type": "string", "description": "Short explanation of why the repay is being made."}, "user_intent": {"type": "boolean", "description": "Must be true for prepare mode."}, "approval_token": {"type": "string", "description": "Host-issued approval token required for execute/intent_execute mode."}, }, "required": ["market", "reserve", "amount_ui", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="kamino_earn_deposit", description=( "Preview, prepare, or execute a Kamino Earn vault deposit using a decimal token amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "kvault": {"type": "string", "description": "Kamino Earn vault address."}, "amount_ui": { "type": "string", "description": "Decimal token amount to deposit, as a string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "Prefer intent_preview then intent_execute after explicit chat confirmation: intent_execute re-derives the Kamino Earn transaction and executes within the approved parameters without round-tripping the preview payload. Legacy preview/prepare/execute remains supported.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds for intent_preview (1-300, default 120).", }, "purpose": {"type": "string", "description": "Short explanation of why the deposit is being made."}, "user_intent": {"type": "boolean", "description": "Must be true for prepare mode."}, "approval_token": {"type": "string", "description": "Host-issued approval token required for execute/intent_execute mode."}, }, "required": ["kvault", "amount_ui", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="kamino_earn_withdraw", description=( "Preview, prepare, or execute a Kamino Earn vault withdraw using a decimal token amount. " "Prepare returns an execution plan only, and execute requires a host-issued approval token bound to the previewed operation." ), input_schema={ "type": "object", "properties": { "kvault": {"type": "string", "description": "Kamino Earn vault address."}, "amount_ui": { "type": "string", "description": "Decimal token amount to withdraw, as a string.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute", "intent_preview", "intent_execute"], "description": "Prefer intent_preview then intent_execute after explicit chat confirmation: intent_execute re-derives the Kamino Earn transaction and executes within the approved parameters without round-tripping the preview payload. Legacy preview/prepare/execute remains supported.", }, "valid_for_seconds": { "type": "integer", "description": "Optional intent validity window in seconds for intent_preview (1-300, default 120).", }, "purpose": {"type": "string", "description": "Short explanation of why the withdraw is being made."}, "user_intent": {"type": "boolean", "description": "Must be true for prepare mode."}, "approval_token": {"type": "string", "description": "Host-issued approval token required for execute/intent_execute mode."}, }, "required": ["kvault", "amount_ui", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="close_empty_token_accounts", description=( "Preview or execute closing zero-balance SPL token accounts owned by the wallet. " "Use preview first, then execute only after explicit user approval." ), input_schema={ "type": "object", "properties": { "limit": { "type": "integer", "description": "Maximum number of empty token accounts to close in one transaction. Defaults to 8.", }, "mode": { "type": "string", "enum": ["preview", "execute"], "description": "preview lists closeable accounts, execute attempts to close them.", }, "purpose": { "type": "string", "description": "Short explanation of why the close operation is being made.", }, "user_intent": { "type": "boolean", "description": "Reserved for parity with other sensitive actions.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="medium", ) ) tools.append( AgentToolSpec( name="deactivate_solana_stake", description=( "Preview, prepare, or execute deactivation for a native Solana stake account." ), input_schema={ "type": "object", "properties": { "stake_account": { "type": "string", "description": "Stake account address to deactivate.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns a summary, prepare returns an execution plan without signed transaction bytes, execute attempts to send.", }, "purpose": { "type": "string", "description": "Short explanation of why the deactivation is being made.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["stake_account", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="withdraw_solana_stake", description=( "Preview, prepare, or execute withdrawal from a native Solana stake account." ), input_schema={ "type": "object", "properties": { "stake_account": { "type": "string", "description": "Stake account address to withdraw from.", }, "amount": { "type": "number", "description": "Amount of SOL to withdraw.", }, "recipient": { "type": "string", "description": "Optional destination wallet address. Defaults to the connected wallet.", }, "mode": { "type": "string", "enum": ["preview", "prepare", "execute"], "description": "preview returns a summary, prepare returns an execution plan without signed transaction bytes, execute attempts to send.", }, "purpose": { "type": "string", "description": "Short explanation of why the withdraw is being made.", }, "user_intent": { "type": "boolean", "description": "Must be true for prepare mode.", }, "approval_token": { "type": "string", "description": "Host-issued approval token required for execute mode.", }, }, "required": ["stake_account", "amount", "mode", "purpose"], "additionalProperties": False, }, read_only=False, requires_explicit_user_intent=True, risk_level="high", ) ) tools.append( AgentToolSpec( name="get_autonomous_session", description=( "Return the status of the current autonomous execution session, if any: " "whether it is active, its allow-lists, spend limits, operation count, and expiry. " "Read-only." ), input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ) ) tools.append( AgentToolSpec( name="start_autonomous_session", description=( "Start an autonomous execution session so subsequent wallet writes can execute " "WITHOUT a per-transaction human approval, bounded by the supplied limits. " "This grants standing authority, so it requires a host-issued approval_token bound " "to the exact session policy; an agent cannot start a session on its own. " "Preview the policy first (mode=preview) to obtain the confirmation summary the host signs." ), input_schema={ "type": "object", "properties": { "mode": { "type": "string", "enum": ["preview", "execute"], "description": "preview returns the policy summary to be confirmed; execute starts the session.", }, "allowed_tools": { "type": "array", "items": {"type": "string"}, "description": "Wallet tools the agent may execute autonomously.", }, "allowed_networks": { "type": "array", "items": {"type": "string"}, "description": "Networks the agent may operate on autonomously.", }, "allow_mainnet": { "type": "boolean", "description": "Must be true to permit autonomous execution on real-money networks.", }, "allowed_recipients": { "type": "array", "items": {"type": "string"}, "description": "Destination addresses the agent may send to.", }, "allow_any_recipient": { "type": "boolean", "description": "Allow destinations not on the allow-list (spend caps and simulation still apply).", }, "require_simulation": { "type": "boolean", "description": "Require a passing simulation before each autonomous execute (default true).", }, "max_per_tx_lamports": {"type": "integer", "description": "Per-transaction spend cap in smallest units (0 = unlimited)."}, "max_hourly_lamports": {"type": "integer", "description": "Hourly cumulative spend cap (0 = unlimited)."}, "max_daily_lamports": {"type": "integer", "description": "Daily cumulative spend cap (0 = unlimited)."}, "max_txs_per_minute": {"type": "integer", "description": "Transaction rate cap per minute (0 = unlimited)."}, "max_operations": {"type": "integer", "description": "Maximum operations the session may approve (0 = unlimited)."}, "session_ttl_seconds": {"type": "integer", "description": "Session lifetime in seconds (0 = no expiry)."}, "approval_ttl_seconds": {"type": "integer", "description": "TTL applied to each auto-issued approval token (default 120)."}, "approval_token": { "type": "string", "description": "Host-issued approval token bound to this session policy. Required for mode=execute.", }, }, "required": ["mode"], "additionalProperties": False, }, read_only=False, risk_level="high", ) ) tools.append( AgentToolSpec( name="stop_autonomous_session", description=( "End the current autonomous execution session. Always allowed; this only removes " "standing authority and returns the wallet to per-transaction human approval." ), input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=False, risk_level="low", ) ) tools.append( AgentToolSpec( name="agentlayer_autonomous_status", description=( "Return AgentLayer high-trust autonomous permission status. " "The autonomous permission group covers every wallet write tool (transfers, " "bridges, Solana swaps, staking, x402 payments, generic contract calls, Base " "swaps, and EVM DeFi management), not just base_swaps/defi_tools." ), input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ) ) tools.append( AgentToolSpec( name="agentlayer_autonomous_approve", description=( "Enable the high-trust autonomous permission group. The scope parameter is kept " "for compatibility; choosing base_swaps or defi_tools enables the full group, " "which covers every wallet write tool -- transfers, bridges, Solana swaps, " "staking, x402 payments, generic contract calls, Base swaps, and supported EVM " "DeFi management tools -- until revoked." ), input_schema={ "type": "object", "properties": { "scope": { "type": "string", "enum": ["all", "base_swaps", "defi_tools"], "description": "Scope to enable. Use \"all\" (recommended) -- base_swaps and defi_tools are deprecated aliases that enable the exact same full autonomous permission group, covering every wallet write tool.", }, "purpose": { "type": "string", "description": "Short explanation of why the standing permission is being enabled.", }, "user_intent": { "type": "boolean", "description": "Must be true after the user explicitly asks to enable this permission.", }, }, "required": ["scope", "purpose", "user_intent"], "additionalProperties": False, }, read_only=False, risk_level="high", ) ) tools.append( AgentToolSpec( name="agentlayer_autonomous_revoke", description=( "Disable the full high-trust autonomous permission group. " "The scope parameter is kept for compatibility; either value revokes all autonomous permissions." ), input_schema={ "type": "object", "properties": { "scope": { "type": "string", "enum": ["all", "base_swaps", "defi_tools"], "description": "Scope to revoke. Use \"all\" (recommended) -- base_swaps and defi_tools are deprecated aliases that revoke the exact same full autonomous permission group.", } }, "required": ["scope"], "additionalProperties": False, }, read_only=False, risk_level="low", ) ) tools.extend(self._x402_tool_specs()) return tools def get_runtime_instructions(self) -> str: """Return the instruction block to inject into the agent runtime.""" return WALLET_RUNTIME_INSTRUCTIONS @staticmethod def _autonomous_permission_tool_specs() -> list[AgentToolSpec]: return [ AgentToolSpec( name="agentlayer_autonomous_status", description=( "Return AgentLayer high-trust autonomous permission status. " "The autonomous permission group covers every wallet write tool (transfers, " "bridges, Solana swaps, staking, x402 payments, generic contract calls, Base " "swaps, and EVM DeFi management), not just base_swaps/defi_tools." ), input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, read_only=True, risk_level="low", ), AgentToolSpec( name="agentlayer_autonomous_approve", description=( "Enable the high-trust autonomous permission group. The scope parameter is kept " "for compatibility; choosing base_swaps or defi_tools enables the full group, " "which covers every wallet write tool -- transfers, bridges, Solana swaps, " "staking, x402 payments, generic contract calls, Base swaps, and supported EVM " "DeFi management tools -- until revoked." ), input_schema={ "type": "object", "properties": { "scope": { "type": "string", "enum": ["all", "base_swaps", "defi_tools"], "description": "Scope to enable. Use \"all\" (recommended) -- base_swaps and defi_tools are deprecated aliases that enable the exact same full autonomous permission group, covering every wallet write tool.", }, "purpose": { "type": "string", "description": "Short explanation of why the standing permission is being enabled.", }, "user_intent": { "type": "boolean", "description": "Must be true after the user explicitly asks to enable this permission.", }, }, "required": ["scope", "purpose", "user_intent"], "additionalProperties": False, }, read_only=False, risk_level="high", ), AgentToolSpec( name="agentlayer_autonomous_revoke", description=( "Disable the full high-trust autonomous permission group. " "The scope parameter is kept for compatibility; either value revokes all autonomous permissions." ), input_schema={ "type": "object", "properties": { "scope": { "type": "string", "enum": ["all", "base_swaps", "defi_tools"], "description": "Scope to revoke. Use \"all\" (recommended) -- base_swaps and defi_tools are deprecated aliases that revoke the exact same full autonomous permission group.", } }, "required": ["scope"], "additionalProperties": False, }, read_only=False, risk_level="low", ), ] async def invoke(self, tool_name: str, arguments: dict[str, Any] | None = None) -> AgentToolResult: """Dispatch an agent-facing tool call to the wallet backend.""" args = arguments or {} try: active_backend = self._resolve_backend_for_args(args) if tool_name == "get_autonomous_session": from agent_wallet import autonomous_session return AgentToolResult(tool=tool_name, ok=True, data=autonomous_session.session_status()) if tool_name == "stop_autonomous_session": from agent_wallet import autonomous_session return AgentToolResult(tool=tool_name, ok=True, data=autonomous_session.stop_session()) if tool_name == "agentlayer_autonomous_status": from agent_wallet import autonomous_permissions return AgentToolResult(tool=tool_name, ok=True, data=autonomous_permissions.status()) if tool_name == "agentlayer_autonomous_approve": from agent_wallet import autonomous_permissions scope = str(args.get("scope") or "").strip() purpose = args.get("purpose") if scope not in autonomous_permissions.SUPPORTED_SCOPES: raise WalletBackendError("scope must be one of: all, base_swaps, defi_tools.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if args.get("user_intent") is not True: raise WalletBackendError( "agentlayer_autonomous_approve requires user_intent=true after the user explicitly asks for this permission." ) data = autonomous_permissions.approve_all( approved_by="agentlayer_autonomous_approve" ) return AgentToolResult( tool=tool_name, ok=True, data=data, ) if tool_name == "agentlayer_autonomous_revoke": from agent_wallet import autonomous_permissions scope = str(args.get("scope") or "").strip() if scope not in autonomous_permissions.SUPPORTED_SCOPES: raise WalletBackendError("scope must be one of: all, base_swaps, defi_tools.") return AgentToolResult( tool=tool_name, ok=True, data=autonomous_permissions.revoke_all(), ) if tool_name == "start_autonomous_session": from agent_wallet import autonomous_session from agent_wallet.autonomous_session import config_to_dict from agent_wallet.nonce_registry import require_single_use config = self._build_session_config_from_args(args) policy_summary = {"operation": "start_autonomous_session", **config_to_dict(config)} mode = str(args.get("mode") or "").strip().lower() if mode not in {"preview", "execute"}: raise WalletBackendError("mode must be 'preview' or 'execute'.") if mode == "preview": return AgentToolResult( tool=tool_name, ok=True, data={ "mode": "preview", "confirmation_summary": policy_summary, "execute_requires_approval_token": True, "approval_hint": {"host_must_issue_token_for": policy_summary}, }, ) # execute: starting a session grants standing authority, so it # always requires a genuine host token — never the autonomous # fallback (an agent must not be able to widen its own envelope). approval_token = args.get("approval_token") if not isinstance(approval_token, str) or not approval_token.strip(): raise WalletBackendError( "start_autonomous_session execute requires a host-issued approval_token " "bound to the previewed session policy." ) verify_approval_token( approval_token.strip(), tool_name="start_autonomous_session", network=str(getattr(active_backend, "network", "unknown")), summary=policy_summary, require_mainnet_confirmation=bool(config.allow_mainnet), ) require_single_use(approval_token.strip()) status = autonomous_session.start_session(config) return AgentToolResult(tool=tool_name, ok=True, data=status) if tool_name == "x402_search_services": query = args.get("query") discovery_provider = args.get("discovery_provider", "auto") network = args.get("network") asset = args.get("asset") scheme = args.get("scheme") max_usd_price = args.get("max_usd_price") limit = args.get("limit", 10) for field_name, value in ( ("query", query), ("discovery_provider", discovery_provider), ("network", network), ("asset", asset), ("scheme", scheme), ("max_usd_price", max_usd_price), ): if value is not None and not isinstance(value, str): raise WalletBackendError(f"{field_name} must be a string when provided.") if not isinstance(limit, int) or limit <= 0: raise WalletBackendError("limit must be a positive integer.") data = await x402.search_services( query=query, discovery_provider=discovery_provider, network=network, asset=asset, scheme=scheme, max_usd_price=max_usd_price, limit=limit, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "x402_get_service_details": reference = args.get("reference") discovery_provider = args.get("discovery_provider", "auto") if not isinstance(reference, str) or not reference.strip(): raise WalletBackendError("reference is required.") if discovery_provider is not None and not isinstance(discovery_provider, str): raise WalletBackendError("discovery_provider must be a string when provided.") data = await x402.get_service_details( reference=reference.strip(), discovery_provider=discovery_provider, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "x402_preview_request": url = args.get("url") method = args.get("method", "GET") headers = args.get("headers") query = args.get("query") json_body = args.get("json_body") text_body = args.get("text_body") if not isinstance(url, str) or not url.strip(): raise WalletBackendError("url is required.") if method is not None and not isinstance(method, str): raise WalletBackendError("method must be a string when provided.") if headers is not None and not isinstance(headers, dict): raise WalletBackendError("headers must be an object when provided.") if query is not None and not isinstance(query, dict): raise WalletBackendError("query must be an object when provided.") if text_body is not None and not isinstance(text_body, str): raise WalletBackendError("text_body must be a string when provided.") data = await x402.preview_request( backend=active_backend, url=url.strip(), method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, ) if data.get("payment_required"): data = self._annotate_x402_payload(data, mode="preview") return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "x402_pay_request": url = args.get("url") method = args.get("method", "GET") headers = args.get("headers") query = args.get("query") json_body = args.get("json_body") text_body = args.get("text_body") purpose = args.get("purpose") if not isinstance(url, str) or not url.strip(): raise WalletBackendError("url is required.") if method is not None and not isinstance(method, str): raise WalletBackendError("method must be a string when provided.") if headers is not None and not isinstance(headers, dict): raise WalletBackendError("headers must be an object when provided.") if query is not None and not isinstance(query, dict): raise WalletBackendError("query must be an object when provided.") if text_body is not None and not isinstance(text_body, str): raise WalletBackendError("text_body must be a string when provided.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") approved_preview = args.get("_approved_preview") if approved_preview is not None and not isinstance(approved_preview, dict): raise WalletBackendError("_approved_preview must be an object when provided.") approval_token = args.get("approval_token") # Resolve the exact payment about to be made -- reusing # _approved_preview when it still matches this request (same # fingerprint check pay_and_fetch applies internally), or a # fresh probe otherwise -- so the summary bound into the # approval token below is never stale or re-priced. preview_payload = await x402.resolve_payment_preview( backend=active_backend, url=url.strip(), method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, approved_preview=approved_preview, ) de_minimis = False if preview_payload.get("payment_required"): de_minimis = x402.is_de_minimis_payment(preview_payload) if not de_minimis: summary = self._build_confirmation_summary( action_label="x402 paid request", payload=preview_payload, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=summary, action_label="x402 paid request", backend=active_backend, ) data = await x402.pay_and_fetch( backend=active_backend, url=url.strip(), method=method, headers=headers, query=query, json_body=json_body, text_body=text_body, approved_preview=preview_payload, ) data["purpose"] = purpose.strip() data = self._annotate_x402_payload(data, mode="execute") return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_wallet_capabilities": data = active_backend.get_capabilities().to_dict() data["network"] = str(getattr(active_backend, "network", "unknown")) data["address"] = await active_backend.get_address() data["is_mainnet"] = self._is_mainnet_for_backend(active_backend) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_wallet_address": address = await active_backend.get_address() return AgentToolResult( tool=tool_name, ok=True, data={ "address": address, "configured": bool(address), "network": str(getattr(active_backend, "network", "unknown")), "is_mainnet": self._is_mainnet_for_backend(active_backend), }, ) if tool_name == "get_wallet_balance": address = args.get("address") if address is not None and not isinstance(address, str): raise WalletBackendError("address must be a string when provided.") data = await active_backend.get_balance(address=address) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_lifi_supported_chains": data = await active_backend.get_lifi_supported_chains() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_lifi_quote": from_chain = args.get("from_chain") to_chain = args.get("to_chain") from_token = args.get("from_token") to_token = args.get("to_token") amount_in_raw = args.get("amount_in_raw") from_address = args.get("from_address") to_address = args.get("to_address") slippage = self._normalize_lifi_slippage(args.get("slippage")) allow_bridges = self._normalize_optional_string_list(args.get("allow_bridges"), field_name="allow_bridges") deny_bridges = self._normalize_optional_string_list(args.get("deny_bridges"), field_name="deny_bridges") prefer_bridges = self._normalize_optional_string_list(args.get("prefer_bridges"), field_name="prefer_bridges") if not isinstance(from_chain, str) or not from_chain.strip(): raise WalletBackendError("from_chain is required.") if not isinstance(to_chain, str) or not to_chain.strip(): raise WalletBackendError("to_chain is required.") if not isinstance(from_token, str) or not from_token.strip(): raise WalletBackendError("from_token is required.") if not isinstance(to_token, str) or not to_token.strip(): raise WalletBackendError("to_token is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") if from_address is not None and not isinstance(from_address, str): raise WalletBackendError("from_address must be a string when provided.") if to_address is not None and not isinstance(to_address, str): raise WalletBackendError("to_address must be a string when provided.") data = await active_backend.get_lifi_quote( from_chain=from_chain.strip(), to_chain=to_chain.strip(), from_token=from_token.strip(), to_token=to_token.strip(), amount_in_raw=amount_in_raw.strip(), from_address=from_address.strip() if isinstance(from_address, str) and from_address.strip() else None, to_address=to_address.strip() if isinstance(to_address, str) and to_address.strip() else None, slippage=slippage, allow_bridges=allow_bridges, deny_bridges=deny_bridges, prefer_bridges=prefer_bridges, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_lifi_transfer_status": tx_hash = args.get("tx_hash") bridge = args.get("bridge") from_chain = args.get("from_chain") to_chain = args.get("to_chain") if not isinstance(tx_hash, str) or not tx_hash.strip(): raise WalletBackendError("tx_hash is required.") if bridge is not None and not isinstance(bridge, str): raise WalletBackendError("bridge must be a string when provided.") if from_chain is not None and not isinstance(from_chain, str): raise WalletBackendError("from_chain must be a string when provided.") if to_chain is not None and not isinstance(to_chain, str): raise WalletBackendError("to_chain must be a string when provided.") data = await active_backend.get_lifi_transfer_status( tx_hash=tx_hash.strip(), bridge=bridge.strip() if isinstance(bridge, str) and bridge.strip() else None, from_chain=from_chain.strip() if isinstance(from_chain, str) and from_chain.strip() else None, to_chain=to_chain.strip() if isinstance(to_chain, str) and to_chain.strip() else None, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_btc_transfer_history": direction = args.get("direction", "all") limit = args.get("limit", 10) skip = args.get("skip", 0) if not isinstance(direction, str) or direction not in {"incoming", "outgoing", "all"}: raise WalletBackendError("direction must be 'incoming', 'outgoing', or 'all'.") if not isinstance(limit, int) or limit < 0: raise WalletBackendError("limit must be a non-negative integer.") if not isinstance(skip, int) or skip < 0: raise WalletBackendError("skip must be a non-negative integer.") data = await self.backend.get_btc_transfer_history( direction=direction, limit=limit, skip=skip, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_btc_fee_rates": data = await self.backend.get_btc_fee_rates() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_btc_max_spendable": fee_rate = args.get("fee_rate") if fee_rate is not None and (not isinstance(fee_rate, int) or fee_rate <= 0): raise WalletBackendError("fee_rate must be a positive integer when provided.") data = await self.backend.get_btc_max_spendable(fee_rate=fee_rate) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_network": data = await active_backend.get_evm_network_info() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "set_evm_network": requested_network = args.get("network") network = self._normalize_evm_tool_network(requested_network) self.backend = self.backend.with_network(network) data = await self.backend.get_evm_network_info() data["selected_network"] = network data["session_active_network"] = str(getattr(self.backend, "network", "unknown")) data["network_switch_persistent_for_runtime_session"] = True data["usage"] = ( "Subsequent EVM tool calls in this runtime session use this network by " "default. You can still override a single call with its network parameter." ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_token_balance": token_address = args.get("token_address") if not isinstance(token_address, str) or not token_address.strip(): raise WalletBackendError("token_address is required.") data = await active_backend.get_evm_token_balance(token_address.strip()) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_token_metadata": token_address = args.get("token_address") if not isinstance(token_address, str) or not token_address.strip(): raise WalletBackendError("token_address is required.") data = await active_backend.get_evm_token_metadata(token_address.strip()) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_fee_rates": data = await active_backend.get_evm_fee_rates() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_transaction_receipt": tx_hash = args.get("tx_hash") if not isinstance(tx_hash, str) or not tx_hash.strip(): raise WalletBackendError("tx_hash is required.") data = await active_backend.get_evm_transaction_receipt(tx_hash.strip()) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_aave_account": data = await active_backend.get_evm_aave_account() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_aave_reserves": data = await active_backend.get_evm_aave_reserves() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_aave_positions": data = await active_backend.get_evm_aave_positions() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "manage_evm_aave_position": operation = args.get("operation") token_address = args.get("token_address") amount_raw = args.get("amount_raw") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if operation not in {"supply", "withdraw", "borrow", "repay"}: raise WalletBackendError("operation must be one of: supply, withdraw, borrow, repay.") if not isinstance(token_address, str) or not token_address.strip(): raise WalletBackendError("token_address is required.") if not isinstance(amount_raw, str) or not amount_raw.strip().isdigit(): raise WalletBackendError("amount_raw must be a positive integer string.") if int(amount_raw.strip()) <= 0: raise WalletBackendError("amount_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "operation": str(operation), "token_address": token_address.strip(), "amount_raw": amount_raw.strip(), } if mode == "preview": preview = await active_backend.preview_evm_aave_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM Aave V3 operation", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_aave_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM Aave V3 operation", ), action_label="EVM Aave V3 operation", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_defi_permission( active_backend=active_backend, tool_name=tool_name, action_label="EVM Aave V3 operation", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_evm_aave_operation, ) expected_summary = { "operation": "EVM Aave V3 operation", "network": str(getattr(active_backend, "network", "unknown")), "aave_operation": str(operation), "token_address": token_address.strip(), "amount_raw": amount_raw.strip(), } for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM Aave V3 operation", backend=active_backend, ) result = await active_backend.send_evm_aave_operation( **preview_kwargs, expected_quote_fingerprint=( str(approval_summary_copy.get("quote_fingerprint")).strip() if approval_summary_copy.get("quote_fingerprint") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM Aave V3 operation", mode="execute", ), ) if tool_name == "get_evm_lido_overview": data = await active_backend.get_evm_lido_overview() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_lido_positions": data = await active_backend.get_evm_lido_positions() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "manage_evm_lido_position": operation = args.get("operation") amount_raw = args.get("amount_raw") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if operation not in {"stake_eth_for_wsteth", "wrap_steth", "unwrap_wsteth"}: raise WalletBackendError( "operation must be one of: stake_eth_for_wsteth, wrap_steth, unwrap_wsteth." ) if not isinstance(amount_raw, str) or not amount_raw.strip().isdigit(): raise WalletBackendError("amount_raw must be a positive integer string.") if int(amount_raw.strip()) <= 0: raise WalletBackendError("amount_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "operation": str(operation), "amount_raw": amount_raw.strip(), } if mode == "preview": preview = await active_backend.preview_evm_lido_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM Lido operation", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_lido_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM Lido operation", ), action_label="EVM Lido operation", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_defi_permission( active_backend=active_backend, tool_name=tool_name, action_label="EVM Lido operation", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_evm_lido_operation, ) expected_summary = { "operation": "EVM Lido operation", "network": str(getattr(active_backend, "network", "unknown")), "lido_operation": str(operation), "amount_raw": amount_raw.strip(), } for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM Lido operation", backend=active_backend, ) result = await active_backend.send_evm_lido_operation( **preview_kwargs, expected_quote_fingerprint=( str(approval_summary_copy.get("quote_fingerprint")).strip() if approval_summary_copy.get("quote_fingerprint") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM Lido operation", mode="execute", ), ) if tool_name == "get_evm_lido_withdrawal_requests": data = await active_backend.get_evm_lido_withdrawal_requests() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "manage_evm_lido_withdrawal": operation = args.get("operation") amount_raw = args.get("amount_raw") request_id = args.get("request_id") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if operation not in { "request_withdrawal_steth", "request_withdrawal_wsteth", "claim_withdrawal", }: raise WalletBackendError( "operation must be one of: request_withdrawal_steth, request_withdrawal_wsteth, claim_withdrawal." ) if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if operation == "claim_withdrawal": if not isinstance(request_id, str) or not request_id.strip().isdigit(): raise WalletBackendError("request_id must be a positive integer string.") if int(request_id.strip()) <= 0: raise WalletBackendError("request_id must be greater than zero.") else: if not isinstance(amount_raw, str) or not amount_raw.strip().isdigit(): raise WalletBackendError("amount_raw must be a positive integer string.") if int(amount_raw.strip()) <= 0: raise WalletBackendError("amount_raw must be greater than zero.") preview_kwargs = { "operation": str(operation), **( {"amount_raw": amount_raw.strip()} if isinstance(amount_raw, str) and amount_raw.strip() else {} ), **( {"request_id": request_id.strip()} if isinstance(request_id, str) and request_id.strip() else {} ), } if mode == "preview": preview = await active_backend.preview_evm_lido_withdrawal(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM Lido withdrawal", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_lido_withdrawal(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM Lido withdrawal", ), action_label="EVM Lido withdrawal", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_defi_permission( active_backend=active_backend, tool_name=tool_name, action_label="EVM Lido withdrawal", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_evm_lido_withdrawal, ) expected_summary = { "operation": "EVM Lido withdrawal", "network": str(getattr(active_backend, "network", "unknown")), "lido_withdrawal_operation": str(operation), } if operation == "claim_withdrawal": expected_summary["request_id"] = request_id.strip() else: expected_summary["amount_raw"] = amount_raw.strip() for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM Lido withdrawal", backend=active_backend, ) result = await active_backend.send_evm_lido_withdrawal( **preview_kwargs, expected_quote_fingerprint=( str(approval_summary_copy.get("quote_fingerprint")).strip() if approval_summary_copy.get("quote_fingerprint") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM Lido withdrawal", mode="execute", ), ) if tool_name == "get_evm_morpho_vaults": data = await active_backend.get_evm_morpho_vaults( vault_address=( str(args.get("vault_address")).strip() if isinstance(args.get("vault_address"), str) and args.get("vault_address").strip() else None ), limit=int(args.get("limit")) if args.get("limit") is not None else None, listed_only=bool(args.get("listed_only", True)), asset_address=( str(args.get("asset_address")).strip() if isinstance(args.get("asset_address"), str) and args.get("asset_address").strip() else None ), min_tvl_usd=self._normalize_optional_non_negative_number( args.get("min_tvl_usd"), field_name="min_tvl_usd" ), min_net_apy=self._normalize_optional_non_negative_number( args.get("min_net_apy"), field_name="min_net_apy" ), order_by=( str(args.get("order_by")).strip() if isinstance(args.get("order_by"), str) and args.get("order_by").strip() else None ), order_direction=( str(args.get("order_direction")).strip() if isinstance(args.get("order_direction"), str) and args.get("order_direction").strip() else None ), ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_morpho_markets": data = await active_backend.get_evm_morpho_markets( market_id=( str(args.get("market_id")).strip() if isinstance(args.get("market_id"), str) and args.get("market_id").strip() else None ), limit=int(args.get("limit")) if args.get("limit") is not None else None, listed_only=bool(args.get("listed_only", True)), search=( str(args.get("search")).strip() if isinstance(args.get("search"), str) and args.get("search").strip() else None ), collateral_asset_address=( str(args.get("collateral_asset_address")).strip() if isinstance(args.get("collateral_asset_address"), str) and args.get("collateral_asset_address").strip() else None ), loan_asset_address=( str(args.get("loan_asset_address")).strip() if isinstance(args.get("loan_asset_address"), str) and args.get("loan_asset_address").strip() else None ), min_supply_usd=self._normalize_optional_non_negative_number( args.get("min_supply_usd"), field_name="min_supply_usd" ), min_net_supply_apy=self._normalize_optional_non_negative_number( args.get("min_net_supply_apy"), field_name="min_net_supply_apy" ), order_by=( str(args.get("order_by")).strip() if isinstance(args.get("order_by"), str) and args.get("order_by").strip() else None ), order_direction=( str(args.get("order_direction")).strip() if isinstance(args.get("order_direction"), str) and args.get("order_direction").strip() else None ), ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_evm_morpho_positions": data = await active_backend.get_evm_morpho_positions() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "manage_evm_morpho_vault_position": operation = args.get("operation") token_address = args.get("token_address") vault_address = args.get("vault_address") vault_preset = args.get("vault_preset") amount_raw = args.get("amount_raw") native_amount_raw = args.get("native_amount_raw") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if operation not in {"supply", "withdraw"}: raise WalletBackendError("operation must be one of: supply, withdraw.") if not isinstance(token_address, str) or not token_address.strip(): raise WalletBackendError("token_address is required.") if bool(vault_address) == bool(vault_preset): raise WalletBackendError("Provide exactly one of vault_address or vault_preset.") if operation == "supply": has_amount = isinstance(amount_raw, str) and amount_raw.strip().isdigit() and int(amount_raw.strip()) > 0 has_native_amount = ( isinstance(native_amount_raw, str) and native_amount_raw.strip().isdigit() and int(native_amount_raw.strip()) > 0 ) if not has_amount and not has_native_amount: raise WalletBackendError( "amount_raw or native_amount_raw must be a positive integer string for Morpho vault supply." ) else: if not isinstance(amount_raw, str) or not amount_raw.strip().isdigit(): raise WalletBackendError("amount_raw must be a positive integer string.") if int(amount_raw.strip()) <= 0: raise WalletBackendError("amount_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "operation": str(operation), "token_address": token_address.strip(), **({"vault_address": vault_address.strip()} if isinstance(vault_address, str) and vault_address.strip() else {}), **({"vault_preset": vault_preset.strip()} if isinstance(vault_preset, str) and vault_preset.strip() else {}), **({"amount_raw": amount_raw.strip()} if isinstance(amount_raw, str) and amount_raw.strip() else {}), **( {"native_amount_raw": native_amount_raw.strip()} if isinstance(native_amount_raw, str) and native_amount_raw.strip() else {} ), } if mode == "preview": preview = await active_backend.preview_evm_morpho_vault_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM Morpho vault operation", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_morpho_vault_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM Morpho vault operation", ), action_label="EVM Morpho vault operation", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_defi_permission( active_backend=active_backend, tool_name=tool_name, action_label="EVM Morpho vault operation", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_evm_morpho_vault_operation, ) expected_summary = { "operation": "EVM Morpho vault operation", "network": str(getattr(active_backend, "network", "unknown")), "morpho_surface": "vault", "morpho_operation": str(operation), "token_address": token_address.strip(), } if isinstance(amount_raw, str) and amount_raw.strip(): expected_summary["amount_raw"] = amount_raw.strip() if isinstance(native_amount_raw, str) and native_amount_raw.strip(): expected_summary["native_amount_raw"] = native_amount_raw.strip() target = approval_summary_copy.get("target") if not isinstance(target, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) expected_target_key = "vaultAddress" if vault_address else "vaultPreset" expected_target_value = vault_address.strip() if vault_address else vault_preset.strip() for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) actual_target_value = target.get(expected_target_key) if vault_address: # The runtime lowercases the resolved vault address in the # approval binding, so compare addresses case-insensitively. target_matches = ( isinstance(actual_target_value, str) and actual_target_value.lower() == expected_target_value.lower() ) else: target_matches = actual_target_value == expected_target_value if not target_matches: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM Morpho vault operation", backend=active_backend, ) result = await active_backend.send_evm_morpho_vault_operation( **preview_kwargs, expected_quote_fingerprint=( str(approval_summary_copy.get("quote_fingerprint")).strip() if approval_summary_copy.get("quote_fingerprint") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM Morpho vault operation", mode="execute", ), ) if tool_name == "manage_evm_morpho_market_position": operation = args.get("operation") token_address = args.get("token_address") market_id = args.get("market_id") market_preset = args.get("market_preset") amount_raw = args.get("amount_raw") native_amount_raw = args.get("native_amount_raw") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if operation not in {"supply_collateral", "borrow", "repay", "withdraw_collateral"}: raise WalletBackendError( "operation must be one of: supply_collateral, borrow, repay, withdraw_collateral." ) if not isinstance(token_address, str) or not token_address.strip(): raise WalletBackendError("token_address is required.") if bool(market_id) == bool(market_preset): raise WalletBackendError("Provide exactly one of market_id or market_preset.") if operation == "supply_collateral": has_amount = isinstance(amount_raw, str) and amount_raw.strip().isdigit() and int(amount_raw.strip()) > 0 has_native_amount = ( isinstance(native_amount_raw, str) and native_amount_raw.strip().isdigit() and int(native_amount_raw.strip()) > 0 ) if not has_amount and not has_native_amount: raise WalletBackendError( "amount_raw or native_amount_raw must be a positive integer string for Morpho supply_collateral." ) elif operation == "repay": normalized_amount = str(amount_raw or "").strip().lower() if normalized_amount != "max": if not normalized_amount.isdigit() or int(normalized_amount) <= 0: raise WalletBackendError( "amount_raw must be a positive integer string or the literal max for Morpho repay." ) else: if not isinstance(amount_raw, str) or not amount_raw.strip().isdigit(): raise WalletBackendError("amount_raw must be a positive integer string.") if int(amount_raw.strip()) <= 0: raise WalletBackendError("amount_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "operation": str(operation), "token_address": token_address.strip(), **({"market_id": market_id.strip()} if isinstance(market_id, str) and market_id.strip() else {}), **({"market_preset": market_preset.strip()} if isinstance(market_preset, str) and market_preset.strip() else {}), **({"amount_raw": amount_raw.strip()} if isinstance(amount_raw, str) and amount_raw.strip() else {}), **( {"native_amount_raw": native_amount_raw.strip()} if isinstance(native_amount_raw, str) and native_amount_raw.strip() else {} ), } if mode == "preview": preview = await active_backend.preview_evm_morpho_market_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM Morpho market operation", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_morpho_market_operation(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM Morpho market operation", ), action_label="EVM Morpho market operation", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_defi_permission( active_backend=active_backend, tool_name=tool_name, action_label="EVM Morpho market operation", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_evm_morpho_market_operation, ) expected_summary = { "operation": "EVM Morpho market operation", "network": str(getattr(active_backend, "network", "unknown")), "morpho_surface": "market", "morpho_operation": str(operation), "token_address": token_address.strip(), } if isinstance(amount_raw, str) and amount_raw.strip(): expected_summary["amount_raw"] = amount_raw.strip() if isinstance(native_amount_raw, str) and native_amount_raw.strip(): expected_summary["native_amount_raw"] = native_amount_raw.strip() target = approval_summary_copy.get("target") if not isinstance(target, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) expected_target_key = "marketId" if market_id else "marketPreset" expected_target_value = market_id.strip() if market_id else market_preset.strip() for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) actual_target_value = target.get(expected_target_key) if market_id: # Market ids are hex hashes; the runtime may normalize their # case in the approval binding, so compare case-insensitively. target_matches = ( isinstance(actual_target_value, str) and actual_target_value.lower() == expected_target_value.lower() ) else: target_matches = actual_target_value == expected_target_value if not target_matches: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM Morpho market operation", backend=active_backend, ) result = await active_backend.send_evm_morpho_market_operation( **preview_kwargs, expected_quote_fingerprint=( str(approval_summary_copy.get("quote_fingerprint")).strip() if approval_summary_copy.get("quote_fingerprint") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM Morpho market operation", mode="execute", ), ) if tool_name == "get_evm_swap_quote": token_in = args.get("token_in") token_out = args.get("token_out") amount_in_raw = args.get("amount_in_raw") if not isinstance(token_in, str) or not token_in.strip(): raise WalletBackendError("token_in is required.") if not isinstance(token_out, str) or not token_out.strip(): raise WalletBackendError("token_out is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") data = await active_backend.get_evm_swap_quote( token_in=self._canonicalize_velora_token_identifier(token_in), token_out=self._canonicalize_velora_token_identifier(token_out), amount_in_raw=amount_in_raw.strip(), ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "swap_evm_tokens": token_in = args.get("token_in") token_out = args.get("token_out") amount_in_raw = args.get("amount_in_raw") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(token_in, str) or not token_in.strip(): raise WalletBackendError("token_in is required.") if not isinstance(token_out, str) or not token_out.strip(): raise WalletBackendError("token_out is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "token_in": self._canonicalize_velora_token_identifier(token_in), "token_out": self._canonicalize_velora_token_identifier(token_out), "amount_in_raw": amount_in_raw.strip(), } if mode == "preview": preview = await active_backend.preview_evm_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM swap", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM swap", ), action_label="EVM swap", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_base_swap_permission( active_backend=active_backend, tool_name=tool_name, action_label="EVM swap", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_evm_swap, ) expected_summary = { "operation": "EVM swap", "network": str(getattr(active_backend, "network", "unknown")), "token_in": preview_kwargs["token_in"], "token_out": preview_kwargs["token_out"], "input_amount_raw": amount_in_raw.strip(), } for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM swap", backend=active_backend, ) bound_quote_fingerprint = approval_summary_copy.get("quote_fingerprint") bound_minimum_output_amount_raw = approval_summary_copy.get("minimum_output_amount_raw") if isinstance(bound_quote_fingerprint, str) and bound_quote_fingerprint.strip(): result = await active_backend.send_evm_swap( **preview_kwargs, expected_quote_fingerprint=bound_quote_fingerprint.strip(), minimum_output_amount_raw=( str(bound_minimum_output_amount_raw).strip() if bound_minimum_output_amount_raw is not None else None ), ) else: result = await active_backend.send_evm_swap( **preview_kwargs, minimum_output_amount_raw=( str(bound_minimum_output_amount_raw).strip() if bound_minimum_output_amount_raw is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM swap", mode="execute", ), ) if tool_name == "swap_evm_lifi_cross_chain_tokens": token_in = args.get("token_in") destination_chain = args.get("destination_chain") output_token = args.get("output_token") destination_address = args.get("destination_address") amount_in_raw = args.get("amount_in_raw") slippage = self._normalize_lifi_slippage(args.get("slippage")) allow_bridges = self._normalize_optional_string_list(args.get("allow_bridges"), field_name="allow_bridges") deny_bridges = self._normalize_optional_string_list(args.get("deny_bridges"), field_name="deny_bridges") prefer_bridges = self._normalize_optional_string_list(args.get("prefer_bridges"), field_name="prefer_bridges") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(token_in, str) or not token_in.strip(): raise WalletBackendError("token_in is required.") if not isinstance(destination_chain, str) or not destination_chain.strip(): raise WalletBackendError("destination_chain is required.") if not isinstance(output_token, str) or not output_token.strip(): raise WalletBackendError("output_token is required.") if not isinstance(destination_address, str) or not destination_address.strip(): raise WalletBackendError("destination_address is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "token_in": token_in.strip(), "destination_chain": destination_chain.strip(), "output_token": output_token.strip(), "destination_address": destination_address.strip(), "amount_in_raw": amount_in_raw.strip(), "slippage": slippage, "allow_bridges": allow_bridges, "deny_bridges": deny_bridges, "prefer_bridges": prefer_bridges, } if mode == "preview": preview = await active_backend.preview_evm_lifi_cross_chain_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM LI.FI cross-chain swap", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_lifi_cross_chain_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM LI.FI cross-chain swap", ), action_label="EVM LI.FI cross-chain swap", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) source_chain_id = self._canonicalize_lifi_chain_identifier( getattr(active_backend, "network", "unknown") ) destination_chain_id = self._canonicalize_lifi_chain_identifier(destination_chain) expected_summary = { "operation": "EVM LI.FI cross-chain swap", "network": str(getattr(active_backend, "network", "unknown")), "source_chain": source_chain_id, "destination_chain": destination_chain_id, "token_in": self._canonicalize_lifi_token_identifier( token_in, chain_id=source_chain_id, ), "output_token": self._canonicalize_lifi_token_identifier( output_token, chain_id=destination_chain_id, ), "destination_address": destination_address.strip(), "input_amount_raw": amount_in_raw.strip(), } for key, expected_value in expected_summary.items(): actual_value = approval_summary.get(key) if key in {"source_chain", "destination_chain"}: actual_value = self._canonicalize_lifi_chain_identifier(actual_value) if key == "token_in": actual_value = self._canonicalize_lifi_token_identifier( actual_value, chain_id=source_chain_id, ) if key == "output_token": actual_value = self._canonicalize_lifi_token_identifier( actual_value, chain_id=destination_chain_id, ) if actual_value != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: # No host token: build the exact preview a human would # have seen and let _require_execute_approval's # autonomous_session / agentlayer_autonomous_approve # fallback authorize it. fresh_preview = await active_backend.preview_evm_lifi_cross_chain_swap(**preview_kwargs) approval_summary_copy = self._build_confirmation_summary( action_label="EVM LI.FI cross-chain swap", payload=fresh_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="EVM LI.FI cross-chain swap", backend=active_backend, ) result = await active_backend.send_evm_lifi_cross_chain_swap( token_in=str(approval_summary_copy.get("token_in") or token_in).strip(), destination_chain=str( approval_summary_copy.get("destination_chain") or destination_chain ).strip(), output_token=str(approval_summary_copy.get("output_token") or output_token).strip(), destination_address=str( approval_summary_copy.get("destination_address") or destination_address ).strip(), amount_in_raw=str(approval_summary_copy.get("input_amount_raw") or amount_in_raw).strip(), slippage=approval_summary_copy.get("slippage", slippage), allow_bridges=allow_bridges, deny_bridges=deny_bridges, prefer_bridges=prefer_bridges, minimum_output_amount_raw=( str(approval_summary_copy.get("minimum_output_amount_raw")).strip() if approval_summary_copy.get("minimum_output_amount_raw") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM LI.FI cross-chain swap", mode="execute", ), ) if tool_name == "get_uniswap_swap_quote": token_in = args.get("token_in") token_out = args.get("token_out") amount_in_raw = args.get("amount_in_raw") slippage_bps = args.get("slippage_bps") if not isinstance(token_in, str) or not token_in.strip(): raise WalletBackendError("token_in is required.") if not isinstance(token_out, str) or not token_out.strip(): raise WalletBackendError("token_out is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") if slippage_bps is not None and (not isinstance(slippage_bps, int) or slippage_bps < 0 or slippage_bps > 5000): raise WalletBackendError("slippage_bps must be an integer between 0 and 5000.") data = await active_backend.get_uniswap_swap_quote( token_in=token_in.strip(), token_out=token_out.strip(), amount_in_raw=amount_in_raw.strip(), slippage_bps=slippage_bps, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "search_uniswap_pairs": query = args.get("query") token_address = args.get("token_address") chain = args.get("chain") dex_id = args.get("dex_id") all_chains = bool(args.get("all_chains", False)) limit = args.get("limit", 10) if query is not None and not isinstance(query, str): raise WalletBackendError("query must be a string.") if token_address is not None and not isinstance(token_address, str): raise WalletBackendError("token_address must be a string.") if not (isinstance(query, str) and query.strip()) and not ( isinstance(token_address, str) and token_address.strip() ): raise WalletBackendError("Either query or token_address is required.") if chain is not None and not isinstance(chain, str): raise WalletBackendError("chain must be a string.") if dex_id is not None and not isinstance(dex_id, str): raise WalletBackendError("dex_id must be a string.") if not isinstance(limit, int) or limit < 1 or limit > 30: raise WalletBackendError("limit must be an integer between 1 and 30.") data = await active_backend.search_uniswap_pairs( query=query, token_address=token_address, chain=chain, dex_id=dex_id, all_chains=all_chains, limit=limit, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "swap_evm_uniswap_tokens": token_in = args.get("token_in") token_out = args.get("token_out") amount_in_raw = args.get("amount_in_raw") slippage_bps = args.get("slippage_bps") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(token_in, str) or not token_in.strip(): raise WalletBackendError("token_in is required.") if not isinstance(token_out, str) or not token_out.strip(): raise WalletBackendError("token_out is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if slippage_bps is not None and (not isinstance(slippage_bps, int) or slippage_bps < 0 or slippage_bps > 5000): raise WalletBackendError("slippage_bps must be an integer between 0 and 5000.") preview_kwargs = { "token_in": token_in.strip(), "token_out": token_out.strip(), "amount_in_raw": amount_in_raw.strip(), "slippage_bps": slippage_bps, } if mode == "preview": preview = await active_backend.preview_uniswap_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Uniswap swap", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_uniswap_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Uniswap swap", ), action_label="Uniswap swap", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: approval_token, approval_summary_copy = await self._authorize_base_swap_permission( active_backend=active_backend, tool_name=tool_name, action_label="Uniswap swap", preview_kwargs=preview_kwargs, preview_method=active_backend.preview_uniswap_swap, ) expected_summary = { "operation": "Uniswap swap", "network": str(getattr(active_backend, "network", "unknown")), "token_in": token_in.strip(), "token_out": token_out.strip(), "input_amount_raw": amount_in_raw.strip(), } for key, expected_value in expected_summary.items(): if approval_summary_copy.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="Uniswap swap", backend=active_backend, ) bound_quote_fingerprint = approval_summary_copy.get("quote_fingerprint") bound_minimum_output_amount_raw = approval_summary_copy.get("minimum_output_amount_raw") result = await active_backend.send_uniswap_swap( **preview_kwargs, expected_quote_fingerprint=( bound_quote_fingerprint.strip() if isinstance(bound_quote_fingerprint, str) and bound_quote_fingerprint.strip() else None ), minimum_output_amount_raw=( str(bound_minimum_output_amount_raw).strip() if bound_minimum_output_amount_raw is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Uniswap swap", mode="execute", ), ) if tool_name == "issue_wallet_approval": from agent_wallet.approval import issue_approval_token target_tool = args.get("tool_name") summary = args.get("summary") mainnet_confirmed = args.get("mainnet_confirmed", False) if not isinstance(target_tool, str) or not target_tool.strip(): raise WalletBackendError("tool_name is required.") if not isinstance(summary, dict) or not summary: raise WalletBackendError("summary must be a non-empty object matching the prepare response confirmation_summary.") if not isinstance(mainnet_confirmed, bool): raise WalletBackendError("mainnet_confirmed must be a boolean.") if self._is_mainnet_for_backend(active_backend) and not mainnet_confirmed: raise WalletBackendError( "mainnet_confirmed must be true for mainnet operations. " "Set mainnet_confirmed: true to confirm you accept the risk of an irreversible mainnet transaction." ) network = str(getattr(active_backend, "network", "unknown")) token = issue_approval_token( tool_name=target_tool.strip(), network=network, summary=summary, mainnet_confirmed=mainnet_confirmed, issued_by="claude-code-bridge", ) return AgentToolResult( tool=tool_name, ok=True, data={ "approval_token": token, "tool_name": target_tool.strip(), "network": network, "mainnet_confirmed": mainnet_confirmed, "usage": f"Pass approval_token to {target_tool.strip()} with mode='execute'.", }, ) if tool_name == "get_wallet_portfolio": address = args.get("address") if address is not None and not isinstance(address, str): raise WalletBackendError("address must be a string when provided.") data = await self.backend.get_portfolio(address=address) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_solana_token_prices": mints = args.get("mints") if not isinstance(mints, list) or not mints: raise WalletBackendError("mints must be a non-empty array of strings.") if not all(isinstance(item, str) for item in mints): raise WalletBackendError("Each mint must be a string.") data = await self.backend.get_token_prices(mints=mints) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_solana_staking_validators": limit = args.get("limit", 20) include_delinquent = args.get("include_delinquent", False) if not isinstance(limit, int) or limit <= 0: raise WalletBackendError("limit must be a positive integer.") if not isinstance(include_delinquent, bool): raise WalletBackendError("include_delinquent must be a boolean.") data = await self.backend.get_staking_validators( limit=limit, include_delinquent=include_delinquent, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_solana_stake_account": stake_account = args.get("stake_account") if not isinstance(stake_account, str) or not stake_account.strip(): raise WalletBackendError("stake_account is required.") data = await self.backend.get_stake_account(stake_account.strip()) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_flash_trade_markets": pool_name = args.get("pool_name") if pool_name is not None and not isinstance(pool_name, str): raise WalletBackendError("pool_name must be a string when provided.") data = await active_backend.get_flash_trade_markets(pool_name=pool_name) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_flash_trade_positions": owner = args.get("owner") pool_name = args.get("pool_name") if owner is not None and not isinstance(owner, str): raise WalletBackendError("owner must be a string when provided.") if pool_name is not None and not isinstance(pool_name, str): raise WalletBackendError("pool_name must be a string when provided.") data = await active_backend.get_flash_trade_positions( owner=owner, pool_name=pool_name, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "flash_trade_open_position": pool_name = args.get("pool_name") market_symbol = args.get("market_symbol") collateral_symbol = args.get("collateral_symbol") collateral_amount_raw = args.get("collateral_amount_raw") leverage = args.get("leverage") side = args.get("side") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(pool_name, str) or not pool_name.strip(): raise WalletBackendError("pool_name is required.") if not isinstance(market_symbol, str) or not market_symbol.strip(): raise WalletBackendError("market_symbol is required.") if not isinstance(collateral_symbol, str) or not collateral_symbol.strip(): raise WalletBackendError("collateral_symbol is required.") if not isinstance(collateral_amount_raw, str) or not collateral_amount_raw.strip(): raise WalletBackendError("collateral_amount_raw is required.") if not isinstance(leverage, str) or not leverage.strip(): raise WalletBackendError("leverage is required.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await active_backend.preview_flash_trade_open_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), collateral_symbol=collateral_symbol.strip(), collateral_amount_raw=collateral_amount_raw.strip(), leverage=leverage.strip(), side=side, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Flash Trade open position", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) prepared = await active_backend.prepare_flash_trade_open_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), collateral_symbol=collateral_symbol.strip(), collateral_amount_raw=collateral_amount_raw.strip(), leverage=leverage.strip(), side=side, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=prepared, action_label="Flash Trade open position", ), action_label="Flash Trade open position", mode="prepare", ), ) execute_preview = None if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) expected_summary = { "operation": "Flash Trade open position", "pool_name": pool_name.strip(), "market_symbol": market_symbol.strip(), "collateral_symbol": collateral_symbol.strip(), "collateral_amount_raw": collateral_amount_raw.strip(), "leverage": leverage.strip(), "side": side, } for key, expected_value in expected_summary.items(): if approval_summary.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) approved_preview = args.get("_approved_preview") if isinstance(approval_summary_copy.get("_preview_digest"), str): if not isinstance(approved_preview, dict): raise WalletBackendError( "Approved Flash Trade preview payload is required for execute mode. Generate a new preview and approval before execute." ) if preview_payload_digest(approved_preview) != approval_summary_copy["_preview_digest"]: raise WalletBackendError( "approved preview payload does not match the approval token. Generate a new preview and approval before execute." ) execute_preview = dict(approved_preview) else: execute_preview = await active_backend.preview_flash_trade_open_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), collateral_symbol=collateral_symbol.strip(), collateral_amount_raw=collateral_amount_raw.strip(), leverage=leverage.strip(), side=side, ) else: # No host token: build the same identity summary the # human path validates, and let _require_execute_approval's # autonomous_session / agentlayer_autonomous_approve # fallback authorize it. approval_summary_copy = { "operation": "Flash Trade open position", "pool_name": pool_name.strip(), "market_symbol": market_symbol.strip(), "collateral_symbol": collateral_symbol.strip(), "collateral_amount_raw": collateral_amount_raw.strip(), "leverage": leverage.strip(), "side": side, } execute_preview = await active_backend.preview_flash_trade_open_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), collateral_symbol=collateral_symbol.strip(), collateral_amount_raw=collateral_amount_raw.strip(), leverage=leverage.strip(), side=side, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="Flash Trade open position", backend=active_backend, ) result = await active_backend.execute_flash_trade_open_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), collateral_symbol=collateral_symbol.strip(), collateral_amount_raw=collateral_amount_raw.strip(), leverage=leverage.strip(), side=side, approved_preview=execute_preview, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Flash Trade open position", mode="execute", ), ) if tool_name == "flash_trade_close_position": pool_name = args.get("pool_name") market_symbol = args.get("market_symbol") side = args.get("side") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(pool_name, str) or not pool_name.strip(): raise WalletBackendError("pool_name is required.") if not isinstance(market_symbol, str) or not market_symbol.strip(): raise WalletBackendError("market_symbol is required.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await active_backend.preview_flash_trade_close_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), side=side, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Flash Trade close position", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) prepared = await active_backend.prepare_flash_trade_close_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), side=side, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=prepared, action_label="Flash Trade close position", ), action_label="Flash Trade close position", mode="prepare", ), ) execute_preview = None if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(active_backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(active_backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) expected_summary = { "operation": "Flash Trade close position", "pool_name": pool_name.strip(), "market_symbol": market_symbol.strip(), "side": side, } for key, expected_value in expected_summary.items(): if approval_summary.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) approved_preview = args.get("_approved_preview") if isinstance(approval_summary_copy.get("_preview_digest"), str): if not isinstance(approved_preview, dict): raise WalletBackendError( "Approved Flash Trade preview payload is required for execute mode. Generate a new preview and approval before execute." ) if preview_payload_digest(approved_preview) != approval_summary_copy["_preview_digest"]: raise WalletBackendError( "approved preview payload does not match the approval token. Generate a new preview and approval before execute." ) execute_preview = dict(approved_preview) else: execute_preview = await active_backend.preview_flash_trade_close_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), side=side, ) else: # No host token: build the same identity summary the # human path validates, and let _require_execute_approval's # autonomous_session / agentlayer_autonomous_approve # fallback authorize it. approval_summary_copy = { "operation": "Flash Trade close position", "pool_name": pool_name.strip(), "market_symbol": market_symbol.strip(), "side": side, } execute_preview = await active_backend.preview_flash_trade_close_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), side=side, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="Flash Trade close position", backend=active_backend, ) result = await active_backend.execute_flash_trade_close_position( pool_name=pool_name.strip(), market_symbol=market_symbol.strip(), side=side, approved_preview=execute_preview, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Flash Trade close position", mode="execute", ), ) if tool_name == "get_kamino_lend_markets": data = await self.backend.get_kamino_lend_markets() return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_portfolio": user = args.get("user") if user is not None and not isinstance(user, str): raise WalletBackendError("user must be a string when provided.") data = await self.backend.get_kamino_portfolio(user=user) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_vaults": vault_address = args.get("vault_address") if vault_address is not None and not isinstance(vault_address, str): raise WalletBackendError("vault_address must be a string when provided.") token_mint = args.get("token_mint") if token_mint is not None and not isinstance(token_mint, str): raise WalletBackendError("token_mint must be a string when provided.") limit = args.get("limit") if limit is not None and (not isinstance(limit, int) or isinstance(limit, bool)): raise WalletBackendError("limit must be an integer when provided.") data = await self.backend.get_kamino_vaults( vault_address=vault_address, token_mint=token_mint, include_metrics=bool(args.get("include_metrics", False)), limit=limit, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_earn_positions": user = args.get("user") if user is not None and not isinstance(user, str): raise WalletBackendError("user must be a string when provided.") data = await self.backend.get_kamino_earn_positions(user=user) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_liquidity_positions": user = args.get("user") if user is not None and not isinstance(user, str): raise WalletBackendError("user must be a string when provided.") data = await self.backend.get_kamino_liquidity_positions(user=user) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_lend_market_reserves": market = args.get("market") if not isinstance(market, str) or not market.strip(): raise WalletBackendError("market is required.") data = await self.backend.get_kamino_lend_market_reserves(market=market.strip()) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_lend_user_obligations": market = args.get("market") user = args.get("user") if not isinstance(market, str) or not market.strip(): raise WalletBackendError("market is required.") if user is not None and not isinstance(user, str): raise WalletBackendError("user must be a string when provided.") data = await self.backend.get_kamino_lend_user_obligations( market=market.strip(), user=user, ) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_lend_user_rewards": user = args.get("user") if user is not None and not isinstance(user, str): raise WalletBackendError("user must be a string when provided.") data = await self.backend.get_kamino_lend_user_rewards(user=user) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "get_kamino_open_positions": user = args.get("user") if user is not None and not isinstance(user, str): raise WalletBackendError("user must be a string when provided.") data = await self.backend.get_kamino_open_positions(user=user) return AgentToolResult(tool=tool_name, ok=True, data=data) if tool_name == "sign_wallet_message": user_confirmed = args.get("user_confirmed") if user_confirmed is not True: raise WalletBackendError( "Message signing requires explicit user confirmation." ) message = args.get("message") purpose = args.get("purpose") if not isinstance(message, str) or not message.strip(): raise WalletBackendError("message is required.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") signature = await self.backend.sign_message(message.strip()) return AgentToolResult( tool=tool_name, ok=True, data={ "signature": signature, "purpose": purpose.strip(), "message": message.strip(), "sign_only": self.backend.get_capabilities().sign_only, }, ) if tool_name == "transfer_sol": recipient = args.get("recipient") amount = args.get("amount") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(recipient, str) or not recipient.strip(): raise WalletBackendError("recipient is required.") if not isinstance(amount, (int, float)) or amount <= 0: raise WalletBackendError("amount must be a positive number.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await self.backend.preview_native_transfer( recipient=recipient.strip(), amount_native=float(amount), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="SOL transfer", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_native_transfer( recipient=recipient.strip(), amount_native=float(amount), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="SOL transfer", ), action_label="SOL transfer", mode="prepare", ), ) execute_preview = await self.backend.preview_native_transfer( recipient=recipient.strip(), amount_native=float(amount), ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="SOL transfer", payload=execute_preview, ), action_label="SOL transfer", ) result = await self.backend.send_native_transfer( recipient=recipient.strip(), amount_native=float(amount), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="SOL transfer", mode="execute", ), ) if tool_name == "transfer_btc": recipient = args.get("recipient") amount_sats = args.get("amount_sats") fee_rate = args.get("fee_rate") confirmation_target = args.get("confirmation_target") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(recipient, str) or not recipient.strip(): raise WalletBackendError("recipient is required.") if not isinstance(amount_sats, int) or amount_sats <= 0: raise WalletBackendError("amount_sats must be a positive integer.") if fee_rate is not None and (not isinstance(fee_rate, int) or fee_rate <= 0): raise WalletBackendError("fee_rate must be a positive integer when provided.") if confirmation_target is not None and ( not isinstance(confirmation_target, int) or confirmation_target <= 0 ): raise WalletBackendError( "confirmation_target must be a positive integer when provided." ) if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "recipient": recipient.strip(), "amount_sats": amount_sats, "fee_rate": fee_rate, "confirmation_target": confirmation_target, } if mode == "preview": preview = await self.backend.preview_btc_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="BTC transfer", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_btc_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="BTC transfer", ), action_label="BTC transfer", mode="prepare", ), ) execute_preview = await self.backend.preview_btc_transfer(**preview_kwargs) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="BTC transfer", payload=execute_preview, ), action_label="BTC transfer", ) result = await self.backend.send_btc_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="BTC transfer", mode="execute", ), ) if tool_name == "transfer_evm_native": recipient = args.get("recipient") amount_wei = args.get("amount_wei") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(recipient, str) or not recipient.strip(): raise WalletBackendError("recipient is required.") if not isinstance(amount_wei, str) or not amount_wei.strip().isdigit(): raise WalletBackendError("amount_wei must be a positive integer string.") if int(amount_wei.strip()) <= 0: raise WalletBackendError("amount_wei must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "recipient": recipient.strip(), "amount_wei": amount_wei.strip(), } if mode == "preview": preview = await active_backend.preview_evm_native_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM native transfer", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_native_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM native transfer", ), action_label="EVM native transfer", mode="prepare", ), ) execute_preview = await active_backend.preview_evm_native_transfer(**preview_kwargs) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="EVM native transfer", payload=execute_preview, ), action_label="EVM native transfer", backend=active_backend, ) result = await active_backend.send_evm_native_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM native transfer", mode="execute", ), ) if tool_name == "transfer_evm_token": token_address = args.get("token_address") recipient = args.get("recipient") amount_raw = args.get("amount_raw") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(token_address, str) or not token_address.strip(): raise WalletBackendError("token_address is required.") if not isinstance(recipient, str) or not recipient.strip(): raise WalletBackendError("recipient is required.") if not isinstance(amount_raw, str) or not amount_raw.strip().isdigit(): raise WalletBackendError("amount_raw must be a positive integer string.") if int(amount_raw.strip()) <= 0: raise WalletBackendError("amount_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "token_address": token_address.strip(), "recipient": recipient.strip(), "amount_raw": amount_raw.strip(), } if mode == "preview": preview = await active_backend.preview_evm_token_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="EVM token transfer", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await active_backend.preview_evm_token_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="EVM token transfer", ), action_label="EVM token transfer", mode="prepare", ), ) execute_preview = await active_backend.preview_evm_token_transfer(**preview_kwargs) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="EVM token transfer", payload=execute_preview, ), action_label="EVM token transfer", backend=active_backend, ) result = await active_backend.send_evm_token_transfer(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="EVM token transfer", mode="execute", ), ) if tool_name in { "kamino_earn_deposit", "kamino_earn_withdraw", }: kvault = args.get("kvault") amount_ui = args.get("amount_ui") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") valid_for_seconds = args.get("valid_for_seconds", 120) if not isinstance(kvault, str) or not kvault.strip(): raise WalletBackendError("kvault is required.") if not isinstance(amount_ui, str) or not amount_ui.strip(): raise WalletBackendError("amount_ui is required.") if mode not in {"preview", "prepare", "execute", "intent_preview", "intent_execute"}: raise WalletBackendError( "mode must be 'preview', 'prepare', 'execute', 'intent_preview' or 'intent_execute'." ) if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "intent_preview" and ( not isinstance(valid_for_seconds, int) or valid_for_seconds <= 0 or valid_for_seconds > 300 ): raise WalletBackendError("valid_for_seconds must be an integer between 1 and 300.") action_label_map = { "kamino_earn_deposit": "Kamino Earn deposit", "kamino_earn_withdraw": "Kamino Earn withdraw", } intent_action_label_map = { "kamino_earn_deposit": "Kamino Earn deposit intent", "kamino_earn_withdraw": "Kamino Earn withdraw intent", } preview_method_map = { "kamino_earn_deposit": self.backend.preview_kamino_earn_deposit, "kamino_earn_withdraw": self.backend.preview_kamino_earn_withdraw, } intent_preview_method_map = { "kamino_earn_deposit": self.backend.preview_kamino_earn_deposit_intent, "kamino_earn_withdraw": self.backend.preview_kamino_earn_withdraw_intent, } execute_method_map = { "kamino_earn_deposit": self.backend.execute_kamino_earn_deposit, "kamino_earn_withdraw": self.backend.execute_kamino_earn_withdraw, } action_label = action_label_map[tool_name] intent_action_label = intent_action_label_map[tool_name] preview_method = preview_method_map[tool_name] intent_preview_method = intent_preview_method_map[tool_name] execute_method = execute_method_map[tool_name] if mode == "intent_preview": intent_preview = await intent_preview_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), valid_for_seconds=valid_for_seconds, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( intent_preview, action_label=intent_action_label, mode="preview", ), ) if mode == "intent_execute": if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(self.backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new intent preview and approval before execute." ) expected_summary = { "operation": intent_action_label, "network": str(getattr(self.backend, "network", "unknown")), "kvault": kvault.strip(), "amount_ui": amount_ui.strip(), } for key, expected_value in expected_summary.items(): if approval_summary.get(key) != expected_value: raise WalletBackendError( f"approval_token does not match the requested {action_label} intent. Generate a fresh intent preview and approval before execute." ) if approval_summary.get("recipient_policy") != "owner-only": raise WalletBackendError("approved Kamino intent recipient policy is invalid.") if approval_summary.get("spend_policy") != "exact-amount": raise WalletBackendError("approved Kamino intent spend policy is invalid.") current_owner = await self.backend.get_address() approved_owner = approval_summary.get("owner") if approved_owner and current_owner and str(approved_owner) != str(current_owner): raise WalletBackendError( "approval_token does not match the active wallet owner. Generate a fresh intent preview and approval before execute." ) valid_until = approval_summary.get("valid_until_epoch_seconds") if valid_until is not None and int(time.time()) > int(valid_until): raise WalletBackendError( "Approved Kamino intent has expired. Create a fresh intent preview." ) approval_summary_copy = dict(approval_summary) else: # No host token: build the exact intent preview a # human would have seen (same call as # mode=intent_preview), and let # _require_execute_approval's autonomous_session / # agentlayer_autonomous_approve fallback authorize it. intent_preview = await intent_preview_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), valid_for_seconds=valid_for_seconds, ) approval_summary_copy = self._build_confirmation_summary( action_label=intent_action_label, payload=intent_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label=intent_action_label, ) result = await execute_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), approved_preview=None, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label=action_label, mode="execute", ), ) if mode == "preview": preview = await preview_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label=action_label, mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await preview_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label=action_label, ), action_label=action_label, mode="prepare", ), ) execute_preview = None if isinstance(approval_token, str) and approval_token.strip(): approved_preview = args.get("_approved_preview") approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(self.backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) if isinstance(approval_summary_copy.get("_preview_digest"), str): if not isinstance(approved_preview, dict): raise WalletBackendError( f"Approved {action_label} preview payload is required for execute mode. Generate a new preview and approval before execute." ) if preview_payload_digest(approved_preview) != approval_summary_copy["_preview_digest"]: raise WalletBackendError( "approved preview payload does not match the approval token. Generate a new preview and approval before execute." ) execute_preview = dict(approved_preview) else: execute_preview = await preview_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), ) else: # No host token: same fallback as intent_execute above. execute_preview = await preview_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), ) approval_summary_copy = self._build_confirmation_summary( action_label=action_label, payload=execute_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label=action_label, ) result = await execute_method( kvault=kvault.strip(), amount_ui=amount_ui.strip(), approved_preview=execute_preview, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label=action_label, mode="execute", ), ) if tool_name in { "kamino_lend_deposit", "kamino_lend_withdraw", "kamino_lend_borrow", "kamino_lend_repay", }: market = args.get("market") reserve = args.get("reserve") amount_ui = args.get("amount_ui") obligation_address = args.get("obligation_address") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") valid_for_seconds = args.get("valid_for_seconds", 120) if not isinstance(market, str) or not market.strip(): raise WalletBackendError("market is required.") if not isinstance(reserve, str) or not reserve.strip(): raise WalletBackendError("reserve is required.") if not isinstance(amount_ui, str) or not amount_ui.strip(): raise WalletBackendError("amount_ui is required.") if obligation_address is not None and not isinstance(obligation_address, str): raise WalletBackendError("obligation_address must be a string when provided.") if mode not in {"preview", "prepare", "execute", "intent_preview", "intent_execute"}: raise WalletBackendError( "mode must be 'preview', 'prepare', 'execute', 'intent_preview' or 'intent_execute'." ) if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "intent_preview" and ( not isinstance(valid_for_seconds, int) or valid_for_seconds <= 0 or valid_for_seconds > 300 ): raise WalletBackendError("valid_for_seconds must be an integer between 1 and 300.") action_label_map = { "kamino_lend_deposit": "Kamino deposit", "kamino_lend_withdraw": "Kamino withdraw", "kamino_lend_borrow": "Kamino borrow", "kamino_lend_repay": "Kamino repay", } intent_action_label_map = { "kamino_lend_deposit": "Kamino deposit intent", "kamino_lend_withdraw": "Kamino withdraw intent", "kamino_lend_borrow": "Kamino borrow intent", "kamino_lend_repay": "Kamino repay intent", } preview_method_map = { "kamino_lend_deposit": self.backend.preview_kamino_lend_deposit, "kamino_lend_withdraw": self.backend.preview_kamino_lend_withdraw, "kamino_lend_borrow": self.backend.preview_kamino_lend_borrow, "kamino_lend_repay": self.backend.preview_kamino_lend_repay, } intent_preview_method_map = { "kamino_lend_deposit": self.backend.preview_kamino_lend_deposit_intent, "kamino_lend_withdraw": self.backend.preview_kamino_lend_withdraw_intent, "kamino_lend_borrow": self.backend.preview_kamino_lend_borrow_intent, "kamino_lend_repay": self.backend.preview_kamino_lend_repay_intent, } execute_method_map = { "kamino_lend_deposit": self.backend.execute_kamino_lend_deposit, "kamino_lend_withdraw": self.backend.execute_kamino_lend_withdraw, "kamino_lend_borrow": self.backend.execute_kamino_lend_borrow, "kamino_lend_repay": self.backend.execute_kamino_lend_repay, } action_label = action_label_map[tool_name] intent_action_label = intent_action_label_map[tool_name] preview_method = preview_method_map[tool_name] intent_preview_method = intent_preview_method_map[tool_name] execute_method = execute_method_map[tool_name] normalized_obligation_address = ( obligation_address.strip() if isinstance(obligation_address, str) and obligation_address.strip() else None ) if mode == "intent_preview": intent_preview = await intent_preview_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=normalized_obligation_address, valid_for_seconds=valid_for_seconds, ) if bool(intent_preview.get("requires_obligation_address")): raise WalletBackendError( f"{action_label} requires obligation_address when multiple Kamino obligations match the selected position." ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( intent_preview, action_label=intent_action_label, mode="preview", ), ) if mode == "intent_execute": if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(self.backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new intent preview and approval before execute." ) expected_summary = { "operation": intent_action_label, "network": str(getattr(self.backend, "network", "unknown")), "market": market.strip(), "reserve": reserve.strip(), "amount_ui": amount_ui.strip(), } for key, expected_value in expected_summary.items(): if approval_summary.get(key) != expected_value: raise WalletBackendError( f"approval_token does not match the requested {action_label} intent. Generate a fresh intent preview and approval before execute." ) if approval_summary.get("recipient_policy") != "owner-only": raise WalletBackendError("approved Kamino intent recipient policy is invalid.") if approval_summary.get("spend_policy") != "exact-amount": raise WalletBackendError("approved Kamino intent spend policy is invalid.") current_owner = await self.backend.get_address() approved_owner = approval_summary.get("owner") if approved_owner and current_owner and str(approved_owner) != str(current_owner): raise WalletBackendError( "approval_token does not match the active wallet owner. Generate a fresh intent preview and approval before execute." ) approved_obligation = approval_summary.get("obligation_address") if ( normalized_obligation_address is not None and approved_obligation is not None and str(approved_obligation) != normalized_obligation_address ): raise WalletBackendError( "approval_token does not match the requested obligation. Generate a fresh intent preview and approval before execute." ) valid_until = approval_summary.get("valid_until_epoch_seconds") if valid_until is not None and int(time.time()) > int(valid_until): raise WalletBackendError( "Approved Kamino intent has expired. Create a fresh intent preview." ) approval_summary_copy = dict(approval_summary) else: # No host token: build the exact intent preview a # human would have seen (same call as # mode=intent_preview), and let # _require_execute_approval's autonomous_session / # agentlayer_autonomous_approve fallback authorize it. intent_preview = await intent_preview_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=normalized_obligation_address, valid_for_seconds=valid_for_seconds, ) if bool(intent_preview.get("requires_obligation_address")): raise WalletBackendError( f"{action_label} requires obligation_address when multiple Kamino obligations match the selected position." ) approval_summary_copy = self._build_confirmation_summary( action_label=intent_action_label, payload=intent_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label=intent_action_label, ) approved_obligation = approval_summary_copy.get("obligation_address") result = await execute_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=( str(approved_obligation).strip() if approved_obligation else normalized_obligation_address ), approved_preview=None, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label=action_label, mode="execute", ), ) if mode == "preview": preview = await preview_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=obligation_address.strip() if isinstance(obligation_address, str) and obligation_address.strip() else None, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label=action_label, mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await preview_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=obligation_address.strip() if isinstance(obligation_address, str) and obligation_address.strip() else None, ) if bool(preview.get("requires_obligation_address")): raise WalletBackendError( f"{action_label} requires obligation_address when multiple Kamino obligations match the selected position." ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label=action_label, ), action_label=action_label, mode="prepare", ), ) execute_preview = None if isinstance(approval_token, str) and approval_token.strip(): approved_preview = args.get("_approved_preview") approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(self.backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) if isinstance(approval_summary_copy.get("_preview_digest"), str): if not isinstance(approved_preview, dict): raise WalletBackendError( f"Approved {action_label} preview payload is required for execute mode. Generate a new preview and approval before execute." ) if preview_payload_digest(approved_preview) != approval_summary_copy["_preview_digest"]: raise WalletBackendError( "approved preview payload does not match the approval token. Generate a new preview and approval before execute." ) execute_preview = dict(approved_preview) else: execute_preview = await preview_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=obligation_address.strip() if isinstance(obligation_address, str) and obligation_address.strip() else None, ) else: # No host token: same fallback as intent_execute above. execute_preview = await preview_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=obligation_address.strip() if isinstance(obligation_address, str) and obligation_address.strip() else None, ) approval_summary_copy = self._build_confirmation_summary( action_label=action_label, payload=execute_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label=action_label, ) result = await execute_method( market=market.strip(), reserve=reserve.strip(), amount_ui=amount_ui.strip(), obligation_address=obligation_address.strip() if isinstance(obligation_address, str) and obligation_address.strip() else None, approved_preview=execute_preview, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label=action_label, mode="execute", ), ) if tool_name == "stake_sol_native": vote_account = args.get("vote_account") amount = args.get("amount") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(vote_account, str) or not vote_account.strip(): raise WalletBackendError("vote_account is required.") if not isinstance(amount, (int, float)) or amount <= 0: raise WalletBackendError("amount must be a positive number.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await self.backend.preview_native_stake( vote_account=vote_account.strip(), amount_native=float(amount), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Native staking", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_native_stake( vote_account=vote_account.strip(), amount_native=float(amount), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Native staking", ), action_label="Native staking", mode="prepare", ), ) execute_preview = await self.backend.preview_native_stake( vote_account=vote_account.strip(), amount_native=float(amount), ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="Native staking", payload=execute_preview, ), action_label="Native staking", ) result = await self.backend.execute_native_stake( vote_account=vote_account.strip(), amount_native=float(amount), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Native staking", mode="execute", ), ) if tool_name == "transfer_spl_token": recipient = args.get("recipient") mint = args.get("mint") amount = args.get("amount") decimals = args.get("decimals") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(recipient, str) or not recipient.strip(): raise WalletBackendError("recipient is required.") if not isinstance(mint, str) or not mint.strip(): raise WalletBackendError("mint is required.") if not isinstance(amount, (int, float)) or amount <= 0: raise WalletBackendError("amount must be a positive number.") if decimals is not None and not isinstance(decimals, int): raise WalletBackendError("decimals must be an integer when provided.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await self.backend.preview_spl_transfer( recipient=recipient.strip(), mint=mint.strip(), amount_ui=float(amount), decimals=decimals, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="SPL token transfer", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_spl_transfer( recipient=recipient.strip(), mint=mint.strip(), amount_ui=float(amount), decimals=decimals, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="SPL token transfer", ), action_label="SPL token transfer", mode="prepare", ), ) execute_preview = await self.backend.preview_spl_transfer( recipient=recipient.strip(), mint=mint.strip(), amount_ui=float(amount), decimals=decimals, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="SPL token transfer", payload=execute_preview, ), action_label="SPL token transfer", ) result = await self.backend.send_spl_transfer( recipient=recipient.strip(), mint=mint.strip(), amount_ui=float(amount), decimals=decimals, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="SPL token transfer", mode="execute", ), ) if tool_name == "swap_solana_tokens": input_mint = args.get("input_mint") output_mint = args.get("output_mint") amount = args.get("amount") slippage_bps = args.get("slippage_bps", 300) minimum_output_amount_raw = args.get("minimum_output_amount_raw") max_fee_lamports = args.get("max_fee_lamports") valid_for_seconds = args.get("valid_for_seconds", 120) max_attempts = args.get("max_attempts", 3) mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(input_mint, str) or not input_mint.strip(): raise WalletBackendError("input_mint is required.") if not isinstance(output_mint, str) or not output_mint.strip(): raise WalletBackendError("output_mint is required.") if not isinstance(amount, (int, float)) or amount <= 0: raise WalletBackendError("amount must be a positive number.") if not isinstance(slippage_bps, int) or slippage_bps <= 0: raise WalletBackendError("slippage_bps must be a positive integer.") if minimum_output_amount_raw is not None and ( not isinstance(minimum_output_amount_raw, int) or minimum_output_amount_raw <= 0 ): raise WalletBackendError("minimum_output_amount_raw must be a positive integer when provided.") if max_fee_lamports is not None and ( not isinstance(max_fee_lamports, int) or max_fee_lamports < 0 ): raise WalletBackendError("max_fee_lamports must be a non-negative integer when provided.") if not isinstance(valid_for_seconds, int) or valid_for_seconds <= 0 or valid_for_seconds > 120: raise WalletBackendError("valid_for_seconds must be an integer between 1 and 120.") if not isinstance(max_attempts, int) or max_attempts <= 0 or max_attempts > 5: raise WalletBackendError("max_attempts must be an integer between 1 and 5.") if mode not in {"preview", "prepare", "execute", "intent_preview", "intent_execute"}: raise WalletBackendError( "mode must be 'preview', 'prepare', 'execute', 'intent_preview' or 'intent_execute'." ) if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode in {"intent_preview", "intent_execute"}: slippage_bps = max(slippage_bps, 300) valid_for_seconds = max(valid_for_seconds, 120) max_attempts = max(max_attempts, 3) if mode == "preview": preview = await self.backend.preview_swap( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Swap", mode="preview", ), ) if mode == "intent_preview": intent_preview = await self.backend.preview_swap_intent( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, minimum_output_amount_raw=minimum_output_amount_raw, max_fee_lamports=max_fee_lamports, valid_for_seconds=valid_for_seconds, max_attempts=max_attempts, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( intent_preview, action_label="Swap intent", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_swap( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Swap", ), action_label="Swap", mode="prepare", ), ) if mode == "intent_execute": if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(self.backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new intent preview and approval before execute." ) expected_summary = { "operation": "Swap intent", "network": str(getattr(self.backend, "network", "unknown")), "input_mint": input_mint.strip(), "output_mint": output_mint.strip(), } for key, expected_value in expected_summary.items(): if approval_summary.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested swap intent. Generate a fresh intent preview and approval before execute." ) current_owner = await self.backend.get_address() approved_owner = approval_summary.get("owner") if approved_owner and current_owner and str(approved_owner) != str(current_owner): raise WalletBackendError( "approval_token does not match the active wallet owner. Generate a fresh intent preview and approval before execute." ) try: approved_amount = float(approval_summary.get("input_amount_ui")) approved_slippage = int( approval_summary.get("max_slippage_bps") if approval_summary.get("max_slippage_bps") is not None else approval_summary.get("slippage_bps") ) except (TypeError, ValueError): raise WalletBackendError( "approval_token does not match the requested swap intent. Generate a fresh intent preview and approval before execute." ) if approved_amount != float(amount) or approved_slippage != slippage_bps: raise WalletBackendError( "approval_token does not match the requested swap intent. Generate a fresh intent preview and approval before execute." ) if approval_summary.get("recipient_policy") != "owner-only": raise WalletBackendError("approved swap intent recipient policy is invalid.") if approval_summary.get("spend_policy") != "exact-input": raise WalletBackendError("approved swap intent spend policy is invalid.") approval_summary_copy = dict(approval_summary) else: # No host token: build the exact intent preview a human # would have seen (same call as mode=intent_preview), # and let _require_execute_approval's autonomous_session / # agentlayer_autonomous_approve fallback authorize it. intent_preview = await self.backend.preview_swap_intent( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, minimum_output_amount_raw=minimum_output_amount_raw, max_fee_lamports=max_fee_lamports, valid_for_seconds=valid_for_seconds, max_attempts=max_attempts, ) approval_summary_copy = self._build_confirmation_summary( action_label="Swap intent", payload=intent_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="Swap intent", ) try: approved_min_output_raw = ( int(approval_summary_copy["minimum_output_amount_raw"]) if approval_summary_copy.get("minimum_output_amount_raw") is not None else None ) approved_max_fee_lamports = ( int(approval_summary_copy["max_fee_lamports"]) if approval_summary_copy.get("max_fee_lamports") is not None else None ) approved_valid_until = ( int(approval_summary_copy["valid_until_epoch_seconds"]) if approval_summary_copy.get("valid_until_epoch_seconds") is not None else None ) approved_max_attempts = int(approval_summary_copy.get("max_attempts") or max_attempts) except (TypeError, ValueError): raise WalletBackendError( "approval_token does not contain valid swap intent limits. Generate a fresh intent preview and approval before execute." ) result = await self.backend.execute_swap_intent( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, minimum_output_amount_raw=approved_min_output_raw, max_fee_lamports=approved_max_fee_lamports, valid_until_epoch_seconds=approved_valid_until, max_attempts=approved_max_attempts, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Swap", mode="execute", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet_for_backend(self.backend), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) expected_summary = { "operation": "Swap", "network": str(getattr(self.backend, "network", "unknown")), "input_mint": input_mint.strip(), "output_mint": output_mint.strip(), "slippage_bps": slippage_bps, } for key, expected_value in expected_summary.items(): if approval_summary.get(key) != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) try: approved_amount = float(approval_summary.get("input_amount_ui")) except (TypeError, ValueError): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) if approved_amount != float(amount): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) approved_preview = args.get("_approved_preview") if isinstance(approval_summary_copy.get("_preview_digest"), str): if not isinstance(approved_preview, dict): raise WalletBackendError( "Approved swap preview payload is required for execute mode. Generate a new preview and approval before execute." ) if preview_payload_digest(approved_preview) != approval_summary_copy["_preview_digest"]: raise WalletBackendError( "approved preview payload does not match the approval token. Generate a new preview and approval before execute." ) execute_preview = dict(approved_preview) else: execute_preview = await self.backend.preview_swap( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, ) else: # No host token: same fallback as intent_execute above -- # build a fresh preview and let _require_execute_approval's # autonomous_session / agentlayer_autonomous_approve # fallback authorize it. execute_preview = await self.backend.preview_swap( input_mint=input_mint.strip(), output_mint=output_mint.strip(), amount_ui=float(amount), slippage_bps=slippage_bps, ) approval_summary_copy = self._build_confirmation_summary( action_label="Swap", payload=execute_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="Swap", ) result = await self.backend.execute_swap_from_preview(execute_preview) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Swap", mode="execute", ), ) if tool_name == "swap_solana_lifi_cross_chain_tokens": input_token = args.get("input_token") destination_chain = args.get("destination_chain") output_token = args.get("output_token") destination_address = args.get("destination_address") amount_in_raw = args.get("amount_in_raw") slippage = self._normalize_lifi_slippage(args.get("slippage")) allow_bridges = self._normalize_optional_string_list(args.get("allow_bridges"), field_name="allow_bridges") deny_bridges = self._normalize_optional_string_list(args.get("deny_bridges"), field_name="deny_bridges") prefer_bridges = self._normalize_optional_string_list(args.get("prefer_bridges"), field_name="prefer_bridges") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(input_token, str) or not input_token.strip(): raise WalletBackendError("input_token is required.") if not isinstance(destination_chain, str) or not destination_chain.strip(): raise WalletBackendError("destination_chain is required.") if not isinstance(output_token, str) or not output_token.strip(): raise WalletBackendError("output_token is required.") if not isinstance(destination_address, str) or not destination_address.strip(): raise WalletBackendError("destination_address is required.") if not isinstance(amount_in_raw, str) or not amount_in_raw.strip().isdigit(): raise WalletBackendError("amount_in_raw must be a positive integer string.") if int(amount_in_raw.strip()) <= 0: raise WalletBackendError("amount_in_raw must be greater than zero.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "input_token": input_token.strip(), "destination_chain": destination_chain.strip(), "output_token": output_token.strip(), "destination_address": destination_address.strip(), "amount_in_raw": amount_in_raw.strip(), "slippage": slippage, "allow_bridges": allow_bridges, "deny_bridges": deny_bridges, "prefer_bridges": prefer_bridges, } if mode == "preview": preview = await self.backend.preview_solana_lifi_cross_chain_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Solana LI.FI cross-chain swap", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_solana_lifi_cross_chain_swap(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Solana LI.FI cross-chain swap", ), action_label="Solana LI.FI cross-chain swap", mode="prepare", ), ) if isinstance(approval_token, str) and approval_token.strip(): approval_payload = inspect_approval_token( approval_token, tool_name=tool_name, network=str(getattr(self.backend, "network", "unknown")), require_mainnet_confirmation=self._is_mainnet(), ) approval_summary = approval_payload.get("binding", {}).get("summary") if not isinstance(approval_summary, dict): raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) destination_chain_id = self._canonicalize_lifi_chain_identifier(destination_chain) expected_summary = { "operation": "Solana LI.FI cross-chain swap", "network": str(getattr(self.backend, "network", "unknown")), "source_chain": "solana", "destination_chain": destination_chain_id, "input_token": self._canonicalize_lifi_token_identifier( input_token, chain_id="1151111081099710", ), "output_token": self._canonicalize_lifi_token_identifier( output_token, chain_id=destination_chain_id, ), "destination_address": destination_address.strip(), "input_amount_raw": amount_in_raw.strip(), } for key, expected_value in expected_summary.items(): actual_value = approval_summary.get(key) if key == "destination_chain": actual_value = self._canonicalize_lifi_chain_identifier(actual_value) if key == "input_token": actual_value = self._canonicalize_lifi_token_identifier( actual_value, chain_id="1151111081099710", ) if key == "output_token": actual_value = self._canonicalize_lifi_token_identifier( actual_value, chain_id=destination_chain_id, ) if actual_value != expected_value: raise WalletBackendError( "approval_token does not match the requested operation. Generate a new approval after previewing the exact action." ) approval_summary_copy = dict(approval_summary) else: # No host token: build the exact preview a human would # have seen and let _require_execute_approval's # autonomous_session / agentlayer_autonomous_approve # fallback authorize it. fresh_preview = await self.backend.preview_solana_lifi_cross_chain_swap(**preview_kwargs) approval_summary_copy = self._build_confirmation_summary( action_label="Solana LI.FI cross-chain swap", payload=fresh_preview, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=approval_summary_copy, action_label="Solana LI.FI cross-chain swap", ) result = await self.backend.execute_solana_lifi_cross_chain_swap( input_token=str(approval_summary_copy.get("input_token") or input_token).strip(), destination_chain=str( approval_summary_copy.get("destination_chain_id") or approval_summary_copy.get("destination_chain") or destination_chain ).strip(), output_token=str(approval_summary_copy.get("output_token") or output_token).strip(), destination_address=str( approval_summary_copy.get("destination_address") or destination_address ).strip(), amount_in_raw=str(approval_summary_copy.get("input_amount_raw") or amount_in_raw).strip(), slippage=approval_summary_copy.get("slippage", slippage), allow_bridges=allow_bridges, deny_bridges=deny_bridges, prefer_bridges=prefer_bridges, minimum_output_amount_raw=( str(approval_summary_copy.get("minimum_output_amount_raw")).strip() if approval_summary_copy.get("minimum_output_amount_raw") is not None else None ), ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Solana LI.FI cross-chain swap", mode="execute", ), ) if tool_name == "launch_bags_token": name = args.get("name") symbol = args.get("symbol") description = args.get("description") image_url = args.get("image_url") website = args.get("website") twitter = args.get("twitter") telegram = args.get("telegram") discord = args.get("discord") base_mint = args.get("base_mint") claimers = args.get("claimers") basis_points = args.get("basis_points") initial_buy_sol = args.get("initial_buy_sol") bags_config_type = args.get("bags_config_type") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") for field_name, value in ( ("name", name), ("symbol", symbol), ("description", description), ("base_mint", base_mint), ): if not isinstance(value, str) or not value.strip(): raise WalletBackendError(f"{field_name} is required.") for field_name, value in ( ("image_url", image_url), ("website", website), ("twitter", twitter), ("telegram", telegram), ("discord", discord), ): if value is not None and not isinstance(value, str): raise WalletBackendError(f"{field_name} must be a string when provided.") if not isinstance(claimers, list) or not claimers or not all( isinstance(item, str) for item in claimers ): raise WalletBackendError("claimers must be a non-empty array of strings.") if not isinstance(basis_points, list) or not basis_points or not all( isinstance(item, int) for item in basis_points ): raise WalletBackendError("basis_points must be a non-empty array of integers.") if not isinstance(initial_buy_sol, (int, float)) or initial_buy_sol < 0: raise WalletBackendError("initial_buy_sol must be a non-negative number.") if bags_config_type is not None and not isinstance(bags_config_type, int): raise WalletBackendError("bags_config_type must be an integer when provided.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") preview_kwargs = { "name": name.strip(), "symbol": symbol.strip(), "description": description.strip(), "image_url": image_url.strip() if isinstance(image_url, str) and image_url.strip() else None, "website": website.strip() if isinstance(website, str) and website.strip() else None, "twitter": twitter.strip() if isinstance(twitter, str) and twitter.strip() else None, "telegram": telegram.strip() if isinstance(telegram, str) and telegram.strip() else None, "discord": discord.strip() if isinstance(discord, str) and discord.strip() else None, "base_mint": base_mint.strip(), "claimers": claimers, "basis_points": basis_points, "initial_buy_sol": float(initial_buy_sol), "bags_config_type": bags_config_type, } if mode == "preview": preview = await self.backend.preview_bags_token_launch(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Bags token launch", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_bags_token_launch(**preview_kwargs) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Bags token launch", ), action_label="Bags token launch", mode="prepare", ), ) execute_preview = await self.backend.preview_bags_token_launch(**preview_kwargs) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="Bags token launch", payload=execute_preview, ), action_label="Bags token launch", ) result = await self.backend.execute_bags_token_launch_from_preview(execute_preview) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Bags token launch", mode="execute", ), ) if tool_name == "close_empty_token_accounts": limit = args.get("limit", 8) mode = args.get("mode") purpose = args.get("purpose") approval_token = args.get("approval_token") if not isinstance(limit, int) or limit <= 0: raise WalletBackendError("limit must be a positive integer.") if mode not in {"preview", "execute"}: raise WalletBackendError("mode must be 'preview' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await self.backend.preview_close_empty_token_accounts(limit=limit) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Close token accounts", mode="preview", ), ) execute_preview = await self.backend.preview_close_empty_token_accounts(limit=limit) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="Close token accounts", payload=execute_preview, ), action_label="Close token accounts", ) result = await self.backend.close_empty_token_accounts(limit=limit) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Close token accounts", mode="execute", ), ) if tool_name == "deactivate_solana_stake": stake_account = args.get("stake_account") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(stake_account, str) or not stake_account.strip(): raise WalletBackendError("stake_account is required.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await self.backend.preview_deactivate_stake(stake_account.strip()) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Stake deactivation", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_deactivate_stake(stake_account.strip()) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Stake deactivation", ), action_label="Stake deactivation", mode="prepare", ), ) execute_preview = await self.backend.preview_deactivate_stake(stake_account.strip()) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="Stake deactivation", payload=execute_preview, ), action_label="Stake deactivation", ) result = await self.backend.execute_deactivate_stake(stake_account.strip()) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Stake deactivation", mode="execute", ), ) if tool_name == "withdraw_solana_stake": stake_account = args.get("stake_account") amount = args.get("amount") recipient = args.get("recipient") mode = args.get("mode") purpose = args.get("purpose") user_intent = args.get("user_intent", False) approval_token = args.get("approval_token") if not isinstance(stake_account, str) or not stake_account.strip(): raise WalletBackendError("stake_account is required.") if not isinstance(amount, (int, float)) or amount <= 0: raise WalletBackendError("amount must be a positive number.") if recipient is not None and not isinstance(recipient, str): raise WalletBackendError("recipient must be a string when provided.") if mode not in {"preview", "prepare", "execute"}: raise WalletBackendError("mode must be 'preview', 'prepare' or 'execute'.") if not isinstance(purpose, str) or not purpose.strip(): raise WalletBackendError("purpose is required.") if mode == "preview": preview = await self.backend.preview_withdraw_stake( stake_account=stake_account.strip(), amount_native=float(amount), recipient=recipient.strip() if isinstance(recipient, str) else None, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( preview, action_label="Stake withdraw", mode="preview", ), ) if mode == "prepare": self._require_prepare_intent(user_intent) preview = await self.backend.preview_withdraw_stake( stake_account=stake_account.strip(), amount_native=float(amount), recipient=recipient.strip() if isinstance(recipient, str) else None, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( self._build_prepare_plan( preview_payload=preview, action_label="Stake withdraw", ), action_label="Stake withdraw", mode="prepare", ), ) execute_preview = await self.backend.preview_withdraw_stake( stake_account=stake_account.strip(), amount_native=float(amount), recipient=recipient.strip() if isinstance(recipient, str) else None, ) self._require_execute_approval( approval_token=approval_token, tool_name=tool_name, summary=self._build_confirmation_summary( action_label="Stake withdraw", payload=execute_preview, ), action_label="Stake withdraw", ) result = await self.backend.execute_withdraw_stake( stake_account=stake_account.strip(), amount_native=float(amount), recipient=recipient.strip() if isinstance(recipient, str) else None, ) return AgentToolResult( tool=tool_name, ok=True, data=self._annotate_sensitive_payload( result, action_label="Stake withdraw", mode="execute", ), ) raise WalletBackendError(f"Unsupported wallet tool: {tool_name}") except Exception as exc: if isinstance(exc, WalletBackendError): return AgentToolResult( tool=tool_name, ok=False, error=str(exc), error_code=exc.code, error_details=exc.details, ) if isinstance(exc, ProviderError): return AgentToolResult( tool=tool_name, ok=False, error=str(exc), error_code=exc.provider, error_details=exc.details, ) return AgentToolResult(tool=tool_name, ok=False, error=str(exc))