exchange_currency
Convert USD to BTC or BTC to USD within your Strike wallet using the Lightning Enable MCP server for automated currency exchange.
Instructions
Exchange currency within your wallet (USD to BTC or BTC to USD). Currently only available with Strike wallet.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| source_currency | Yes | Currency to convert from: USD or BTC | |
| target_currency | Yes | Currency to convert to: BTC or USD | |
| amount | Yes | Amount in source currency (e.g., 100 for $100 or 0.001 for 0.001 BTC) |
Implementation Reference
- The exchange_currency tool handler which validates input and orchestrates the exchange via the StrikeWallet instance.
async def exchange_currency( source_currency: str, target_currency: str, amount: float, wallet: "StrikeWallet | None" = None, ) -> str: """ Exchange currency within your wallet (USD to BTC or BTC to USD). Currently only available with Strike wallet which supports multi-currency accounts. Args: source_currency: Currency to convert from: USD or BTC target_currency: Currency to convert to: BTC or USD amount: Amount in source currency (e.g., 100 for $100 or 0.001 for 0.001 BTC) wallet: Strike wallet instance Returns: JSON with exchange result or error message """ if not source_currency or not source_currency.strip(): return json.dumps({ "success": False, "error": "Source currency is required (USD or BTC)" }) if not target_currency or not target_currency.strip(): return json.dumps({ "success": False, "error": "Target currency is required (BTC or USD)" }) if amount <= 0: return json.dumps({ "success": False, "error": "Amount must be greater than 0" }) if not wallet: return json.dumps({ "success": False, "error": "Currency exchange requires Strike wallet. Set STRIKE_API_KEY environment variable." }) # Verify it's a Strike wallet from ..strike_wallet import StrikeWallet if not isinstance(wallet, StrikeWallet): provider_name = type(wallet).__name__.replace("Wallet", "") return json.dumps({ "success": False, "error": f"{provider_name} does not support currency exchange. Use Strike wallet.", "errorCode": "NOT_SUPPORTED", "hint": "Set STRIKE_API_KEY environment variable for currency exchange." }) try: result = await wallet.exchange_currency( source_currency=source_currency.strip(), target_currency=target_currency.strip(), amount=Decimal(str(amount)), ) if not result.success: return json.dumps({ "success": False, "error": result.error_message, "errorCode": result.error_code, }) # Format amounts for display if result.source_currency == "BTC": source_formatted = f"{result.source_amount:.8f} BTC" else: source_formatted = f"${result.source_amount:,.2f} USD" if result.target_currency == "BTC": target_formatted = f"{result.target_amount:.8f} BTC" else: target_formatted = f"${result.target_amount:,.2f} USD" return json.dumps({ "success": True, "provider": "Strike", "exchange": { "id": result.exchange_id, "sourceCurrency": result.source_currency, "targetCurrency": result.target_currency, "sourceAmount": float(result.source_amount) if result.source_amount else None, "targetAmount": float(result.target_amount) if result.target_amount else None, "rate": float(result.rate) if result.rate else None, "fee": float(result.fee) if result.fee else None, "state": result.state, }, "message": f"Exchanged {source_formatted} for {target_formatted}" }, indent=2) except Exception as e: logger.exception("Error exchanging currency") return json.dumps({ "success": False, "error": sanitize_error(str(e)) }) - The implementation of the currency exchange logic, communicating with the Strike REST API.
async def exchange_currency( self, source_currency: str, target_currency: str, amount: Decimal, ) -> ExchangeResult: """ Exchange currency (USD to BTC or BTC to USD). Args: source_currency: Currency to sell (USD or BTC) target_currency: Currency to buy (BTC or USD) amount: Amount in source currency Returns: ExchangeResult with exchange details """ source_currency = source_currency.upper() target_currency = target_currency.upper() if source_currency not in ("USD", "BTC") or target_currency not in ("USD", "BTC"): return ExchangeResult.failed( "INVALID_CURRENCY", "Strike only supports USD and BTC exchange" ) if source_currency == target_currency: return ExchangeResult.failed( "SAME_CURRENCY", "Source and target currency must be different" ) if amount <= 0: return ExchangeResult.failed("INVALID_AMOUNT", "Amount must be positive") try: # Create exchange quote quote_request = { "sell": source_currency, "buy": target_currency, "amount": {"currency": source_currency, "amount": str(amount)}, } logger.info(f"Creating exchange: {amount} {source_currency} -> {target_currency}") quote = await self._request("POST", "/currency-exchange-quotes", quote_request) quote_id = quote.get("id") if not quote_id: return ExchangeResult.failed("INVALID_QUOTE", "No quote ID returned") # Execute exchange result = await self._request( "PATCH", f"/currency-exchange-quotes/{quote_id}/execute" ) source_amt = Decimal("0") target_amt = Decimal("0") fee = None if result.get("sourceAmount", {}).get("amount"): source_amt = Decimal(str(result["sourceAmount"]["amount"])) if result.get("targetAmount", {}).get("amount"): target_amt = Decimal(str(result["targetAmount"]["amount"])) if result.get("fee", {}).get("amount"): fee = Decimal(str(result["fee"]["amount"])) rate = target_amt / source_amt if source_amt > 0 else None logger.info(f"Exchange completed: {source_amt} {source_currency} -> {target_amt} {target_currency}") return ExchangeResult.succeeded( exchange_id=result.get("id", quote_id), source_currency=source_currency, target_currency=target_currency, source_amount=source_amt, target_amount=target_amt, rate=rate, fee=fee, state=result.get("state", "COMPLETED"), ) except StrikeError as e: return ExchangeResult.failed("API_ERROR", str(e))