API reference

Generated from docstrings in the algopay package (install algopay-sdk from PyPI). For narrative guides, see Getting started and the user guide. Repo navigation: Documentation map (includes TypeScript package path — this page is Python-only).

AlgoPay

Payment SDK for AI agents on Algorand (USDC ASA, x402).

Uses local wallet storage and Algod/Indexer — no custodial Circle API.

Source code in python/src/algopay/client.py

class AlgoPay:
    """
    Payment SDK for AI agents on Algorand (USDC ASA, x402).

    Uses local wallet storage and Algod/Indexer — no custodial Circle API.
    """

    def __init__(
        self,
        network: Network | None = None,
        algod_url: str | None = None,
        indexer_url: str | None = None,
        usdc_asa_id: int | None = None,
        log_level: int | str | None = None,
    ) -> None:
        if log_level is None:
            log_level = os.environ.get("ALGOPAY_LOG_LEVEL", "INFO")

        from algopay.core.logging import configure_logging, get_logger

        configure_logging(level=log_level)
        self._logger = get_logger("client")

        overrides: dict[str, Any] = {}
        if network is not None:
            overrides["network"] = network
        if algod_url:
            overrides["algod_url"] = algod_url
        if indexer_url:
            overrides["indexer_url"] = indexer_url
        if usdc_asa_id is not None:
            overrides["usdc_asa_id"] = usdc_asa_id

        self._config = Config.from_env(**overrides)
        self._logger.info("Initializing AlgoPay (network=%s)", self._config.network.value)

        backend = self._config.storage_backend
        self._storage = get_storage(backend, redis_url=self._config.redis_url)
        self._ledger = Ledger(self._storage)
        self._guard_manager = GuardManager(self._storage)
        self._chain = AlgorandClient(self._config)

        self._wallet_service = WalletService(self._config, self._chain)

        self._router = PaymentRouter(self._config, self._wallet_service)
        self._router.register_adapter(X402Adapter(self._config, self._wallet_service))
        self._router.register_adapter(TransferAdapter(self._config, self._wallet_service))

        self._intent_service = PaymentIntentService(self._storage)
        self._batch_processor = BatchProcessor(self._router)

    @property
    def config(self) -> Config:
        return self._config

    @property
    def wallet(self) -> WalletService:
        return self._wallet_service

    @property
    def guards(self) -> GuardManager:
        return self._guard_manager

    @property
    def ledger(self) -> Ledger:
        return self._ledger

    async def __aenter__(self) -> AlgoPay:
        """Async context manager entry."""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Async context manager exit."""
        pass

    async def get_balance(self, wallet_id: str) -> Decimal:
        """Get USDC balance for a wallet."""
        return self._wallet_service.get_usdc_balance_amount(wallet_id)

    async def create_wallet(
        self,
        blockchain: Network | str | None = None,
        wallet_set_id: str | None = None,
        account_type: AccountType = AccountType.EOA,
        name: str | None = None,
    ) -> WalletInfo:
        """
        Create a new wallet.

        Args:
            blockchain: Blockchain network (default: config.network)
            wallet_set_id: ID of existing wallet set. If None, creates a new set using `name` or default.
            account_type: Wallet type (EOA or SCA)
            name: Name for new wallet set if creating one (default: "default-set")

        Returns:
            Created WalletInfo
        """
        if not wallet_set_id:
            # Create a new set automatically
            set_name = name or f"set-{uuid.uuid4().hex[:8]}"
            wallet_set = self._wallet_service.create_wallet_set(name=set_name)
            wallet_set_id = wallet_set.id

        return self._wallet_service.create_wallet(
            wallet_set_id=wallet_set_id,
            blockchain=blockchain,
            account_type=account_type,
        )

    async def create_wallet_set(self, name: str | None = None) -> WalletSetInfo:
        """Create a new wallet set."""
        return self._wallet_service.create_wallet_set(name)

    async def list_wallets(self, wallet_set_id: str | None = None) -> list[WalletInfo]:
        """List wallets (optional filter by set)."""
        return self._wallet_service.list_wallets(wallet_set_id)

    async def list_wallet_sets(self) -> list[WalletSetInfo]:
        """List available wallet sets."""
        return self._wallet_service.list_wallet_sets()

    async def get_wallet(self, wallet_id: str) -> WalletInfo:
        """Get details of a specific wallet."""
        return self._wallet_service.get_wallet(wallet_id)

    async def list_transactions(
        self, wallet_id: str | None = None, blockchain: Network | str | None = None
    ) -> list[TransactionInfo]:
        """List transactions for a wallet or globally."""
        return self._wallet_service.list_transactions(wallet_id, blockchain)

    async def pay(
        self,
        wallet_id: str,
        recipient: str,
        amount: AmountType,
        destination_chain: Network | str | None = None,
        wallet_set_id: str | None = None,
        purpose: str | None = None,
        idempotency_key: str | None = None,
        fee_level: FeeLevel = FeeLevel.MEDIUM,
        skip_guards: bool = False,
        metadata: dict[str, Any] | None = None,
        wait_for_completion: bool = False,
        timeout_seconds: float | None = None,
        **kwargs: Any,
    ) -> PaymentResult:
        """
        Execute a payment with automatic routing (x402 URL or Algorand address transfer) and guard checks.

        Args:
            wallet_id: Source wallet ID (REQUIRED)
            recipient: Payment recipient (Algorand address or https URL for x402)
            amount: Max USDC for x402 quote or exact amount for direct transfer
            destination_chain: Ignored on Algorand (reserved for future bridges)
            wallet_set_id: Wallet set ID for hierarchical guards
            purpose: Human-readable purpose
            idempotency_key: Unique key for deduplication
            fee_level: Transaction fee level (Algorand min fee multiplier)
            skip_guards: Skip guard checks (dangerous!)
            metadata: Additional metadata
            wait_for_completion: Wait for transaction confirmation
            timeout_seconds: Maximum wait time
            **kwargs: Extra arguments passed to the payment router / adapters

        Returns:
            PaymentResult with transaction details
        """
        if not wallet_id:
            raise ValidationError("wallet_id is required")

        amount_decimal = Decimal(str(amount))
        if amount_decimal <= 0:
            raise ValidationError(f"Payment amount must be positive. Got: {amount_decimal}")

        idempotency_key = idempotency_key or str(uuid.uuid4())

        meta = metadata or {}
        meta["idempotency_key"] = idempotency_key

        context = PaymentContext(
            wallet_id=wallet_id,
            wallet_set_id=wallet_set_id,
            recipient=recipient,
            amount=amount_decimal,
            purpose=purpose,
            metadata=meta,
        )

        ledger_entry = LedgerEntry(
            wallet_id=wallet_id,
            recipient=recipient,
            amount=amount_decimal,
            purpose=purpose,
            metadata=metadata or {},
        )
        await self._ledger.record(ledger_entry)

        guards_chain = None
        reservation_tokens = []
        guards_passed: list[str] = []

        if not skip_guards:
            guards_chain = await self._guard_manager.get_guard_chain(
                wallet_id=wallet_id, wallet_set_id=wallet_set_id
            )
            try:
                reservation_tokens = await guards_chain.reserve(context)
                guards_passed = [g.name for g in guards_chain]
            except ValueError as e:
                await self._ledger.update_status(
                    ledger_entry.id,
                    LedgerEntryStatus.BLOCKED,
                    tx_hash=None,
                )

                return PaymentResult(
                    success=False,
                    transaction_id=None,
                    blockchain_tx=None,
                    amount=amount_decimal,
                    recipient=recipient,
                    method=PaymentMethod.TRANSFER,
                    status=PaymentStatus.BLOCKED,
                    error=f"Blocked by guard: {e}",
                    guards_passed=guards_passed,
                    metadata={"guard_reason": str(e)},
                )

        try:
            result = await self._router.pay(
                wallet_id=wallet_id,
                recipient=recipient,
                amount=amount_decimal,
                purpose=purpose,
                guards_passed=guards_passed,
                fee_level=fee_level,
                idempotency_key=idempotency_key,
                destination_chain=destination_chain,
                wait_for_completion=wait_for_completion,
                timeout_seconds=timeout_seconds,
                **kwargs,
            )

            if result.success:
                await self._ledger.update_status(
                    ledger_entry.id,
                    LedgerEntryStatus.COMPLETED
                    if result.status == PaymentStatus.COMPLETED
                    else LedgerEntryStatus.PENDING,
                    result.blockchain_tx,
                    metadata_updates={"transaction_id": result.transaction_id},
                )

                if guards_chain:
                    await guards_chain.commit(reservation_tokens)
            else:
                await self._ledger.update_status(
                    ledger_entry.id,
                    LedgerEntryStatus.FAILED,
                )
                if guards_chain:
                    await guards_chain.release(reservation_tokens)

            return result

        except Exception:
            if guards_chain:
                await guards_chain.release(reservation_tokens)

            await self._ledger.update_status(
                ledger_entry.id,
                LedgerEntryStatus.FAILED,
            )
            raise

    async def simulate(
        self,
        wallet_id: str,
        recipient: str,
        amount: Decimal | str,
        wallet_set_id: str | None = None,
        **kwargs: Any,
    ) -> SimulationResult:
        """
        Simulate a payment without executing.

        Checks:
        - Guards would pass
        - Balance is sufficient
        - Recipient is valid

        Args:
            wallet_id: Source wallet ID (REQUIRED)
            recipient: Payment recipient
            amount: Amount to simulate
            wallet_set_id: Optional wallet set ID (for set-level guards)
            **kwargs: Additional parameters

        Returns:
            SimulationResult with would_succeed and details
        """
        if not wallet_id:
            return SimulationResult(
                would_succeed=False,
                route=PaymentMethod.TRANSFER,
                reason="wallet_id is required",
            )

        amount_decimal = Decimal(str(amount))

        # Check guards first
        context = PaymentContext(
            wallet_id=wallet_id,
            wallet_set_id=wallet_set_id,
            recipient=recipient,
            amount=amount_decimal,
            purpose="Simulation",
        )

        allowed, reason, _ = await self._guard_manager.check(context)
        if not allowed:
            return SimulationResult(
                would_succeed=False,
                route=PaymentMethod.TRANSFER,
                reason=f"Would be blocked by guard: {reason}",
            )

        # Check via router
        return await self._router.simulate(
            wallet_id=wallet_id,
            recipient=recipient,
            amount=amount_decimal,
            **kwargs,
        )

    def can_pay(self, recipient: str) -> bool:
        """
        Check if a recipient can be paid.

        Args:
            recipient: Payment recipient

        Returns:
            True if an adapter can handle this recipient
        """
        return self._router.can_handle(recipient)

    def detect_method(self, recipient: str) -> PaymentMethod | None:
        """Detect which payment method would be used for a recipient."""
        return self._router.detect_method(recipient)

    @property
    def intents(self) -> PaymentIntentService:
        """Get intent management service."""
        return self._intent_service

    async def create_payment_intent(
        self,
        wallet_id: str,
        recipient: str,
        amount: AmountType,
        purpose: str | None = None,
        idempotency_key: str | None = None,
        **kwargs: Any,
    ) -> PaymentIntent:
        """Create a Payment Intent (Authorize)."""
        # Simulate check (Routing + Guards) strictly
        sim_result = await self.simulate(
            wallet_id=wallet_id, recipient=recipient, amount=amount, **kwargs
        )

        if not sim_result.would_succeed:
            raise PaymentError(f"Authorization failed: {sim_result.reason}")

        # Create Intent
        metadata = kwargs.copy()
        metadata.update(
            {
                "purpose": purpose,
                "idempotency_key": idempotency_key,
                "simulated_route": sim_result.route.value,
            }
        )

        intent = await self._intent_service.create(
            wallet_id=wallet_id, recipient=recipient, amount=Decimal(str(amount)), metadata=metadata
        )
        return intent

    async def confirm_payment_intent(self, intent_id: str) -> PaymentResult:
        """Confirm and execute a Payment Intent (Capture)."""
        intent = await self._intent_service.get(intent_id)
        if not intent:
            raise ValidationError(f"Intent not found: {intent_id}")

        if intent.status != PaymentIntentStatus.REQUIRES_CONFIRMATION:
            raise ValidationError(f"Intent cannot be confirmed. Status: {intent.status}")

        try:
            # Update to Processing
            await self._intent_service.update_status(intent.id, PaymentIntentStatus.PROCESSING)

            # Prepare exec args from intent + metadata
            exec_kwargs = intent.metadata.copy()

            # Remove internal metadata keys that aren't for routing
            purpose = exec_kwargs.pop("purpose", None)
            idempotency_key = exec_kwargs.pop("idempotency_key", None)
            exec_kwargs.pop("simulated_route", None)

            # Execute Pay
            result = await self.pay(
                wallet_id=intent.wallet_id,
                recipient=intent.recipient,
                amount=intent.amount,
                purpose=purpose,
                idempotency_key=idempotency_key,
                **exec_kwargs,
            )

            if result.success:
                await self._intent_service.update_status(intent.id, PaymentIntentStatus.SUCCEEDED)
            else:
                await self._intent_service.update_status(intent.id, PaymentIntentStatus.FAILED)

            return result

        except Exception as e:
            # Mark failed on exception
            await self._intent_service.update_status(intent.id, PaymentIntentStatus.FAILED)
            raise e

    async def get_payment_intent(self, intent_id: str) -> PaymentIntent | None:
        """Get Payment Intent by ID."""
        return await self._intent_service.get(intent_id)

    async def cancel_payment_intent(self, intent_id: str) -> PaymentIntent:
        """Cancel a Payment Intent."""
        intent = await self._intent_service.get(intent_id)
        if not intent:
            raise ValidationError(f"Intent not found: {intent_id}")

        if intent.status not in (PaymentIntentStatus.REQUIRES_CONFIRMATION,):
            raise ValidationError(f"Cannot cancel intent in status: {intent.status}")

        return await self._intent_service.update_status(intent.id, PaymentIntentStatus.CANCELED)

    async def batch_pay(
        self, requests: list[PaymentRequest], concurrency: int = 5
    ) -> BatchPaymentResult:
        """
        Execute multiple payments in batch.

        Args:
            requests: List of payment requests to execute
            concurrency: Maximum number of concurrent executions (default 5)

        Returns:
            BatchPaymentResult containing status of all payments
        """
        return await self._batch_processor.process(requests, concurrency)

    async def sync_transaction(self, entry_id: str) -> LedgerEntry:
        """Synchronize a ledger entry with Algorand indexer (confirmed tx)."""
        entry = await self._ledger.get(entry_id)
        if not entry:
            raise ValidationError(f"Ledger entry not found: {entry_id}")

        tx_id = entry.tx_hash or entry.metadata.get("transaction_id")
        if not tx_id:
            raise ValidationError("Ledger entry has no transaction id / tx_hash to sync")

        tx = self._chain.transaction_by_id(tx_id)
        if not tx:
            raise PaymentError(f"Transaction not found on indexer: {tx_id}")

        new_status = LedgerEntryStatus.COMPLETED
        await self._ledger.update_status(
            entry.id,
            new_status,
            tx_hash=tx_id,
            metadata_updates={
                "last_synced": datetime.now(timezone.utc).isoformat(),
                "indexer_confirmed": True,
            },
        )

        updated = await self._ledger.get(entry.id)
        return updated  # type: ignore

    async def add_budget_guard(
        self,
        wallet_id: str,
        daily_limit: str | Decimal | None = None,
        hourly_limit: str | Decimal | None = None,
        total_limit: str | Decimal | None = None,
        name: str = "budget",
    ) -> None:
        """
        Add a budget guard to a wallet.

        Enforce spending limits over time periods (Atomic & Reliable).

        Args:
            wallet_id: Target wallet ID
            daily_limit: Max spend per 24h
            hourly_limit: Max spend per 1h
            total_limit: Max total spend (lifetime)
            name: Custom name for the guard
        """
        from algopay.guards.budget import BudgetGuard

        d_limit = Decimal(str(daily_limit)) if daily_limit else None
        h_limit = Decimal(str(hourly_limit)) if hourly_limit else None
        t_limit = Decimal(str(total_limit)) if total_limit else None

        guard = BudgetGuard(
            daily_limit=d_limit, hourly_limit=h_limit, total_limit=t_limit, name=name
        )
        await self._guard_manager.add_guard(wallet_id, guard)

    async def add_budget_guard_for_set(
        self,
        wallet_set_id: str,
        daily_limit: str | Decimal | None = None,
        hourly_limit: str | Decimal | None = None,
        total_limit: str | Decimal | None = None,
        name: str = "budget",
    ) -> None:
        """
        Add a budget guard to a wallet set (applies to ALL wallets in the set).

        Args:
            wallet_set_id: Target wallet set ID
            daily_limit: Max spend per 24h
            hourly_limit: Max spend per 1h
            total_limit: Max total spend (lifetime)
            name: Custom name for the guard
        """
        from algopay.guards.budget import BudgetGuard

        d_limit = Decimal(str(daily_limit)) if daily_limit else None
        h_limit = Decimal(str(hourly_limit)) if hourly_limit else None
        t_limit = Decimal(str(total_limit)) if total_limit else None

        guard = BudgetGuard(
            daily_limit=d_limit, hourly_limit=h_limit, total_limit=t_limit, name=name
        )
        await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

    async def add_single_tx_guard(
        self,
        wallet_id: str,
        max_amount: str | Decimal,
        min_amount: str | Decimal | None = None,
        name: str = "single_tx",
    ) -> None:
        """
        Add a Single Transaction Limit guard.

        Args:
            wallet_id: Target wallet ID
            max_amount: Max amount per transaction
            min_amount: Min amount per transaction
            name: Guard name
        """
        from algopay.guards.single_tx import SingleTxGuard

        guard = SingleTxGuard(
            max_amount=Decimal(str(max_amount)),
            min_amount=Decimal(str(min_amount)) if min_amount else None,
            name=name,
        )
        await self._guard_manager.add_guard(wallet_id, guard)

    async def add_recipient_guard(
        self,
        wallet_id: str,
        mode: str = "whitelist",
        addresses: list[str] | None = None,
        patterns: list[str] | None = None,
        domains: list[str] | None = None,
        name: str = "recipient",
    ) -> None:
        """
        Add a Recipient Access Control guard.

        Args:
            wallet_id: Target wallet ID
            mode: 'whitelist' (allow specific) or 'blacklist' (block specific)
            addresses: List of allowed/blocked addresses
            patterns: List of regex patterns
            domains: List of allowed/blocked domains (for x402/URLs)
            name: Guard name
        """
        from algopay.guards.recipient import RecipientGuard

        guard = RecipientGuard(
            mode=mode, addresses=addresses, patterns=patterns, domains=domains, name=name
        )
        await self._guard_manager.add_guard(wallet_id, guard)

    async def add_rate_limit_guard(
        self,
        wallet_id: str,
        max_per_minute: int | None = None,
        max_per_hour: int | None = None,
        max_per_day: int | None = None,
        name: str = "rate_limit",
    ) -> None:
        """
        Add a rate limit guard to a wallet.

        Limit number of transactions per time window.

        Args:
            wallet_id: Target wallet ID
            max_per_minute: Max txs per minute
            max_per_hour: Max txs per hour
            max_per_day: Max txs per day
            name: Custom name for the guard
        """
        from algopay.guards.rate_limit import RateLimitGuard

        guard = RateLimitGuard(
            max_per_minute=max_per_minute,
            max_per_hour=max_per_hour,
            max_per_day=max_per_day,
            name=name,
        )
        await self._guard_manager.add_guard(wallet_id, guard)

    async def add_confirm_guard(
        self,
        wallet_id: str,
        threshold: str | Decimal | None = None,
        always_confirm: bool = False,
        name: str = "confirm",
    ) -> None:
        """
        Add a confirmation guard to a wallet (Human-in-the-Loop).

        Payments above the threshold require explicit confirmation via callback
        or external handling (e.g., webhook approval).

        Args:
            wallet_id: Target wallet ID
            threshold: Amount above which confirmation is required
            always_confirm: If True, require confirmation for ALL payments
            name: Custom name for the guard
        """
        from algopay.guards.confirm import ConfirmGuard

        t_threshold = Decimal(str(threshold)) if threshold else None

        guard = ConfirmGuard(threshold=t_threshold, always_confirm=always_confirm, name=name)
        await self._guard_manager.add_guard(wallet_id, guard)

    async def add_confirm_guard_for_set(
        self,
        wallet_set_id: str,
        threshold: str | Decimal | None = None,
        always_confirm: bool = False,
        name: str = "confirm",
    ) -> None:
        """
        Add a confirmation guard to a wallet set (applies to ALL wallets in the set).

        Args:
            wallet_set_id: Target wallet set ID
            threshold: Amount above which confirmation is required
            always_confirm: If True, require confirmation for ALL payments
            name: Custom name for the guard
        """
        from algopay.guards.confirm import ConfirmGuard

        t_threshold = Decimal(str(threshold)) if threshold else None

        guard = ConfirmGuard(threshold=t_threshold, always_confirm=always_confirm, name=name)
        await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

    async def add_justification_guard(
        self,
        wallet_id: str,
        min_length: int = 1,
        name: str = "justification",
    ) -> None:
        """
        Require a non-empty payment ``purpose`` (justification), Locus-style.

        Args:
            wallet_id: Target wallet ID
            min_length: Minimum stripped length of ``purpose`` (default: 1)
            name: Guard name
        """
        from algopay.guards.justification import JustificationGuard

        guard = JustificationGuard(min_length=min_length, name=name)
        await self._guard_manager.add_guard(wallet_id, guard)

    async def add_justification_guard_for_set(
        self,
        wallet_set_id: str,
        min_length: int = 1,
        name: str = "justification",
    ) -> None:
        """Require justification for every wallet in the set."""
        from algopay.guards.justification import JustificationGuard

        guard = JustificationGuard(min_length=min_length, name=name)
        await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

    async def add_rate_limit_guard_for_set(
        self,
        wallet_set_id: str,
        max_per_minute: int | None = None,
        max_per_hour: int | None = None,
        max_per_day: int | None = None,
        name: str = "rate_limit",
    ) -> None:
        """
        Add a rate limit guard to a wallet set (applies to ALL wallets in the set).

        Args:
            wallet_set_id: Target wallet set ID
            max_per_minute: Max txs per minute
            max_per_hour: Max txs per hour
            max_per_day: Max txs per day
            name: Custom name for the guard
        """
        from algopay.guards.rate_limit import RateLimitGuard

        guard = RateLimitGuard(
            max_per_minute=max_per_minute,
            max_per_hour=max_per_hour,
            max_per_day=max_per_day,
            name=name,
        )
        await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

    async def add_recipient_guard_for_set(
        self,
        wallet_set_id: str,
        mode: str = "whitelist",
        addresses: list[str] | None = None,
        patterns: list[str] | None = None,
        domains: list[str] | None = None,
        name: str = "recipient",
    ) -> None:
        """
        Add a Recipient Access Control guard to a wallet set.

        Args:
            wallet_set_id: Target wallet set ID
            mode: 'whitelist' (allow specific) or 'blacklist' (block specific)
            addresses: List of allowed/blocked addresses
            patterns: List of regex patterns
            domains: List of allowed/blocked domains (for x402/URLs)
            name: Guard name
        """
        from algopay.guards.recipient import RecipientGuard

        guard = RecipientGuard(
            mode=mode, addresses=addresses, patterns=patterns, domains=domains, name=name
        )
        await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

    async def list_guards(self, wallet_id: str) -> list[str]:
        """
        List all guard names registered for a wallet.

        Args:
            wallet_id: Target wallet ID

        Returns:
            List of guard names
        """
        return await self._guard_manager.list_wallet_guard_names(wallet_id)

    async def list_guards_for_set(self, wallet_set_id: str) -> list[str]:
        """
        List all guard names registered for a wallet set.

        Args:
            wallet_set_id: Target wallet set ID

        Returns:
            List of guard names
        """
        return await self._guard_manager.list_wallet_set_guard_names(wallet_set_id)

intents property

intents

Get intent management service.

__aenter__ async

__aenter__()

Async context manager entry.

Source code in python/src/algopay/client.py

async def __aenter__(self) -> AlgoPay:
    """Async context manager entry."""
    return self

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)

Async context manager exit.

Source code in python/src/algopay/client.py

async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
    """Async context manager exit."""
    pass

get_balance async

get_balance(wallet_id)

Get USDC balance for a wallet.

Source code in python/src/algopay/client.py

async def get_balance(self, wallet_id: str) -> Decimal:
    """Get USDC balance for a wallet."""
    return self._wallet_service.get_usdc_balance_amount(wallet_id)

create_wallet async

create_wallet(blockchain=None, wallet_set_id=None, account_type=AccountType.EOA, name=None)

Create a new wallet.

Parameters:

Name	Type	Description	Default
blockchain	Network | str | None	Blockchain network (default: config.network)	None
wallet_set_id	str | None	ID of existing wallet set. If None, creates a new set using name or default.	None
account_type	AccountType	Wallet type (EOA or SCA)	EOA
name	str | None	Name for new wallet set if creating one (default: "default-set")	None

Returns:

Type	Description
WalletInfo	Created WalletInfo

Source code in python/src/algopay/client.py

async def create_wallet(
    self,
    blockchain: Network | str | None = None,
    wallet_set_id: str | None = None,
    account_type: AccountType = AccountType.EOA,
    name: str | None = None,
) -> WalletInfo:
    """
    Create a new wallet.

    Args:
        blockchain: Blockchain network (default: config.network)
        wallet_set_id: ID of existing wallet set. If None, creates a new set using `name` or default.
        account_type: Wallet type (EOA or SCA)
        name: Name for new wallet set if creating one (default: "default-set")

    Returns:
        Created WalletInfo
    """
    if not wallet_set_id:
        # Create a new set automatically
        set_name = name or f"set-{uuid.uuid4().hex[:8]}"
        wallet_set = self._wallet_service.create_wallet_set(name=set_name)
        wallet_set_id = wallet_set.id

    return self._wallet_service.create_wallet(
        wallet_set_id=wallet_set_id,
        blockchain=blockchain,
        account_type=account_type,
    )

create_wallet_set async

create_wallet_set(name=None)

Create a new wallet set.

Source code in python/src/algopay/client.py

async def create_wallet_set(self, name: str | None = None) -> WalletSetInfo:
    """Create a new wallet set."""
    return self._wallet_service.create_wallet_set(name)

list_wallets async

list_wallets(wallet_set_id=None)

List wallets (optional filter by set).

Source code in python/src/algopay/client.py

async def list_wallets(self, wallet_set_id: str | None = None) -> list[WalletInfo]:
    """List wallets (optional filter by set)."""
    return self._wallet_service.list_wallets(wallet_set_id)

list_wallet_sets async

list_wallet_sets()

List available wallet sets.

Source code in python/src/algopay/client.py

async def list_wallet_sets(self) -> list[WalletSetInfo]:
    """List available wallet sets."""
    return self._wallet_service.list_wallet_sets()

get_wallet async

get_wallet(wallet_id)

Get details of a specific wallet.

Source code in python/src/algopay/client.py

async def get_wallet(self, wallet_id: str) -> WalletInfo:
    """Get details of a specific wallet."""
    return self._wallet_service.get_wallet(wallet_id)

list_transactions async

list_transactions(wallet_id=None, blockchain=None)

List transactions for a wallet or globally.

Source code in python/src/algopay/client.py

async def list_transactions(
    self, wallet_id: str | None = None, blockchain: Network | str | None = None
) -> list[TransactionInfo]:
    """List transactions for a wallet or globally."""
    return self._wallet_service.list_transactions(wallet_id, blockchain)

pay async

pay(wallet_id, recipient, amount, destination_chain=None, wallet_set_id=None, purpose=None, idempotency_key=None, fee_level=FeeLevel.MEDIUM, skip_guards=False, metadata=None, wait_for_completion=False, timeout_seconds=None, **kwargs)

Execute a payment with automatic routing (x402 URL or Algorand address transfer) and guard checks.

Parameters:

Name	Type	Description	Default
wallet_id	str	Source wallet ID (REQUIRED)	required
recipient	str	Payment recipient (Algorand address or https URL for x402)	required
amount	AmountType	Max USDC for x402 quote or exact amount for direct transfer	required
destination_chain	Network | str | None	Ignored on Algorand (reserved for future bridges)	None
wallet_set_id	str | None	Wallet set ID for hierarchical guards	None
purpose	str | None	Human-readable purpose	None
idempotency_key	str | None	Unique key for deduplication	None
fee_level	FeeLevel	Transaction fee level (Algorand min fee multiplier)	MEDIUM
skip_guards	bool	Skip guard checks (dangerous!)	False
metadata	dict[str, Any] | None	Additional metadata	None
wait_for_completion	bool	Wait for transaction confirmation	False
timeout_seconds	float | None	Maximum wait time	None
**kwargs	Any	Extra arguments passed to the payment router / adapters	{}

Returns:

Type	Description
PaymentResult	PaymentResult with transaction details

Source code in python/src/algopay/client.py

async def pay(
    self,
    wallet_id: str,
    recipient: str,
    amount: AmountType,
    destination_chain: Network | str | None = None,
    wallet_set_id: str | None = None,
    purpose: str | None = None,
    idempotency_key: str | None = None,
    fee_level: FeeLevel = FeeLevel.MEDIUM,
    skip_guards: bool = False,
    metadata: dict[str, Any] | None = None,
    wait_for_completion: bool = False,
    timeout_seconds: float | None = None,
    **kwargs: Any,
) -> PaymentResult:
    """
    Execute a payment with automatic routing (x402 URL or Algorand address transfer) and guard checks.

    Args:
        wallet_id: Source wallet ID (REQUIRED)
        recipient: Payment recipient (Algorand address or https URL for x402)
        amount: Max USDC for x402 quote or exact amount for direct transfer
        destination_chain: Ignored on Algorand (reserved for future bridges)
        wallet_set_id: Wallet set ID for hierarchical guards
        purpose: Human-readable purpose
        idempotency_key: Unique key for deduplication
        fee_level: Transaction fee level (Algorand min fee multiplier)
        skip_guards: Skip guard checks (dangerous!)
        metadata: Additional metadata
        wait_for_completion: Wait for transaction confirmation
        timeout_seconds: Maximum wait time
        **kwargs: Extra arguments passed to the payment router / adapters

    Returns:
        PaymentResult with transaction details
    """
    if not wallet_id:
        raise ValidationError("wallet_id is required")

    amount_decimal = Decimal(str(amount))
    if amount_decimal <= 0:
        raise ValidationError(f"Payment amount must be positive. Got: {amount_decimal}")

    idempotency_key = idempotency_key or str(uuid.uuid4())

    meta = metadata or {}
    meta["idempotency_key"] = idempotency_key

    context = PaymentContext(
        wallet_id=wallet_id,
        wallet_set_id=wallet_set_id,
        recipient=recipient,
        amount=amount_decimal,
        purpose=purpose,
        metadata=meta,
    )

    ledger_entry = LedgerEntry(
        wallet_id=wallet_id,
        recipient=recipient,
        amount=amount_decimal,
        purpose=purpose,
        metadata=metadata or {},
    )
    await self._ledger.record(ledger_entry)

    guards_chain = None
    reservation_tokens = []
    guards_passed: list[str] = []

    if not skip_guards:
        guards_chain = await self._guard_manager.get_guard_chain(
            wallet_id=wallet_id, wallet_set_id=wallet_set_id
        )
        try:
            reservation_tokens = await guards_chain.reserve(context)
            guards_passed = [g.name for g in guards_chain]
        except ValueError as e:
            await self._ledger.update_status(
                ledger_entry.id,
                LedgerEntryStatus.BLOCKED,
                tx_hash=None,
            )

            return PaymentResult(
                success=False,
                transaction_id=None,
                blockchain_tx=None,
                amount=amount_decimal,
                recipient=recipient,
                method=PaymentMethod.TRANSFER,
                status=PaymentStatus.BLOCKED,
                error=f"Blocked by guard: {e}",
                guards_passed=guards_passed,
                metadata={"guard_reason": str(e)},
            )

    try:
        result = await self._router.pay(
            wallet_id=wallet_id,
            recipient=recipient,
            amount=amount_decimal,
            purpose=purpose,
            guards_passed=guards_passed,
            fee_level=fee_level,
            idempotency_key=idempotency_key,
            destination_chain=destination_chain,
            wait_for_completion=wait_for_completion,
            timeout_seconds=timeout_seconds,
            **kwargs,
        )

        if result.success:
            await self._ledger.update_status(
                ledger_entry.id,
                LedgerEntryStatus.COMPLETED
                if result.status == PaymentStatus.COMPLETED
                else LedgerEntryStatus.PENDING,
                result.blockchain_tx,
                metadata_updates={"transaction_id": result.transaction_id},
            )

            if guards_chain:
                await guards_chain.commit(reservation_tokens)
        else:
            await self._ledger.update_status(
                ledger_entry.id,
                LedgerEntryStatus.FAILED,
            )
            if guards_chain:
                await guards_chain.release(reservation_tokens)

        return result

    except Exception:
        if guards_chain:
            await guards_chain.release(reservation_tokens)

        await self._ledger.update_status(
            ledger_entry.id,
            LedgerEntryStatus.FAILED,
        )
        raise

simulate async

simulate(wallet_id, recipient, amount, wallet_set_id=None, **kwargs)

Simulate a payment without executing.

Checks: - Guards would pass - Balance is sufficient - Recipient is valid

Parameters:

Name	Type	Description	Default
wallet_id	str	Source wallet ID (REQUIRED)	required
recipient	str	Payment recipient	required
amount	Decimal | str	Amount to simulate	required
wallet_set_id	str | None	Optional wallet set ID (for set-level guards)	None
**kwargs	Any	Additional parameters	{}

Returns:

Type	Description
SimulationResult	SimulationResult with would_succeed and details

Source code in python/src/algopay/client.py

async def simulate(
    self,
    wallet_id: str,
    recipient: str,
    amount: Decimal | str,
    wallet_set_id: str | None = None,
    **kwargs: Any,
) -> SimulationResult:
    """
    Simulate a payment without executing.

    Checks:
    - Guards would pass
    - Balance is sufficient
    - Recipient is valid

    Args:
        wallet_id: Source wallet ID (REQUIRED)
        recipient: Payment recipient
        amount: Amount to simulate
        wallet_set_id: Optional wallet set ID (for set-level guards)
        **kwargs: Additional parameters

    Returns:
        SimulationResult with would_succeed and details
    """
    if not wallet_id:
        return SimulationResult(
            would_succeed=False,
            route=PaymentMethod.TRANSFER,
            reason="wallet_id is required",
        )

    amount_decimal = Decimal(str(amount))

    # Check guards first
    context = PaymentContext(
        wallet_id=wallet_id,
        wallet_set_id=wallet_set_id,
        recipient=recipient,
        amount=amount_decimal,
        purpose="Simulation",
    )

    allowed, reason, _ = await self._guard_manager.check(context)
    if not allowed:
        return SimulationResult(
            would_succeed=False,
            route=PaymentMethod.TRANSFER,
            reason=f"Would be blocked by guard: {reason}",
        )

    # Check via router
    return await self._router.simulate(
        wallet_id=wallet_id,
        recipient=recipient,
        amount=amount_decimal,
        **kwargs,
    )

can_pay

can_pay(recipient)

Check if a recipient can be paid.

Parameters:

Name	Type	Description	Default
recipient	str	Payment recipient	required

Returns:

Type	Description
bool	True if an adapter can handle this recipient

Source code in python/src/algopay/client.py

def can_pay(self, recipient: str) -> bool:
    """
    Check if a recipient can be paid.

    Args:
        recipient: Payment recipient

    Returns:
        True if an adapter can handle this recipient
    """
    return self._router.can_handle(recipient)

detect_method

detect_method(recipient)

Detect which payment method would be used for a recipient.

Source code in python/src/algopay/client.py

def detect_method(self, recipient: str) -> PaymentMethod | None:
    """Detect which payment method would be used for a recipient."""
    return self._router.detect_method(recipient)

create_payment_intent async

create_payment_intent(wallet_id, recipient, amount, purpose=None, idempotency_key=None, **kwargs)

Create a Payment Intent (Authorize).

Source code in python/src/algopay/client.py

async def create_payment_intent(
    self,
    wallet_id: str,
    recipient: str,
    amount: AmountType,
    purpose: str | None = None,
    idempotency_key: str | None = None,
    **kwargs: Any,
) -> PaymentIntent:
    """Create a Payment Intent (Authorize)."""
    # Simulate check (Routing + Guards) strictly
    sim_result = await self.simulate(
        wallet_id=wallet_id, recipient=recipient, amount=amount, **kwargs
    )

    if not sim_result.would_succeed:
        raise PaymentError(f"Authorization failed: {sim_result.reason}")

    # Create Intent
    metadata = kwargs.copy()
    metadata.update(
        {
            "purpose": purpose,
            "idempotency_key": idempotency_key,
            "simulated_route": sim_result.route.value,
        }
    )

    intent = await self._intent_service.create(
        wallet_id=wallet_id, recipient=recipient, amount=Decimal(str(amount)), metadata=metadata
    )
    return intent

confirm_payment_intent async

confirm_payment_intent(intent_id)

Confirm and execute a Payment Intent (Capture).

Source code in python/src/algopay/client.py

async def confirm_payment_intent(self, intent_id: str) -> PaymentResult:
    """Confirm and execute a Payment Intent (Capture)."""
    intent = await self._intent_service.get(intent_id)
    if not intent:
        raise ValidationError(f"Intent not found: {intent_id}")

    if intent.status != PaymentIntentStatus.REQUIRES_CONFIRMATION:
        raise ValidationError(f"Intent cannot be confirmed. Status: {intent.status}")

    try:
        # Update to Processing
        await self._intent_service.update_status(intent.id, PaymentIntentStatus.PROCESSING)

        # Prepare exec args from intent + metadata
        exec_kwargs = intent.metadata.copy()

        # Remove internal metadata keys that aren't for routing
        purpose = exec_kwargs.pop("purpose", None)
        idempotency_key = exec_kwargs.pop("idempotency_key", None)
        exec_kwargs.pop("simulated_route", None)

        # Execute Pay
        result = await self.pay(
            wallet_id=intent.wallet_id,
            recipient=intent.recipient,
            amount=intent.amount,
            purpose=purpose,
            idempotency_key=idempotency_key,
            **exec_kwargs,
        )

        if result.success:
            await self._intent_service.update_status(intent.id, PaymentIntentStatus.SUCCEEDED)
        else:
            await self._intent_service.update_status(intent.id, PaymentIntentStatus.FAILED)

        return result

    except Exception as e:
        # Mark failed on exception
        await self._intent_service.update_status(intent.id, PaymentIntentStatus.FAILED)
        raise e

get_payment_intent async

get_payment_intent(intent_id)

Get Payment Intent by ID.

Source code in python/src/algopay/client.py

async def get_payment_intent(self, intent_id: str) -> PaymentIntent | None:
    """Get Payment Intent by ID."""
    return await self._intent_service.get(intent_id)

cancel_payment_intent async

cancel_payment_intent(intent_id)

Cancel a Payment Intent.

Source code in python/src/algopay/client.py

async def cancel_payment_intent(self, intent_id: str) -> PaymentIntent:
    """Cancel a Payment Intent."""
    intent = await self._intent_service.get(intent_id)
    if not intent:
        raise ValidationError(f"Intent not found: {intent_id}")

    if intent.status not in (PaymentIntentStatus.REQUIRES_CONFIRMATION,):
        raise ValidationError(f"Cannot cancel intent in status: {intent.status}")

    return await self._intent_service.update_status(intent.id, PaymentIntentStatus.CANCELED)

batch_pay async

batch_pay(requests, concurrency=5)

Execute multiple payments in batch.

Parameters:

Name	Type	Description	Default
requests	list[PaymentRequest]	List of payment requests to execute	required
concurrency	int	Maximum number of concurrent executions (default 5)	5

Returns:

Type	Description
BatchPaymentResult	BatchPaymentResult containing status of all payments

Source code in python/src/algopay/client.py

async def batch_pay(
    self, requests: list[PaymentRequest], concurrency: int = 5
) -> BatchPaymentResult:
    """
    Execute multiple payments in batch.

    Args:
        requests: List of payment requests to execute
        concurrency: Maximum number of concurrent executions (default 5)

    Returns:
        BatchPaymentResult containing status of all payments
    """
    return await self._batch_processor.process(requests, concurrency)

sync_transaction async

sync_transaction(entry_id)

Synchronize a ledger entry with Algorand indexer (confirmed tx).

Source code in python/src/algopay/client.py

async def sync_transaction(self, entry_id: str) -> LedgerEntry:
    """Synchronize a ledger entry with Algorand indexer (confirmed tx)."""
    entry = await self._ledger.get(entry_id)
    if not entry:
        raise ValidationError(f"Ledger entry not found: {entry_id}")

    tx_id = entry.tx_hash or entry.metadata.get("transaction_id")
    if not tx_id:
        raise ValidationError("Ledger entry has no transaction id / tx_hash to sync")

    tx = self._chain.transaction_by_id(tx_id)
    if not tx:
        raise PaymentError(f"Transaction not found on indexer: {tx_id}")

    new_status = LedgerEntryStatus.COMPLETED
    await self._ledger.update_status(
        entry.id,
        new_status,
        tx_hash=tx_id,
        metadata_updates={
            "last_synced": datetime.now(timezone.utc).isoformat(),
            "indexer_confirmed": True,
        },
    )

    updated = await self._ledger.get(entry.id)
    return updated  # type: ignore

add_budget_guard async

add_budget_guard(wallet_id, daily_limit=None, hourly_limit=None, total_limit=None, name='budget')

Add a budget guard to a wallet.

Enforce spending limits over time periods (Atomic & Reliable).

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required
daily_limit	str | Decimal | None	Max spend per 24h	None
hourly_limit	str | Decimal | None	Max spend per 1h	None
total_limit	str | Decimal | None	Max total spend (lifetime)	None
name	str	Custom name for the guard	'budget'

Source code in python/src/algopay/client.py

async def add_budget_guard(
    self,
    wallet_id: str,
    daily_limit: str | Decimal | None = None,
    hourly_limit: str | Decimal | None = None,
    total_limit: str | Decimal | None = None,
    name: str = "budget",
) -> None:
    """
    Add a budget guard to a wallet.

    Enforce spending limits over time periods (Atomic & Reliable).

    Args:
        wallet_id: Target wallet ID
        daily_limit: Max spend per 24h
        hourly_limit: Max spend per 1h
        total_limit: Max total spend (lifetime)
        name: Custom name for the guard
    """
    from algopay.guards.budget import BudgetGuard

    d_limit = Decimal(str(daily_limit)) if daily_limit else None
    h_limit = Decimal(str(hourly_limit)) if hourly_limit else None
    t_limit = Decimal(str(total_limit)) if total_limit else None

    guard = BudgetGuard(
        daily_limit=d_limit, hourly_limit=h_limit, total_limit=t_limit, name=name
    )
    await self._guard_manager.add_guard(wallet_id, guard)

add_budget_guard_for_set async

add_budget_guard_for_set(wallet_set_id, daily_limit=None, hourly_limit=None, total_limit=None, name='budget')

Add a budget guard to a wallet set (applies to ALL wallets in the set).

Parameters:

Name	Type	Description	Default
wallet_set_id	str	Target wallet set ID	required
daily_limit	str | Decimal | None	Max spend per 24h	None
hourly_limit	str | Decimal | None	Max spend per 1h	None
total_limit	str | Decimal | None	Max total spend (lifetime)	None
name	str	Custom name for the guard	'budget'

Source code in python/src/algopay/client.py

async def add_budget_guard_for_set(
    self,
    wallet_set_id: str,
    daily_limit: str | Decimal | None = None,
    hourly_limit: str | Decimal | None = None,
    total_limit: str | Decimal | None = None,
    name: str = "budget",
) -> None:
    """
    Add a budget guard to a wallet set (applies to ALL wallets in the set).

    Args:
        wallet_set_id: Target wallet set ID
        daily_limit: Max spend per 24h
        hourly_limit: Max spend per 1h
        total_limit: Max total spend (lifetime)
        name: Custom name for the guard
    """
    from algopay.guards.budget import BudgetGuard

    d_limit = Decimal(str(daily_limit)) if daily_limit else None
    h_limit = Decimal(str(hourly_limit)) if hourly_limit else None
    t_limit = Decimal(str(total_limit)) if total_limit else None

    guard = BudgetGuard(
        daily_limit=d_limit, hourly_limit=h_limit, total_limit=t_limit, name=name
    )
    await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

add_single_tx_guard async

add_single_tx_guard(wallet_id, max_amount, min_amount=None, name='single_tx')

Add a Single Transaction Limit guard.

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required
max_amount	str | Decimal	Max amount per transaction	required
min_amount	str | Decimal | None	Min amount per transaction	None
name	str	Guard name	'single_tx'

Source code in python/src/algopay/client.py

async def add_single_tx_guard(
    self,
    wallet_id: str,
    max_amount: str | Decimal,
    min_amount: str | Decimal | None = None,
    name: str = "single_tx",
) -> None:
    """
    Add a Single Transaction Limit guard.

    Args:
        wallet_id: Target wallet ID
        max_amount: Max amount per transaction
        min_amount: Min amount per transaction
        name: Guard name
    """
    from algopay.guards.single_tx import SingleTxGuard

    guard = SingleTxGuard(
        max_amount=Decimal(str(max_amount)),
        min_amount=Decimal(str(min_amount)) if min_amount else None,
        name=name,
    )
    await self._guard_manager.add_guard(wallet_id, guard)

add_recipient_guard async

add_recipient_guard(wallet_id, mode='whitelist', addresses=None, patterns=None, domains=None, name='recipient')

Add a Recipient Access Control guard.

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required
mode	str	'whitelist' (allow specific) or 'blacklist' (block specific)	'whitelist'
addresses	list[str] | None	List of allowed/blocked addresses	None
patterns	list[str] | None	List of regex patterns	None
domains	list[str] | None	List of allowed/blocked domains (for x402/URLs)	None
name	str	Guard name	'recipient'

Source code in python/src/algopay/client.py

async def add_recipient_guard(
    self,
    wallet_id: str,
    mode: str = "whitelist",
    addresses: list[str] | None = None,
    patterns: list[str] | None = None,
    domains: list[str] | None = None,
    name: str = "recipient",
) -> None:
    """
    Add a Recipient Access Control guard.

    Args:
        wallet_id: Target wallet ID
        mode: 'whitelist' (allow specific) or 'blacklist' (block specific)
        addresses: List of allowed/blocked addresses
        patterns: List of regex patterns
        domains: List of allowed/blocked domains (for x402/URLs)
        name: Guard name
    """
    from algopay.guards.recipient import RecipientGuard

    guard = RecipientGuard(
        mode=mode, addresses=addresses, patterns=patterns, domains=domains, name=name
    )
    await self._guard_manager.add_guard(wallet_id, guard)

add_rate_limit_guard async

add_rate_limit_guard(wallet_id, max_per_minute=None, max_per_hour=None, max_per_day=None, name='rate_limit')

Add a rate limit guard to a wallet.

Limit number of transactions per time window.

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required
max_per_minute	int | None	Max txs per minute	None
max_per_hour	int | None	Max txs per hour	None
max_per_day	int | None	Max txs per day	None
name	str	Custom name for the guard	'rate_limit'

Source code in python/src/algopay/client.py

async def add_rate_limit_guard(
    self,
    wallet_id: str,
    max_per_minute: int | None = None,
    max_per_hour: int | None = None,
    max_per_day: int | None = None,
    name: str = "rate_limit",
) -> None:
    """
    Add a rate limit guard to a wallet.

    Limit number of transactions per time window.

    Args:
        wallet_id: Target wallet ID
        max_per_minute: Max txs per minute
        max_per_hour: Max txs per hour
        max_per_day: Max txs per day
        name: Custom name for the guard
    """
    from algopay.guards.rate_limit import RateLimitGuard

    guard = RateLimitGuard(
        max_per_minute=max_per_minute,
        max_per_hour=max_per_hour,
        max_per_day=max_per_day,
        name=name,
    )
    await self._guard_manager.add_guard(wallet_id, guard)

add_confirm_guard async

add_confirm_guard(wallet_id, threshold=None, always_confirm=False, name='confirm')

Add a confirmation guard to a wallet (Human-in-the-Loop).

Payments above the threshold require explicit confirmation via callback or external handling (e.g., webhook approval).

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required
threshold	str | Decimal | None	Amount above which confirmation is required	None
always_confirm	bool	If True, require confirmation for ALL payments	False
name	str	Custom name for the guard	'confirm'

Source code in python/src/algopay/client.py

async def add_confirm_guard(
    self,
    wallet_id: str,
    threshold: str | Decimal | None = None,
    always_confirm: bool = False,
    name: str = "confirm",
) -> None:
    """
    Add a confirmation guard to a wallet (Human-in-the-Loop).

    Payments above the threshold require explicit confirmation via callback
    or external handling (e.g., webhook approval).

    Args:
        wallet_id: Target wallet ID
        threshold: Amount above which confirmation is required
        always_confirm: If True, require confirmation for ALL payments
        name: Custom name for the guard
    """
    from algopay.guards.confirm import ConfirmGuard

    t_threshold = Decimal(str(threshold)) if threshold else None

    guard = ConfirmGuard(threshold=t_threshold, always_confirm=always_confirm, name=name)
    await self._guard_manager.add_guard(wallet_id, guard)

add_confirm_guard_for_set async

add_confirm_guard_for_set(wallet_set_id, threshold=None, always_confirm=False, name='confirm')

Add a confirmation guard to a wallet set (applies to ALL wallets in the set).

Parameters:

Name	Type	Description	Default
wallet_set_id	str	Target wallet set ID	required
threshold	str | Decimal | None	Amount above which confirmation is required	None
always_confirm	bool	If True, require confirmation for ALL payments	False
name	str	Custom name for the guard	'confirm'

Source code in python/src/algopay/client.py

async def add_confirm_guard_for_set(
    self,
    wallet_set_id: str,
    threshold: str | Decimal | None = None,
    always_confirm: bool = False,
    name: str = "confirm",
) -> None:
    """
    Add a confirmation guard to a wallet set (applies to ALL wallets in the set).

    Args:
        wallet_set_id: Target wallet set ID
        threshold: Amount above which confirmation is required
        always_confirm: If True, require confirmation for ALL payments
        name: Custom name for the guard
    """
    from algopay.guards.confirm import ConfirmGuard

    t_threshold = Decimal(str(threshold)) if threshold else None

    guard = ConfirmGuard(threshold=t_threshold, always_confirm=always_confirm, name=name)
    await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

add_justification_guard async

add_justification_guard(wallet_id, min_length=1, name='justification')

Require a non-empty payment purpose (justification), Locus-style.

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required
min_length	int	Minimum stripped length of purpose (default: 1)	1
name	str	Guard name	'justification'

Source code in python/src/algopay/client.py

async def add_justification_guard(
    self,
    wallet_id: str,
    min_length: int = 1,
    name: str = "justification",
) -> None:
    """
    Require a non-empty payment ``purpose`` (justification), Locus-style.

    Args:
        wallet_id: Target wallet ID
        min_length: Minimum stripped length of ``purpose`` (default: 1)
        name: Guard name
    """
    from algopay.guards.justification import JustificationGuard

    guard = JustificationGuard(min_length=min_length, name=name)
    await self._guard_manager.add_guard(wallet_id, guard)

add_justification_guard_for_set async

add_justification_guard_for_set(wallet_set_id, min_length=1, name='justification')

Require justification for every wallet in the set.

Source code in python/src/algopay/client.py

async def add_justification_guard_for_set(
    self,
    wallet_set_id: str,
    min_length: int = 1,
    name: str = "justification",
) -> None:
    """Require justification for every wallet in the set."""
    from algopay.guards.justification import JustificationGuard

    guard = JustificationGuard(min_length=min_length, name=name)
    await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

add_rate_limit_guard_for_set async

add_rate_limit_guard_for_set(wallet_set_id, max_per_minute=None, max_per_hour=None, max_per_day=None, name='rate_limit')

Add a rate limit guard to a wallet set (applies to ALL wallets in the set).

Parameters:

Name	Type	Description	Default
wallet_set_id	str	Target wallet set ID	required
max_per_minute	int | None	Max txs per minute	None
max_per_hour	int | None	Max txs per hour	None
max_per_day	int | None	Max txs per day	None
name	str	Custom name for the guard	'rate_limit'

Source code in python/src/algopay/client.py

async def add_rate_limit_guard_for_set(
    self,
    wallet_set_id: str,
    max_per_minute: int | None = None,
    max_per_hour: int | None = None,
    max_per_day: int | None = None,
    name: str = "rate_limit",
) -> None:
    """
    Add a rate limit guard to a wallet set (applies to ALL wallets in the set).

    Args:
        wallet_set_id: Target wallet set ID
        max_per_minute: Max txs per minute
        max_per_hour: Max txs per hour
        max_per_day: Max txs per day
        name: Custom name for the guard
    """
    from algopay.guards.rate_limit import RateLimitGuard

    guard = RateLimitGuard(
        max_per_minute=max_per_minute,
        max_per_hour=max_per_hour,
        max_per_day=max_per_day,
        name=name,
    )
    await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

add_recipient_guard_for_set async

add_recipient_guard_for_set(wallet_set_id, mode='whitelist', addresses=None, patterns=None, domains=None, name='recipient')

Add a Recipient Access Control guard to a wallet set.

Parameters:

Name	Type	Description	Default
wallet_set_id	str	Target wallet set ID	required
mode	str	'whitelist' (allow specific) or 'blacklist' (block specific)	'whitelist'
addresses	list[str] | None	List of allowed/blocked addresses	None
patterns	list[str] | None	List of regex patterns	None
domains	list[str] | None	List of allowed/blocked domains (for x402/URLs)	None
name	str	Guard name	'recipient'

Source code in python/src/algopay/client.py

async def add_recipient_guard_for_set(
    self,
    wallet_set_id: str,
    mode: str = "whitelist",
    addresses: list[str] | None = None,
    patterns: list[str] | None = None,
    domains: list[str] | None = None,
    name: str = "recipient",
) -> None:
    """
    Add a Recipient Access Control guard to a wallet set.

    Args:
        wallet_set_id: Target wallet set ID
        mode: 'whitelist' (allow specific) or 'blacklist' (block specific)
        addresses: List of allowed/blocked addresses
        patterns: List of regex patterns
        domains: List of allowed/blocked domains (for x402/URLs)
        name: Guard name
    """
    from algopay.guards.recipient import RecipientGuard

    guard = RecipientGuard(
        mode=mode, addresses=addresses, patterns=patterns, domains=domains, name=name
    )
    await self._guard_manager.add_guard_for_set(wallet_set_id, guard)

list_guards async

list_guards(wallet_id)

List all guard names registered for a wallet.

Parameters:

Name	Type	Description	Default
wallet_id	str	Target wallet ID	required

Returns:

Type	Description
list[str]	List of guard names

Source code in python/src/algopay/client.py

async def list_guards(self, wallet_id: str) -> list[str]:
    """
    List all guard names registered for a wallet.

    Args:
        wallet_id: Target wallet ID

    Returns:
        List of guard names
    """
    return await self._guard_manager.list_wallet_guard_names(wallet_id)

list_guards_for_set async

list_guards_for_set(wallet_set_id)

List all guard names registered for a wallet set.

Parameters:

Name	Type	Description	Default
wallet_set_id	str	Target wallet set ID	required

Returns:

Type	Description
list[str]	List of guard names

Source code in python/src/algopay/client.py

async def list_guards_for_set(self, wallet_set_id: str) -> list[str]:
    """
    List all guard names registered for a wallet set.

    Args:
        wallet_set_id: Target wallet set ID

    Returns:
        List of guard names
    """
    return await self._guard_manager.list_wallet_set_guard_names(wallet_set_id)

Config

SDK configuration for Algorand.

Source code in python/src/algopay/core/config.py

@dataclass(frozen=True)
class Config:
    """SDK configuration for Algorand."""

    network: Network = Network.ALGORAND_TESTNET
    algod_url: str = ""
    indexer_url: str = ""
    usdc_asa_id: int = 0
    storage_backend: str = "memory"
    redis_url: str | None = None
    log_level: str = "INFO"
    http_timeout: float = 30.0
    request_timeout: float = 30.0
    transaction_poll_interval: float = 2.0
    transaction_poll_timeout: float = 120.0
    env: str = "development"
    default_wallet_id: str | None = None

    @property
    def network_caip2(self) -> str:
        return self.network.to_caip2()

    @classmethod
    def from_env(cls, **overrides: Any) -> Config:
        net_raw = overrides.get("network") or _env("ALGOPAY_NETWORK", "algorand-testnet")
        network = net_raw if isinstance(net_raw, Network) else Network.from_string(str(net_raw))

        usdc = overrides.get("usdc_asa_id")
        if usdc is None and _env("ALGOPAY_USDC_ASA_ID"):
            usdc = int(_env("ALGOPAY_USDC_ASA_ID", "0") or 0)
        if not usdc:
            usdc = network.usdc_asa_id()

        algod = overrides.get("algod_url") or _env("ALGOD_URL") or _env("ALGOPAY_ALGOD_URL") or ""
        indexer = overrides.get("indexer_url") or _env("INDEXER_URL") or _env("ALGOPAY_INDEXER_URL") or ""

        if not algod or not indexer:
            if network == Network.ALGORAND_MAINNET:
                algod = algod or "https://mainnet-api.algonode.cloud"
                indexer = indexer or "https://mainnet-idx.algonode.cloud"
            else:
                algod = algod or "https://testnet-api.algonode.cloud"
                indexer = indexer or "https://testnet-idx.algonode.cloud"

        return cls(
            network=network,
            algod_url=str(algod),
            indexer_url=str(indexer),
            usdc_asa_id=int(usdc),
            storage_backend=str(overrides.get("storage_backend") or _env("ALGOPAY_STORAGE_BACKEND", "memory")),
            redis_url=overrides.get("redis_url") or _env("ALGOPAY_REDIS_URL"),
            log_level=str(overrides.get("log_level") or _env("ALGOPAY_LOG_LEVEL", "INFO")),
            default_wallet_id=overrides.get("default_wallet_id") or _env("ALGOPAY_DEFAULT_WALLET"),
            env=str(overrides.get("env") or _env("ALGOPAY_ENV", "development")),
        )

    def with_updates(self, **updates: Any) -> Config:
        fields = {
            "network": self.network,
            "algod_url": self.algod_url,
            "indexer_url": self.indexer_url,
            "usdc_asa_id": self.usdc_asa_id,
            "storage_backend": self.storage_backend,
            "redis_url": self.redis_url,
            "log_level": self.log_level,
            "default_wallet_id": self.default_wallet_id,
            "env": self.env,
        }
        fields.update(updates)
        return Config(**fields)

WalletService

Source code in python/src/algopay/wallet/service.py

class WalletService:
    def __init__(
        self,
        config: Config,
        chain: AlgorandClient | None = None,
        repository: WalletRepository | None = None,
    ) -> None:
        self._config = config
        self._chain = chain or AlgorandClient(config)
        self._repo = repository or WalletRepository()
        self._wallet_cache: dict[str, WalletInfo] = {}

    @property
    def repository(self) -> WalletRepository:
        return self._repo

    def _record_to_info(self, rec: WalletRecord) -> WalletInfo:
        return WalletInfo(
            id=rec.id,
            address=rec.address,
            blockchain=rec.network_caip2,
            state=rec.state,
            wallet_set_id=rec.wallet_set_id,
            account_type=AccountType.EOA,
            name=rec.name,
            create_date=rec.created_at,
        )

    # --- wallet sets ---
    def create_wallet_set(self, name: str | None = None) -> WalletSetInfo:
        return self._repo.create_wallet_set(name or "AlgoPay Wallet Set")

    def list_wallet_sets(self) -> list[WalletSetInfo]:
        return self._repo.list_wallet_sets()

    def get_wallet_set(self, wallet_set_id: str) -> WalletSetInfo:
        ws = self._repo.get_wallet_set(wallet_set_id)
        if not ws:
            raise WalletError(f"Wallet set not found: {wallet_set_id}")
        return ws

    # --- wallets ---
    def create_wallet(
        self,
        wallet_set_id: str,
        blockchain: Network | str | None = None,
        account_type: AccountType = AccountType.EOA,
        name: str | None = None,
    ) -> WalletInfo:
        net = self._config.network if blockchain is None else (
            blockchain if isinstance(blockchain, Network) else Network.from_string(str(blockchain))
        )
        rec = self._repo.create_wallet(wallet_set_id, net.to_caip2(), name=name)
        info = self._record_to_info(rec)
        self._wallet_cache[info.id] = info
        return info

    def create_wallets(
        self,
        wallet_set_id: str,
        count: int,
        blockchain: Network | str | None = None,
        account_type: AccountType = AccountType.EOA,
    ) -> list[WalletInfo]:
        return [self.create_wallet(wallet_set_id, blockchain, account_type) for _ in range(count)]

    def get_wallet(self, wallet_id: str) -> WalletInfo:
        if wallet_id in self._wallet_cache:
            return self._wallet_cache[wallet_id]
        rec = self._repo.get_wallet(wallet_id)
        if not rec:
            raise WalletError(f"Wallet not found: {wallet_id}", wallet_id=wallet_id)
        info = self._record_to_info(rec)
        self._wallet_cache[wallet_id] = info
        return info

    def list_wallets(self, wallet_set_id: str | None = None, blockchain: object | None = None) -> list[WalletInfo]:
        recs = self._repo.list_wallets(wallet_set_id)
        return [self._record_to_info(r) for r in recs]

    def get_private_key(self, wallet_id: str) -> bytes:
        return self._repo.get_private_key(wallet_id)

    def _private_key_b64_for_signing(self, wallet_id: str) -> str:
        """algosdk transaction.sign expects a base64-encoded key string."""
        raw = self.get_private_key(wallet_id)
        return base64.b64encode(raw).decode()

    # --- balances ---
    def _micro_algo(self, wallet_id: str) -> int:
        w = self.get_wallet(wallet_id)
        info = self._chain.account_info(w.address)
        return int(info.get("amount", 0))

    def get_balances(self, wallet_id: str) -> list[Balance]:
        w = self.get_wallet(wallet_id)
        info = self._chain.account_info(w.address)
        assets = info.get("assets", []) or []
        balances: list[Balance] = []
        for a in assets:
            aid = a["asset-id"]
            amt = Decimal(a.get("amount", 0))
            if aid == self._config.usdc_asa_id:
                balances.append(
                    Balance(
                        amount=amt / Decimal(10**6),
                        token=TokenInfo(
                            id=str(aid),
                            blockchain=w.blockchain,
                            symbol="USDC",
                            name="USDC",
                            decimals=6,
                        ),
                    )
                )
        return balances

    def get_usdc_balance(self, wallet_id: str) -> Balance | None:
        for b in self.get_balances(wallet_id):
            if b.token.symbol == "USDC":
                return b
        return None

    def get_usdc_balance_amount(self, wallet_id: str) -> Decimal:
        b = self.get_usdc_balance(wallet_id)
        return b.amount if b else Decimal("0")

    def ensure_sufficient_balance(self, wallet_id: str, required: Decimal) -> Balance:
        b = self.get_usdc_balance(wallet_id)
        cur = b.amount if b else Decimal("0")
        if cur < required:
            raise InsufficientBalanceError(
                "Insufficient USDC balance",
                current_balance=cur,
                required_amount=required,
                wallet_id=wallet_id,
            )
        return b  # type: ignore[return-value]

    def opt_in_usdc(self, wallet_id: str, fee_level: FeeLevel = FeeLevel.MEDIUM) -> str:
        """Submit 0-amount axfer to opt in to configured USDC ASA. Returns txid."""
        w = self.get_wallet(wallet_id)
        sk_b64 = self._private_key_b64_for_signing(wallet_id)
        sp = self._chain.suggested_params()
        mf = sp.min_fee or 1000
        fee = mf * _fee_level_multiplier(fee_level)
        nsp = transaction.SuggestedParams(
            fee=fee,
            flat_fee=True,
            first=sp.first,
            last=sp.last,
            gh=sp.gh,
            gen=sp.gen,
            min_fee=sp.min_fee,
        )
        txn = transaction.AssetTransferTxn(
            sender=w.address,
            sp=nsp,
            receiver=w.address,
            amt=0,
            index=self._config.usdc_asa_id,
        )
        stxn = txn.sign(sk_b64)
        txid = self._chain.send_transaction(encoding.msgpack_encode(stxn))
        return txid

    def transfer(
        self,
        wallet_id: str,
        destination_address: str,
        amount: Decimal | str,
        fee_level: FeeLevel = FeeLevel.MEDIUM,
        check_balance: bool = True,
        wait_for_completion: bool = False,
        timeout_seconds: float | None = None,
        idempotency_key: str | None = None,
    ) -> TransferResult:
        amount_decimal = Decimal(str(amount))
        if check_balance:
            self.ensure_sufficient_balance(wallet_id, amount_decimal)

        w = self.get_wallet(wallet_id)
        sk_b64 = self._private_key_b64_for_signing(wallet_id)
        sp = self._chain.suggested_params()
        mf = sp.min_fee or 1000
        fee = mf * _fee_level_multiplier(fee_level)
        nsp = transaction.SuggestedParams(
            fee=fee,
            flat_fee=True,
            first=sp.first,
            last=sp.last,
            gh=sp.gh,
            gen=sp.gen,
            min_fee=sp.min_fee,
        )
        micro = int(amount_decimal * Decimal(10**6))
        txn = transaction.AssetTransferTxn(
            sender=w.address,
            sp=nsp,
            receiver=destination_address,
            amt=micro,
            index=self._config.usdc_asa_id,
        )
        stxn = txn.sign(sk_b64)
        txid = self._chain.send_transaction(encoding.msgpack_encode(stxn))
        tx_info = TransactionInfo(
            id=txid,
            state=TransactionState.PENDING,
            blockchain=w.blockchain,
            tx_hash=txid,
            wallet_id=wallet_id,
            source_address=w.address,
            destination_address=destination_address,
        )
        if wait_for_completion:
            timeout = timeout_seconds or self._config.transaction_poll_timeout
            tx_info = self._wait_for_confirmation(txid, timeout)
        return TransferResult(success=True, transaction=tx_info, tx_hash=txid)

    def _wait_for_confirmation(self, txid: str, timeout_seconds: float) -> TransactionInfo:
        deadline = time.time() + timeout_seconds
        poll = self._config.transaction_poll_interval
        while time.time() < deadline:
            try:
                pending = self._chain.pending_transaction_info(txid)
                if "confirmed-round" in pending:
                    return TransactionInfo(
                        id=txid,
                        state=TransactionState.COMPLETE,
                        tx_hash=txid,
                    )
                if pending.get("pool-error"):
                    return TransactionInfo(
                        id=txid,
                        state=TransactionState.FAILED,
                        tx_hash=txid,
                        error_reason=str(pending.get("pool-error")),
                    )
            except Exception:
                pass
            time.sleep(poll)
        return TransactionInfo(id=txid, state=TransactionState.PENDING, tx_hash=txid)

    def list_transactions(self, wallet_id: str | None = None, blockchain: object | None = None) -> list[TransactionInfo]:
        """Best-effort: recent USDC axfers from indexer for wallet address."""
        if not wallet_id:
            return []
        w = self.get_wallet(wallet_id)
        try:
            resp = self._chain.indexer.search_asset_transactions(
                asset_id=self._config.usdc_asa_id,
                address=w.address,
                limit=20,
            )
        except Exception:
            return []
        txs = resp.get("transactions", []) or []
        out: list[TransactionInfo] = []
        for t in txs:
            tid = t.get("id")
            out.append(
                TransactionInfo(
                    id=tid,
                    state=TransactionState.COMPLETE,
                    tx_hash=tid,
                    wallet_id=wallet_id,
                )
            )
        return out

    def create_agent_wallet(
        self,
        agent_name: str,
        blockchain: Network | str | None = None,
        count: int = 1,
    ) -> tuple[WalletSetInfo, WalletInfo | list[WalletInfo]]:
        name = f"agent-{agent_name}"
        existing = next((s for s in self.list_wallet_sets() if s.name == name), None)
        ws = existing or self.create_wallet_set(name)
        if count == 1:
            return ws, self.create_wallet(ws.id, blockchain)
        return ws, self.create_wallets(ws.id, count, blockchain)

opt_in_usdc

opt_in_usdc(wallet_id, fee_level=FeeLevel.MEDIUM)

Submit 0-amount axfer to opt in to configured USDC ASA. Returns txid.

Source code in python/src/algopay/wallet/service.py

def opt_in_usdc(self, wallet_id: str, fee_level: FeeLevel = FeeLevel.MEDIUM) -> str:
    """Submit 0-amount axfer to opt in to configured USDC ASA. Returns txid."""
    w = self.get_wallet(wallet_id)
    sk_b64 = self._private_key_b64_for_signing(wallet_id)
    sp = self._chain.suggested_params()
    mf = sp.min_fee or 1000
    fee = mf * _fee_level_multiplier(fee_level)
    nsp = transaction.SuggestedParams(
        fee=fee,
        flat_fee=True,
        first=sp.first,
        last=sp.last,
        gh=sp.gh,
        gen=sp.gen,
        min_fee=sp.min_fee,
    )
    txn = transaction.AssetTransferTxn(
        sender=w.address,
        sp=nsp,
        receiver=w.address,
        amt=0,
        index=self._config.usdc_asa_id,
    )
    stxn = txn.sign(sk_b64)
    txid = self._chain.send_transaction(encoding.msgpack_encode(stxn))
    return txid

list_transactions

list_transactions(wallet_id=None, blockchain=None)

Best-effort: recent USDC axfers from indexer for wallet address.

Source code in python/src/algopay/wallet/service.py

def list_transactions(self, wallet_id: str | None = None, blockchain: object | None = None) -> list[TransactionInfo]:
    """Best-effort: recent USDC axfers from indexer for wallet address."""
    if not wallet_id:
        return []
    w = self.get_wallet(wallet_id)
    try:
        resp = self._chain.indexer.search_asset_transactions(
            asset_id=self._config.usdc_asa_id,
            address=w.address,
            limit=20,
        )
    except Exception:
        return []
    txs = resp.get("transactions", []) or []
    out: list[TransactionInfo] = []
    for t in txs:
        tid = t.get("id")
        out.append(
            TransactionInfo(
                id=tid,
                state=TransactionState.COMPLETE,
                tx_hash=tid,
                wallet_id=wallet_id,
            )
        )
    return out

Types

Bases: str, Enum

Supported Algorand networks (CAIP-2 style identifier values).

Source code in python/src/algopay/core/types.py

class Network(str, Enum):
    """Supported Algorand networks (CAIP-2 style identifier values)."""

    ALGORAND_MAINNET = "algorand-mainnet"
    ALGORAND_TESTNET = "algorand-testnet"

    @classmethod
    def from_string(cls, value: str) -> Network:
        v = value.strip().lower().replace("_", "-")
        if v.startswith("algorand:"):
            from algopay.core.constants import ALGORAND_MAINNET_CAIP2, ALGORAND_TESTNET_CAIP2

            if value == ALGORAND_MAINNET_CAIP2:
                return cls.ALGORAND_MAINNET
            if value == ALGORAND_TESTNET_CAIP2:
                return cls.ALGORAND_TESTNET
            raise ValueError(f"Unknown Algorand CAIP-2 network: {value}")
        for m in cls:
            if m.value == v:
                return m
        raise ValueError(f"Unknown network: {value}. Use algorand-mainnet or algorand-testnet.")

    def to_caip2(self) -> str:
        from algopay.core.constants import ALGORAND_MAINNET_CAIP2, ALGORAND_TESTNET_CAIP2

        if self == Network.ALGORAND_MAINNET:
            return ALGORAND_MAINNET_CAIP2
        return ALGORAND_TESTNET_CAIP2

    def usdc_asa_id(self) -> int:
        from algopay.core.constants import USDC_MAINNET_ASA_ID, USDC_TESTNET_ASA_ID

        return USDC_MAINNET_ASA_ID if self == Network.ALGORAND_MAINNET else USDC_TESTNET_ASA_ID

Bases: str, Enum

Source code in python/src/algopay/core/types.py

class PaymentMethod(str, Enum):
    X402 = "x402"
    TRANSFER = "transfer"

Bases: str, Enum

Source code in python/src/algopay/core/types.py

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    BLOCKED = "blocked"

Bases: str, Enum

Source code in python/src/algopay/core/types.py

class PaymentIntentStatus(str, Enum):
    REQUIRES_CONFIRMATION = "requires_confirmation"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    CANCELED = "canceled"
    FAILED = "failed"

Bases: str, Enum

Source code in python/src/algopay/core/types.py

class FeeLevel(str, Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"

Source code in python/src/algopay/core/types.py

@dataclass
class WalletInfo:
    id: str
    address: str
    blockchain: str
    state: WalletState
    wallet_set_id: str
    account_type: AccountType = AccountType.EOA
    name: str | None = None
    create_date: datetime | None = None
    update_date: datetime | None = None

Source code in python/src/algopay/core/types.py

@dataclass
class WalletSetInfo:
    id: str
    name: str | None
    create_date: datetime | None = None
    update_date: datetime | None = None

Source code in python/src/algopay/core/types.py

@dataclass
class TransactionInfo:
    id: str
    state: TransactionState
    blockchain: str | None = None
    tx_hash: str | None = None
    wallet_id: str | None = None
    source_address: str | None = None
    destination_address: str | None = None
    error_reason: str | None = None

    def is_terminal(self) -> bool:
        return self.state in (
            TransactionState.COMPLETE,
            TransactionState.FAILED,
        )

    def is_successful(self) -> bool:
        return self.state == TransactionState.COMPLETE

Source code in python/src/algopay/core/types.py

@dataclass
class PaymentRequest:
    wallet_id: str
    recipient: str
    amount: Decimal
    purpose: str | None = None
    idempotency_key: str | None = None
    destination_chain: Network | str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if self.amount <= 0:
            raise ValueError("Amount must be positive")
        if not self.recipient:
            raise ValueError("Recipient is required")
        if not self.wallet_id:
            raise ValueError("wallet_id is required")

Source code in python/src/algopay/core/types.py

@dataclass
class PaymentIntent:
    id: str
    wallet_id: str
    recipient: str
    amount: Decimal
    currency: str
    status: PaymentIntentStatus
    created_at: datetime
    expires_at: datetime | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    client_secret: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "wallet_id": self.wallet_id,
            "recipient": self.recipient,
            "amount": str(self.amount),
            "currency": self.currency,
            "status": self.status.value,
            "created_at": self.created_at.isoformat(),
            "expires_at": self.expires_at.isoformat() if self.expires_at else None,
            "metadata": self.metadata,
            "client_secret": self.client_secret,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> PaymentIntent:
        return cls(
            id=data["id"],
            wallet_id=data["wallet_id"],
            recipient=data["recipient"],
            amount=Decimal(data["amount"]),
            currency=data["currency"],
            status=PaymentIntentStatus(data["status"]),
            created_at=datetime.fromisoformat(data["created_at"]),
            expires_at=datetime.fromisoformat(data["expires_at"]) if data.get("expires_at") else None,
            metadata=data.get("metadata", {}),
            client_secret=data.get("client_secret"),
        )

Source code in python/src/algopay/core/types.py

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str | None
    blockchain_tx: str | None
    amount: Decimal
    recipient: str
    method: PaymentMethod
    status: PaymentStatus
    guards_passed: list[str] = field(default_factory=list)
    error: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    resource_data: Any = None

Source code in python/src/algopay/core/types.py

@dataclass
class SimulationResult:
    would_succeed: bool
    route: PaymentMethod
    guards_that_would_pass: list[str] = field(default_factory=list)
    guards_that_would_fail: list[str] = field(default_factory=list)
    estimated_fee: Decimal | None = None
    reason: str | None = None

Source code in python/src/algopay/core/types.py

@dataclass
class BatchPaymentResult:
    total_count: int
    success_count: int
    failed_count: int
    results: list[PaymentResult]
    transaction_ids: list[str] = field(default_factory=list)

Exceptions

Bases: Exception

Source code in python/src/algopay/core/exceptions.py

class AlgoPayError(Exception):
    def __init__(self, message: str, details: dict[str, Any] | None = None) -> None:
        super().__init__(message)
        self.message = message
        self.details = details or {}

    def __str__(self) -> str:
        if self.details:
            return f"{self.message} | Details: {self.details}"
        return self.message

Bases: AlgoPayError

Source code in python/src/algopay/core/exceptions.py

class ConfigurationError(AlgoPayError):
    pass

Bases: AlgoPayError

Source code in python/src/algopay/core/exceptions.py

class WalletError(AlgoPayError):
    def __init__(
        self,
        message: str,
        wallet_id: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, details)
        self.wallet_id = wallet_id

Bases: AlgoPayError

Source code in python/src/algopay/core/exceptions.py

class PaymentError(AlgoPayError):
    def __init__(
        self,
        message: str,
        recipient: str | None = None,
        amount: Decimal | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, details)
        self.recipient = recipient
        self.amount = amount

Bases: PaymentError

Source code in python/src/algopay/core/exceptions.py

class GuardError(PaymentError):
    def __init__(
        self,
        message: str,
        guard_name: str,
        reason: str,
        recipient: str | None = None,
        amount: Decimal | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, recipient, amount, details)
        self.guard_name = guard_name
        self.reason = reason

    def __str__(self) -> str:
        return f"[{self.guard_name}] {self.reason}"

Bases: PaymentError

Source code in python/src/algopay/core/exceptions.py

class ProtocolError(PaymentError):
    def __init__(
        self,
        message: str,
        protocol: str = "unknown",
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, details=details)
        self.protocol = protocol

    def __str__(self) -> str:
        return f"[{self.protocol}] {self.message}"

Bases: AlgoPayError

Source code in python/src/algopay/core/exceptions.py

class ValidationError(AlgoPayError):
    pass

Bases: PaymentError

Source code in python/src/algopay/core/exceptions.py

class InsufficientBalanceError(PaymentError):
    def __init__(
        self,
        message: str,
        current_balance: Decimal,
        required_amount: Decimal,
        wallet_id: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, details=details, amount=required_amount)
        self.current_balance = current_balance
        self.required_amount = required_amount
        self.wallet_id = wallet_id
        self.shortfall = required_amount - current_balance

    def __str__(self) -> str:
        return (
            f"{self.message} | "
            f"Balance: {self.current_balance}, Required: {self.required_amount}, "
            f"Shortfall: {self.shortfall}"
        )

Bases: AlgoPayError

Source code in python/src/algopay/core/exceptions.py

class NetworkError(AlgoPayError):
    def __init__(
        self,
        message: str,
        status_code: int | None = None,
        url: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, details)
        self.status_code = status_code
        self.url = url

GuardManager

Manages guard registrations using StorageBackend.

Source code in python/src/algopay/guards/manager.py

class GuardManager:
    """Manages guard registrations using StorageBackend."""

    COLLECTION = "guard_registrations"

    def __init__(self, storage: StorageBackend) -> None:
        """
        Initialize GuardManager with storage backend.

        Args:
            storage: StorageBackend for persistence
        """
        self._storage = storage
        self._logger = get_logger("guards")

    def _make_key(self, scope_type: str, scope_id: str) -> str:
        """Make storage key."""
        return f"{scope_type}:{scope_id}"

    # ==================== Add Guards ====================

    async def add_guard(self, wallet_id: str, guard: Guard) -> GuardManager:
        """
        Add a guard for a wallet.

        Args:
            wallet_id: Wallet ID
            guard: Guard to add

        Returns:
            Self for chaining
        """
        key = self._make_key("wallet", wallet_id)

        # Get existing configs
        data = await self._storage.get(self.COLLECTION, key) or {"guards": []}

        # Create config from guard and add
        config = GuardConfig.from_guard(guard)
        data["guards"].append(config.to_dict())

        await self._storage.save(self.COLLECTION, key, data)
        return self

    async def add_guard_for_set(self, wallet_set_id: str, guard: Guard) -> GuardManager:
        """Add a guard for a wallet set."""
        key = self._make_key("wallet_set", wallet_set_id)

        data = await self._storage.get(self.COLLECTION, key) or {"guards": []}
        config = GuardConfig.from_guard(guard)
        data["guards"].append(config.to_dict())

        await self._storage.save(self.COLLECTION, key, data)
        return self

    # ==================== Remove Guards ====================

    async def remove_guard(self, wallet_id: str, guard_name: str) -> bool:
        """Remove a guard from a wallet."""
        key = self._make_key("wallet", wallet_id)
        data = await self._storage.get(self.COLLECTION, key)

        if not data:
            return False

        original_count = len(data.get("guards", []))
        data["guards"] = [g for g in data.get("guards", []) if g.get("name") != guard_name]

        if len(data["guards"]) < original_count:
            await self._storage.save(self.COLLECTION, key, data)
            return True
        return False

    async def remove_guard_from_set(self, wallet_set_id: str, guard_name: str) -> bool:
        """Remove a guard from a wallet set."""
        key = self._make_key("wallet_set", wallet_set_id)
        data = await self._storage.get(self.COLLECTION, key)

        if not data:
            return False

        original_count = len(data.get("guards", []))
        data["guards"] = [g for g in data.get("guards", []) if g.get("name") != guard_name]

        if len(data["guards"]) < original_count:
            await self._storage.save(self.COLLECTION, key, data)
            return True
        return False

    # ==================== Get Guards ====================

    async def get_wallet_guards(self, wallet_id: str) -> GuardChain:
        """Get guards for a wallet."""
        key = self._make_key("wallet", wallet_id)
        data = await self._storage.get(self.COLLECTION, key)

        chain = GuardChain()
        if data:
            for guard_data in data.get("guards", []):
                config = GuardConfig.from_dict(guard_data)
                guard = config.to_guard(self._storage)
                chain.add(guard)

        return chain

    async def get_wallet_set_guards(self, wallet_set_id: str) -> GuardChain:
        """Get guards for a wallet set."""
        key = self._make_key("wallet_set", wallet_set_id)
        data = await self._storage.get(self.COLLECTION, key)

        chain = GuardChain()
        if data:
            for guard_data in data.get("guards", []):
                config = GuardConfig.from_dict(guard_data)
                guard = config.to_guard(self._storage)
                chain.add(guard)

        return chain

    async def list_wallet_guard_names(self, wallet_id: str) -> list[str]:
        """List guard names for a wallet."""
        key = self._make_key("wallet", wallet_id)
        data = await self._storage.get(self.COLLECTION, key)

        if not data:
            return []
        return [g.get("name", "unnamed") for g in data.get("guards", [])]

    async def list_wallet_set_guard_names(self, wallet_set_id: str) -> list[str]:
        """List guard names for a wallet set."""
        key = self._make_key("wallet_set", wallet_set_id)
        data = await self._storage.get(self.COLLECTION, key)

        if not data:
            return []
        return [g.get("name", "unnamed") for g in data.get("guards", [])]

    # ==================== Combined Operations ====================

    async def get_guard_chain(
        self,
        wallet_id: str,
        wallet_set_id: str | None = None,
    ) -> GuardChain:
        """
        Get combined guard chain for a wallet.

        Merges guards from wallet set (if provided) and wallet.
        """
        combined = GuardChain()

        # Add wallet set guards first
        if wallet_set_id:
            set_chain = await self.get_wallet_set_guards(wallet_set_id)
            for guard in set_chain:
                combined.add(guard)

        # Add wallet-specific guards
        wallet_chain = await self.get_wallet_guards(wallet_id)
        for guard in wallet_chain:
            combined.add(guard)

        return combined

    async def check(self, context: PaymentContext) -> tuple[bool, str | None, list[str]]:
        """
        Check guards for a payment context.

        Returns:
            Tuple of (allowed, reason, passed_guards)
        """
        chain = await self.get_guard_chain(
            context.wallet_id,
            context.wallet_set_id,
        )

        if len(chain) == 0:
            return True, None, []

        self._logger.debug(f"Checking {len(chain)} guards for wallet={context.wallet_id}")

        result = await chain.check(context)
        passed = result.metadata.get("passed_guards", []) if result.metadata else []

        if not result.allowed:
            self._logger.warning(
                f"Payment BLOCKED by guard: {result.reason} (Wallet: {context.wallet_id})"
            )
        else:
            self._logger.debug(f"Guards passed: {passed}")

        return result.allowed, result.reason, passed

    async def record_spending(
        self,
        wallet_id: str,
        wallet_set_id: str | None,
        amount: Decimal,
        recipient: str,
        purpose: str | None,
    ) -> None:
        """Record spending in all relevant guards."""
        chain = await self.get_guard_chain(wallet_id, wallet_set_id)

        for guard in chain:
            # BudgetGuard uses record_spending
            if hasattr(guard, "record_spending"):
                await guard.record_spending(
                    amount=amount,
                    wallet_id=wallet_id,
                    recipient=recipient,
                    purpose=purpose,
                )
            # RateLimitGuard uses record_payment (now async with wallet_id)
            if hasattr(guard, "record_payment"):
                await guard.record_payment(wallet_id)

    async def clear_wallet_guards(self, wallet_id: str) -> None:
        """Clear all guards for a wallet."""
        key = self._make_key("wallet", wallet_id)
        await self._storage.delete(self.COLLECTION, key)

    async def clear_wallet_set_guards(self, wallet_set_id: str) -> None:
        """Clear all guards for a wallet set."""
        key = self._make_key("wallet_set", wallet_set_id)
        await self._storage.delete(self.COLLECTION, key)

__init__

__init__(storage)

Initialize GuardManager with storage backend.

Parameters:

Name	Type	Description	Default
storage	StorageBackend	StorageBackend for persistence	required

Source code in python/src/algopay/guards/manager.py

def __init__(self, storage: StorageBackend) -> None:
    """
    Initialize GuardManager with storage backend.

    Args:
        storage: StorageBackend for persistence
    """
    self._storage = storage
    self._logger = get_logger("guards")

add_guard async

add_guard(wallet_id, guard)

Add a guard for a wallet.

Parameters:

Name	Type	Description	Default
wallet_id	str	Wallet ID	required
guard	Guard	Guard to add	required

Returns:

Type	Description
GuardManager	Self for chaining

Source code in python/src/algopay/guards/manager.py

async def add_guard(self, wallet_id: str, guard: Guard) -> GuardManager:
    """
    Add a guard for a wallet.

    Args:
        wallet_id: Wallet ID
        guard: Guard to add

    Returns:
        Self for chaining
    """
    key = self._make_key("wallet", wallet_id)

    # Get existing configs
    data = await self._storage.get(self.COLLECTION, key) or {"guards": []}

    # Create config from guard and add
    config = GuardConfig.from_guard(guard)
    data["guards"].append(config.to_dict())

    await self._storage.save(self.COLLECTION, key, data)
    return self

add_guard_for_set async

add_guard_for_set(wallet_set_id, guard)

Add a guard for a wallet set.

Source code in python/src/algopay/guards/manager.py

async def add_guard_for_set(self, wallet_set_id: str, guard: Guard) -> GuardManager:
    """Add a guard for a wallet set."""
    key = self._make_key("wallet_set", wallet_set_id)

    data = await self._storage.get(self.COLLECTION, key) or {"guards": []}
    config = GuardConfig.from_guard(guard)
    data["guards"].append(config.to_dict())

    await self._storage.save(self.COLLECTION, key, data)
    return self

remove_guard async

remove_guard(wallet_id, guard_name)

Remove a guard from a wallet.

Source code in python/src/algopay/guards/manager.py

async def remove_guard(self, wallet_id: str, guard_name: str) -> bool:
    """Remove a guard from a wallet."""
    key = self._make_key("wallet", wallet_id)
    data = await self._storage.get(self.COLLECTION, key)

    if not data:
        return False

    original_count = len(data.get("guards", []))
    data["guards"] = [g for g in data.get("guards", []) if g.get("name") != guard_name]

    if len(data["guards"]) < original_count:
        await self._storage.save(self.COLLECTION, key, data)
        return True
    return False

remove_guard_from_set async

remove_guard_from_set(wallet_set_id, guard_name)

Remove a guard from a wallet set.

Source code in python/src/algopay/guards/manager.py

async def remove_guard_from_set(self, wallet_set_id: str, guard_name: str) -> bool:
    """Remove a guard from a wallet set."""
    key = self._make_key("wallet_set", wallet_set_id)
    data = await self._storage.get(self.COLLECTION, key)

    if not data:
        return False

    original_count = len(data.get("guards", []))
    data["guards"] = [g for g in data.get("guards", []) if g.get("name") != guard_name]

    if len(data["guards"]) < original_count:
        await self._storage.save(self.COLLECTION, key, data)
        return True
    return False

get_wallet_guards async

get_wallet_guards(wallet_id)

Get guards for a wallet.

Source code in python/src/algopay/guards/manager.py

async def get_wallet_guards(self, wallet_id: str) -> GuardChain:
    """Get guards for a wallet."""
    key = self._make_key("wallet", wallet_id)
    data = await self._storage.get(self.COLLECTION, key)

    chain = GuardChain()
    if data:
        for guard_data in data.get("guards", []):
            config = GuardConfig.from_dict(guard_data)
            guard = config.to_guard(self._storage)
            chain.add(guard)

    return chain

get_wallet_set_guards async

get_wallet_set_guards(wallet_set_id)

Get guards for a wallet set.

Source code in python/src/algopay/guards/manager.py

async def get_wallet_set_guards(self, wallet_set_id: str) -> GuardChain:
    """Get guards for a wallet set."""
    key = self._make_key("wallet_set", wallet_set_id)
    data = await self._storage.get(self.COLLECTION, key)

    chain = GuardChain()
    if data:
        for guard_data in data.get("guards", []):
            config = GuardConfig.from_dict(guard_data)
            guard = config.to_guard(self._storage)
            chain.add(guard)

    return chain

list_wallet_guard_names async

list_wallet_guard_names(wallet_id)

List guard names for a wallet.

Source code in python/src/algopay/guards/manager.py

async def list_wallet_guard_names(self, wallet_id: str) -> list[str]:
    """List guard names for a wallet."""
    key = self._make_key("wallet", wallet_id)
    data = await self._storage.get(self.COLLECTION, key)

    if not data:
        return []
    return [g.get("name", "unnamed") for g in data.get("guards", [])]

list_wallet_set_guard_names async

list_wallet_set_guard_names(wallet_set_id)

List guard names for a wallet set.

Source code in python/src/algopay/guards/manager.py

async def list_wallet_set_guard_names(self, wallet_set_id: str) -> list[str]:
    """List guard names for a wallet set."""
    key = self._make_key("wallet_set", wallet_set_id)
    data = await self._storage.get(self.COLLECTION, key)

    if not data:
        return []
    return [g.get("name", "unnamed") for g in data.get("guards", [])]

get_guard_chain async

get_guard_chain(wallet_id, wallet_set_id=None)

Get combined guard chain for a wallet.

Merges guards from wallet set (if provided) and wallet.

Source code in python/src/algopay/guards/manager.py

async def get_guard_chain(
    self,
    wallet_id: str,
    wallet_set_id: str | None = None,
) -> GuardChain:
    """
    Get combined guard chain for a wallet.

    Merges guards from wallet set (if provided) and wallet.
    """
    combined = GuardChain()

    # Add wallet set guards first
    if wallet_set_id:
        set_chain = await self.get_wallet_set_guards(wallet_set_id)
        for guard in set_chain:
            combined.add(guard)

    # Add wallet-specific guards
    wallet_chain = await self.get_wallet_guards(wallet_id)
    for guard in wallet_chain:
        combined.add(guard)

    return combined

check async

check(context)

Check guards for a payment context.

Returns:

Type	Description
tuple[bool, str | None, list[str]]	Tuple of (allowed, reason, passed_guards)

Source code in python/src/algopay/guards/manager.py

async def check(self, context: PaymentContext) -> tuple[bool, str | None, list[str]]:
    """
    Check guards for a payment context.

    Returns:
        Tuple of (allowed, reason, passed_guards)
    """
    chain = await self.get_guard_chain(
        context.wallet_id,
        context.wallet_set_id,
    )

    if len(chain) == 0:
        return True, None, []

    self._logger.debug(f"Checking {len(chain)} guards for wallet={context.wallet_id}")

    result = await chain.check(context)
    passed = result.metadata.get("passed_guards", []) if result.metadata else []

    if not result.allowed:
        self._logger.warning(
            f"Payment BLOCKED by guard: {result.reason} (Wallet: {context.wallet_id})"
        )
    else:
        self._logger.debug(f"Guards passed: {passed}")

    return result.allowed, result.reason, passed

record_spending async

record_spending(wallet_id, wallet_set_id, amount, recipient, purpose)

Record spending in all relevant guards.

Source code in python/src/algopay/guards/manager.py

async def record_spending(
    self,
    wallet_id: str,
    wallet_set_id: str | None,
    amount: Decimal,
    recipient: str,
    purpose: str | None,
) -> None:
    """Record spending in all relevant guards."""
    chain = await self.get_guard_chain(wallet_id, wallet_set_id)

    for guard in chain:
        # BudgetGuard uses record_spending
        if hasattr(guard, "record_spending"):
            await guard.record_spending(
                amount=amount,
                wallet_id=wallet_id,
                recipient=recipient,
                purpose=purpose,
            )
        # RateLimitGuard uses record_payment (now async with wallet_id)
        if hasattr(guard, "record_payment"):
            await guard.record_payment(wallet_id)

clear_wallet_guards async

clear_wallet_guards(wallet_id)

Clear all guards for a wallet.

Source code in python/src/algopay/guards/manager.py

async def clear_wallet_guards(self, wallet_id: str) -> None:
    """Clear all guards for a wallet."""
    key = self._make_key("wallet", wallet_id)
    await self._storage.delete(self.COLLECTION, key)

clear_wallet_set_guards async

clear_wallet_set_guards(wallet_set_id)

Clear all guards for a wallet set.

Source code in python/src/algopay/guards/manager.py

async def clear_wallet_set_guards(self, wallet_set_id: str) -> None:
    """Clear all guards for a wallet set."""
    key = self._make_key("wallet_set", wallet_set_id)
    await self._storage.delete(self.COLLECTION, key)