Lightning Enable MCP

Instructions

Implementation Reference

• python/lightning-enable-mcp/src/lightning_enable_mcp/tools/exchange_currency.py:20-122 (handler)

  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))
          })

• python/lightning-enable-mcp/src/lightning_enable_mcp/strike_wallet.py:448-530 (handler)

  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))