Skip to main content
Glama

Armor Crypto MCP

Official
by armorwallet
import json import os from pydantic import BaseModel, Field from typing_extensions import List, Optional, Literal, Dict import httpx from dotenv import load_dotenv import ast import operator import math import statistics load_dotenv() BASE_API_URL = os.getenv("BASE_API_URL") # ------------------------------ # BaseModel Definitions # ------------------------------ class WalletTokenPairs(BaseModel): wallet: str = Field(description="The name of wallet. To get wallet names use `get_user_wallets_and_groups_list`") token: str = Field(description="public address of token. To get the address from a token symbol use `get_token_details`") class WalletTokenBalance(BaseModel): wallet: str = Field(description="name of wallet") token: str = Field(description="public address of token") balance: float = Field(description="balance of token") class ConversionRequest(BaseModel): input_amount: float = Field(description="input amount to convert") input_token: str = Field(description="public address of input token") output_token: str = Field(description="public address of output token") class ConversionResponse(BaseModel): input_amount: float = Field(description="input amount before conversion") input_token: str = Field(description="public address of input token") output_token: str = Field(description="public address of output token") output_amount: float = Field(description="output amount after conversion") class SwapQuoteRequest(BaseModel): from_wallet: str = Field(description="The name of the wallet that input_token is in.") input_token: str = Field(description="public mint address of input token. To get the address from a token symbol use `get_token_details`") output_token: str = Field(description="public mint address of output token. To get the address from a token symbol use `get_token_details`") input_amount: float = Field(description="input amount to swap") slippage: float = Field("slippage percentage. To estimate slippage based on liquidity see `get_token_details` for the input_token_symbol. 1.0 for high liquidity and near 20.0 for lower liquidity.") class StakeQuoteRequest(BaseModel): from_wallet: str = Field(description="The name of the wallet that input_token is in.") input_token: str = "So11111111111111111111111111111111111111112" # Hardcoded SOL token address output_token: str = Field(description="the public mint address of the output liquid staking derivative token to stake.") # "jupSoLaHXQiZZTSfEWMTRRgpnyFm8f6sZdosWBjx93v" input_amount: float = Field(description="input amount to swap") class UnstakeQuoteRequest(BaseModel): from_wallet: str = Field(description="The name of the wallet that input_token is in.") input_token: str = Field(description="the public mint address of the input liquid staking derivative token to unstake.") # "jupSoLaHXQiZZTSfEWMTRRgpnyFm8f6sZdosWBjx93v" output_token: str = "So11111111111111111111111111111111111111112" input_amount: float = Field(description="input amount to swap") class SwapQuoteResponse(BaseModel): id: str = Field(description="unique id of the generated swap quote") wallet_address: str = Field(description="public address of the wallet") input_token_symbol: str = Field(description="symbol of the input token") input_token_address: str = Field(description="public address of the input token") output_token_symbol: str = Field(description="symbol of the output token") output_token_address: str = Field(description="public address of the output token") input_amount: float = Field(description="input amount in input token") output_amount: float = Field(description="output amount in output token") slippage: float = Field(description="slippage percentage.") class SwapTransactionRequest(BaseModel): transaction_id: str = Field(description="unique id of the generated swap quote") class StakeTransactionRequest(BaseModel): transaction_id: str = Field(description="unique id of the generated stake quote") class UnstakeTransactionRequest(BaseModel): transaction_id: str = Field(description="unique id of the generated unstake quote") class SwapTransactionResponse(BaseModel): id: str = Field(description="unique id of the swap transaction") transaction_error: Optional[str] = Field(description="error message if the transaction fails") transaction_url: str = Field(description="public url of the transaction") input_amount: float = Field(description="input amount in input token") output_amount: float = Field(description="output amount in output token") status: str = Field(description="status of the transaction") class ListWalletsRequest(BaseModel): is_archived: bool = Field(default=False, description="whether to include archived wallets") class WalletBalance(BaseModel): mint_address: str = Field(description="public mint address of output token. To get the address from a token symbol use `get_token_details`") name: str = Field(description="name of the token") symbol: str = Field(description="symbol of the token") decimals: int = Field(description="number of decimals of the token") amount: float = Field(description="balance of the token") usd_price: str = Field(description="price of the token in USD") usd_amount: float = Field(description="balance of the token in USD") class WalletInfo(BaseModel): id: str = Field(description="wallet id") name: str = Field(description="wallet name") is_archived: bool = Field(description="whether the wallet is archived") public_address: str = Field(description="public address of the wallet") class Wallet(WalletInfo): balances: List[WalletBalance] = Field(description="list of balances of the wallet") class TokenDetailsRequest(BaseModel): query: str = Field(description="token symbol or address") class TokenDetailsResponse(BaseModel): symbol: str = Field(description="symbol of the token") mint_address: str = Field(description="mint address of the token") class TokenSearchRequest(BaseModel): query: str = Field(description="token symbol or address") sort_by: Optional[Literal['decimals', 'holders', 'jupiter', 'verified', 'liquidityUsd', 'marketCapUsd', 'priceUsd', 'totalBuys', 'totalSells', 'totalTransactions', 'volume_5m', 'volume', 'volume_15m', 'volume_30m', 'volume_1h', 'volume_6h', 'volume_12h', 'volume_24h']] = Field(description="Sort token data results by this field") sort_order: Optional[Literal['asc', 'desc']] = Field(default='desc', description="The order of the sorted results") limit: Optional[int] = Field(default=10, description="The number of results to return from the search. Use default unless specified. Should not be over 30 if looking up multiple tokens.") class TokenSearchResponse(BaseModel): name: str = Field(description="name of the token") symbol: str = Field(description="symbol of the token") mint_address: Optional[str] = Field(description="mint address of the token") decimals: Optional[int] = Field(description="number of decimals of the token, returns only if include_details is True") image: Optional[str] = Field(description="image url of the token, returns only if include_details is True") holders: Optional[int] = Field(description="number of holders of the token, returns only if include_details is True") jupiter: Optional[bool] = Field(description="whether the token is supported by Jupiter, returns only if include_details is True") verified: Optional[bool] = Field(description="whether the token is verified, returns only if include_details is True") liquidityUsd: Optional[float] = Field(description="liquidity of the token in USD, returns only if include_details is True") marketCapUsd: Optional[float] = Field(description="market cap of the token in USD, returns only if include_details is True") priceUsd: Optional[float] = Field(description="price of the token in USD, returns only if include_details is True") lpBurn: Optional[float] = Field(description="lp burn of the token, returns only if include_details is True") market: Optional[str] = Field(description="market of the token, returns only if include_details is True") freezeAuthority: Optional[str] = Field(description="freeze authority of the token, returns only if include_details is True") mintAuthority: Optional[str] = Field(description="mint authority of the token, returns only if include_details is True") poolAddress: Optional[str] = Field(description="pool address of the token, returns only if include_details is True") totalBuys: Optional[int] = Field(description="total number of buys of the token, returns only if include_details is True") totalSells: Optional[int] = Field(description="total number of sells of the token, returns only if include_details is True") totalTransactions: Optional[int] = Field(description="total number of transactions of the token, returns only if include_details is True") volume: Optional[float] = Field(description="volume of the token, returns only if include_details is True") volume_5m: Optional[float] = Field(description="volume of the token in the last 5 minutes, returns only if include_details is True") volume_15m: Optional[float] = Field(description="volume of the token in the last 15 minutes, returns only if include_details is True") volume_30m: Optional[float] = Field(description="volume of the token in the last 30 minutes, returns only if include_details is True") volume_1h: Optional[float] = Field(description="volume of the token in the last 1 hour, returns only if include_details is True") volume_6h: Optional[float] = Field(description="volume of the token in the last 6 hours, returns only if include_details is True") volume_12h: Optional[float] = Field(description="volume of the token in the last 12 hours, returns only if include_details is True") volume_24h: Optional[float] = Field(description="volume of the token in the last 24 hours, returns only if include_details is True") class GroupInfo(BaseModel): id: str = Field(description="id of the group") name: str = Field(description="name of the group") is_archived: bool = Field(description="whether the group is archived") class SingleGroupInfo(GroupInfo): wallets: List[WalletInfo] = Field(description="list of wallets in the group") class WalletArchiveOrUnarchiveResponse(BaseModel): wallet_name: str = Field(description="name of the wallet") message: str = Field(description="message of the operation showing if wallet was archived or unarchived") class CreateGroupResponse(BaseModel): id: str = Field(description="id of the group") name: str = Field(description="name of the group") is_archived: bool = Field(description="whether the group is archived") class AddWalletToGroupResponse(BaseModel): wallet_name: str = Field(description="name of the wallet to add to the group") group_name: str = Field(description="name of the group to add the wallet to") message: str = Field(description="message of the operation showing if wallet was added to the group") class GroupArchiveOrUnarchiveResponse(BaseModel): group: str = Field(description="name of the group") class RemoveWalletFromGroupResponse(BaseModel): wallet: str = Field(description="name of the wallet to remove from the group") group: str = Field(description="name of the group to remove the wallet from") class UserWalletsAndGroupsResponse(BaseModel): id: str = Field(description="id of the user") email: str = Field(description="email of the user") first_name: str = Field(description="first name of the user") last_name: str = Field(description="last name of the user") slippage: float = Field(description="slippage set by the user") wallet_groups: List[GroupInfo] = Field(description="list of user's wallet groups") wallets: List[WalletInfo] = Field(description="list of user's wallets") class TransferTokensRequest(BaseModel): from_wallet: str = Field(description="name of the wallet to transfer tokens from") to_wallet_address: str = Field(description="public address of the wallet to transfer tokens to. Use `get_user_wallets_and_group_list` if you only have a wallet name") token: str = Field(description="public contract address of the token to transfer. To get the address from a token symbol use `get_token_details`") amount: float = Field(description="amount of tokens to transfer") class TransferTokenResponse(BaseModel): amount: float = Field(description="amount of tokens transferred") from_wallet_address: str = Field(description="public address of the wallet tokens were transferred from") to_wallet_address: str = Field(description="public address of the wallet tokens were transferred to") token_address: str = Field(description="public address of the token transferred") transaction_url: str = Field(description="public url of the transaction") message: str = Field(description="message of the operation showing if tokens were transferred") class ListDCAOrderRequest(BaseModel): status: Optional[Literal["COMPLETED", "OPEN", "CANCELLED"]] = Field(description="status of the DCA orders, if specified filters the results.") limit: Optional[int] = Field(default=30, description="number of mostrecent results to return") class DCAOrderRequest(BaseModel): wallet: str = Field(description="name of the wallet") input_token: str = Field(description="public address of the input token. To get the address from a token symbol use `get_token_details`") output_token: str = Field(description="public address of the output token. To get the address from a token symbol use `get_token_details`") amount: float = Field(description="total amount of input token to invest") cron_expression: str = Field(description="cron expression for the DCA worker execution frequency") strategy_duration_unit: Literal["MINUTE", "HOUR", "DAY", "WEEK", "MONTH", "YEAR"] = Field(description="unit of the duration of the DCA order") strategy_duration: int = Field(description="Total running time of the DCA order given in strategy duration units, should be more than 0") execution_type: Literal["MULTIPLE", "SINGLE"] = Field(description="set to SINGLE only if the user is asking for a single scheduled order, MULTIPLE if it is a true DCA") token_address_watcher: Optional[str] = Field(description="If the DCA is conditional, public address of the token to watch.") watch_field: Optional[Literal["liquidity", "marketCap", "price"]] = Field(description="If the DCA is conditional, field to watch for the condition") delta_type: Optional[Literal["INCREASE", "DECREASE", "MOVE", "MOVE_DAILY", "AVERAGE_MOVE"]] = Field(description="If the DCA is conditional, the operator of the watch field in the conditional statement") delta_percentage: Optional[float] = Field(description="If the DCA is conditional, percentage of the change to watch for given the delta_type") time_zone: Optional[str] = Field(description="user's time zone. Defaults to UTC") class DCAWatcher(BaseModel): watch_field: Literal["liquidity", "marketCap", "price"] = Field(description="field to watch for the DCA order") delta_type: Literal["INCREASE", "DECREASE", "MOVE", "MOVE_DAILY", "AVERAGE_MOVE"] = Field(description="type of the delta") initial_value: float = Field(description="initial value of the delta") delta_percentage: float = Field(description="percentage of the delta") class TokenData(BaseModel): name: str = Field(description="name of the token") symbol: str = Field(description="symbol of the token") mint_address: str = Field(description="mint address of the token") class DCAOrderResponse(BaseModel): id: str = Field(description="id of the DCA order") amount: float = Field(description="amount of tokens to invest") investment_per_cycle: float = Field(description="amount of tokens to invest per cycle") cycles_completed: int = Field(description="number of cycles completed") total_cycles: int = Field(description="total number of cycles") human_readable_expiry: str = Field(description="human readable expiry date of the DCA order") status: str = Field(description="status of the DCA order") input_token_data: TokenData = Field(description="details of the input token") output_token_data: TokenData = Field(description="details of the output token") wallet_name: str = Field(description="name of the wallet") watchers: List[DCAWatcher] = Field(description="list of watchers for the DCA order") dca_transactions: List[dict] = Field(description="list of DCA transactions") # Can be further typed if structure is known created: str = Field(description="Linux timestamp of the creation of the order") class ListOrderRequest(BaseModel): status: Optional[Literal["OPEN", "CANCELLED", "EXPIRED", "COMPLETED", "FAILED", "IN_PROCESS"]] = Field(description="status of the orders, if specified filters results.") limit: Optional[int] = Field(default=30, description="number of most recent results to return") class CreateOrderRequest(BaseModel): wallet: str = Field(description="name of the wallet") input_token: str = Field(description="public address of the input token") output_token: str = Field(description="public address of the output token") amount: float = Field(description="amount of input token to invest") strategy_duration: int = Field(description="duration of the order") strategy_duration_unit: Literal["MINUTE", "HOUR", "DAY", "WEEK", "MONTH", "YEAR"] = Field(description="unit of the duration of the order") watch_field: Literal["liquidity", "marketCap", "price"] = Field(description="field to watch to execute the order. Can be price, marketCap or liquidity") direction: Literal["ABOVE", "BELOW"] = Field(description="whether or not the order is above or below current market value") token_address_watcher: str = Field(description="public address of the token to watch. should be output token for limit orders and input token for stop loss and take profit orders") target_value: Optional[float] = Field(description="target value to execute the order. You must always specify a target value or delta percentage.") delta_percentage: Optional[float] = Field(description="delta percentage to execute the order. You must always specify a target value or delta percentage.") class OrderWatcher(BaseModel): watch_field: Literal["liquidity", "marketCap", "price"] = Field(description="field being watched for a delta") delta_type: Literal["INCREASE", "DECREASE", "MOVE", "MOVE_DAILY", "AVERAGE_MOVE"] = Field(description="type of delta change") initial_value: float = Field(description="initial value when watcher was created") delta_percentage: float = Field(description="percentage for delta change") watcher_type: Literal["LIMIT", "STOP_LOSS"] = Field(description="type of watcher") buying_price: Optional[float] = Field(description="price at which to buy", default=None) class OrderResponse(BaseModel): id: str = Field(description="unique identifier of the order") amount: float = Field(description="amount of tokens to invest") status: str = Field(description="current status of the order") input_token_data: TokenData = Field(description="details of the input token") output_token_data: TokenData = Field(description="details of the output token") wallet_name: str = Field(description="name of the wallet") execution_type: Literal["LIMIT", "STOP_LOSS", "TAKE_PROFIT"] = Field(description="type of the order") expiry_time: str = Field(description="expiry time of the order in ISO format") watchers: List[OrderWatcher] = Field(description="list of watchers for the order") transaction: Optional[dict] = Field(description="transaction details if any", default=None) created: str = Field(description="ISO 8601 timestamp of the creation of the order") class CancelOrderRequest(BaseModel): order_id: str = Field(description="id of the limit order") class CancelOrderResponse(BaseModel): order_id: str = Field(description="id of the limit order") status: str = Field(description="status of the limit order") class CancelDCAOrderRequest(BaseModel): dca_order_id: str = Field(description="id of the DCA order") class CancelDCAOrderResponse(BaseModel): dca_order_id: str = Field(description="id of the DCA order") status: str = Field(description="status of the DCA order") class ListSingleGroupRequest(BaseModel): group_name: str = Field(description="Name of the group to retrieve details for") class CreateWalletRequest(BaseModel): name: str = Field(description="Name of the wallet to create") class ArchiveWalletsRequest(BaseModel): wallet: str = Field(description="Name of the wallet to archive") class UnarchiveWalletsRequest(BaseModel): wallet: str = Field(description="Name of the wallet to unarchive") class CreateGroupsRequest(BaseModel): name: str = Field(description="Name of the group to create") class AddWalletToGroupRequest(BaseModel): group: str = Field(description="Name of the group to add wallets to") wallet: str = Field(description="Name of the wallet to add to the group") class ArchiveWalletGroupRequest(BaseModel): group: str = Field(description="Name of the group to archive") class UnarchiveWalletGroupRequest(BaseModel): group: str = Field(description="Name of the group to unarchive") class RemoveWalletsFromGroupRequest(BaseModel): group: str = Field(description="Name of the group to remove wallets from") wallet: str = Field(description="List of wallet names to remove from the group") class TopTrendingTokensRequest(BaseModel): time_frame: Literal["5m", "15m", "30m", "1h", "2h", "3h", "4h", "5h", "6h", "12h", "24h"] = Field(default="24h", description="Time frame to get the top trending tokens") class StakeBalanceResponse(BaseModel): total_stake_amount: float = Field(description="Total stake balance in jupSol") total_stake_amount_in_usd: float = Field(description="Total stake balance in USD") class RenameWalletRequest(BaseModel): wallet: str = Field(description="Name of the wallet to rename") new_name: str = Field(description="New name of the wallet") class CandleStickRequest(BaseModel): token_address: str = Field(description="Public mint address of the token. To get the address from a token symbol use `get_token_details`") time_interval: Literal["1s", "5s", "15s", "1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "6h", "8h", "12h", "1d", "3d", "1w", "1mn"] = Field(default="1h", description="Time frame to get the candle sticks. Use larger candle time frames over larger time windows to keep returned candles minimal") time_from: str = Field(description="The time from which to start the candle data in ISO 8601 format. Attempt to change this to keep number of candles returned under 64.") time_to: Optional[str] = Field(default=None, description="The time to end the candle data in ISO 8601 format. Use only for historic analysis.") market_cap: Optional[bool] = Field(default=False, description="Whether to return the marketcap of the token instead of the price") class PrivateKeyRequest(BaseModel): wallet: str = Field(description="Name of the wallet to get the mnemonic or private key for") key_type: Literal['PRIVATE_KEY', 'MNEMONIC'] = Field(description="Whether to return the private or mnemonic key") # ------------------------------ # Container Models for List Inputs # ------------------------------ class RemoveWalletsFromGroupRequestContainer(BaseModel): remove_wallets_from_group_requests: List[RemoveWalletsFromGroupRequest] class AddWalletToGroupRequestContainer(BaseModel): add_wallet_to_group_requests: List[AddWalletToGroupRequest] class CreateWalletRequestContainer(BaseModel): create_wallet_requests: List[CreateWalletRequest] class ArchiveWalletsRequestContainer(BaseModel): archive_wallet_requests: List[ArchiveWalletsRequest] class UnarchiveWalletRequestContainer(BaseModel): unarchive_wallet_requests: List[UnarchiveWalletsRequest] class ArchiveWalletGroupRequestContainer(BaseModel): archive_wallet_group_requests: List[ArchiveWalletGroupRequest] class UnarchiveWalletGroupRequestContainer(BaseModel): unarchive_wallet_group_requests: List[UnarchiveWalletGroupRequest] class WalletTokenPairsContainer(BaseModel): wallet_token_pairs: List[WalletTokenPairs] class CreateGroupsRequestContainer(BaseModel): create_groups_requests: List[CreateGroupsRequest] class ConversionRequestContainer(BaseModel): conversion_requests: List[ConversionRequest] class SwapQuoteRequestContainer(BaseModel): swap_quote_requests: List[SwapQuoteRequest] class StakeQuoteRequestContainer(BaseModel): stake_quote_requests: List[StakeQuoteRequest] class UnstakeQuoteRequestContainer(BaseModel): unstake_quote_requests: List[UnstakeQuoteRequest] class SwapTransactionRequestContainer(BaseModel): swap_transaction_requests: List[SwapTransactionRequest] class StakeTransactionRequestContainer(BaseModel): stake_transaction_requests: List[StakeTransactionRequest] class UnstakeTransactionRequestContainer(BaseModel): unstake_transaction_requests: List[UnstakeTransactionRequest] class TokenSearchResponseContainer(BaseModel): token_search_responses: List[TokenSearchResponse] class TokenDetailsRequestContainer(BaseModel): token_details_requests: List[TokenDetailsRequest] class TokenDetailsResponseContainer(BaseModel): token_details_responses: List[TokenDetailsResponse] class TransferTokensRequestContainer(BaseModel): transfer_tokens_requests: List[TransferTokensRequest] class DCAOrderRequestContainer(BaseModel): dca_order_requests: List[DCAOrderRequest] class CancelDCAOrderRequestContainer(BaseModel): cancel_dca_order_requests: List[CancelDCAOrderRequest] class CreateOrderRequestContainer(BaseModel): create_order_requests: List[CreateOrderRequest] class CreateOrderResponseContainer(BaseModel): create_order_responses: List[OrderResponse] class CancelOrderRequestContainer(BaseModel): cancel_order_requests: List[CancelOrderRequest] class CancelOrderResponseContainer(BaseModel): cancel_order_responses: List[CancelOrderResponse] class RenameWalletRequestContainer(BaseModel): rename_wallet_requests: List[RenameWalletRequest] class ListDCAOrderResponseContainer(BaseModel): list_dca_order_responses: List[DCAOrderResponse] class ListOrderResponseContainer(BaseModel): list_order_responses: List[OrderResponse] # ------------------------------ # API Client # ------------------------------ # Setup logger for the module import logging import traceback class ArmorWalletAPIClient: def __init__(self, access_token: str, base_api_url: str = 'https://app.armorwallet.ai/api/v1', logger=None): self.base_api_url = base_api_url self.access_token = access_token self.logger = logger async def _api_call(self, method: str, endpoint: str, payload: str = None) -> dict: """Utility function for API calls to the wallet. It sets common headers and raises errors on non-2xx responses. """ url = f"{self.base_api_url}/{endpoint}" payload = json.dumps(payload) if self.logger is not None: self.logger.debug(f"Request: {method} {url} Payload: {payload}") headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {self.access_token}' } try: async with httpx.AsyncClient(timeout=30) as client: response = await client.request(method, url, headers=headers, data=payload, follow_redirects=False) if self.logger is not None: self.logger.debug(f"Response status: {response.status_code} Response: {response.text}") if response.status_code >= 400: if self.logger is not None: self.logger.error(f"API Error {response.status_code}: {response.text}") raise Exception(f"API Error {response.status_code}: {response.text}") try: return response.json() except Exception: if self.logger is not None: self.logger.error(f"JSON Parsing: {response.text}") return {"text": response.text} except Exception as e: traceback.print_exc() if self.logger is not None: self.logger.error(f"{e}") return {"text": str(e)} async def get_wallet_token_balance(self, data: WalletTokenPairsContainer) -> List[WalletTokenBalance]: """Get balances from a list of wallet and token pairs.""" # payload = [v.model_dump() for v in data.wallet_token_pairs] payload = data.model_dump(exclude_none=True)['wallet_token_pairs'] return await self._api_call("POST", "tokens/wallet-token-balance/", payload) async def conversion_api(self, data: ConversionRequestContainer) -> List[ConversionResponse]: """Perform a token conversion.""" # payload = [v.model_dump() for v in data.conversion_requests] payload = data.model_dump(exclude_none=True)['conversion_requests'] return await self._api_call("POST", "tokens/token-price-conversion/", payload) async def swap_quote(self, data: SwapQuoteRequestContainer) -> List[SwapQuoteResponse]: """Obtain a swap quote.""" # payload = [v.model_dump() for v in data.swap_quote_requests] payload = data.model_dump(exclude_none=True)['swap_quote_requests'] return await self._api_call("POST", "transactions/quote/", payload) async def stake_quote(self, data: StakeQuoteRequestContainer) -> StakeQuoteRequestContainer: """Obtain a stake quote.""" payload = data.model_dump(exclude_none=True)['stake_quote_requests'] return await self._api_call("POST", "transactions/quote/", payload) async def unstake_quote(self, data: UnstakeQuoteRequestContainer) -> UnstakeQuoteRequestContainer: """Obtain an unstake quote.""" payload = data.model_dump(exclude_none=True)['unstake_quote_requests'] return await self._api_call("POST", "transactions/quote/", payload) async def swap_transaction(self, data: SwapTransactionRequestContainer) -> List[SwapTransactionResponse]: """Execute the swap transactions.""" # payload = [v.model_dump() for v in data.swap_transaction_requests] payload = data.model_dump(exclude_none=True)['swap_transaction_requests'] return await self._api_call("POST", "transactions/swap/", payload) async def stake_transaction(self, data: StakeTransactionRequestContainer) -> StakeTransactionRequestContainer: """Execute the stake transactions.""" payload = data.model_dump(exclude_none=True)['stake_transaction_requests'] return await self._api_call("POST", "transactions/swap/", payload) async def unstake_transaction(self, data: UnstakeTransactionRequestContainer) -> UnstakeTransactionRequestContainer: """Execute the unstake transactions.""" payload = data.model_dump(exclude_none=True)['unstake_transaction_requests'] return await self._api_call("POST", "transactions/swap/", payload) async def get_all_wallets(self, data: ListWalletsRequest) -> List[Wallet]: """Return all wallets with balances.""" return await self._api_call("GET", f"wallets/?is_archived={data.is_archived}") async def search_token(self, data: TokenSearchRequest) -> TokenSearchResponseContainer: """Get details of a token.""" payload = data.model_dump(exclude_none=True) return await self._api_call("POST", "tokens/search-token/", payload) async def get_official_token_address(self, data: TokenDetailsRequestContainer) -> TokenDetailsResponseContainer: """Retrieve the mint address of token.""" payload = data.model_dump(exclude_none=True)['token_details_requests'] return await self._api_call("POST", "tokens/official-token-detail/", payload) async def list_groups(self) -> List[GroupInfo]: """Return a list of wallet groups.""" return await self._api_call("GET", "wallets/groups/") async def list_single_group(self, data: ListSingleGroupRequest) -> SingleGroupInfo: """Return details for a single wallet group.""" return await self._api_call("GET", f"wallets/groups/{data.group_name}/") async def create_wallet(self, data: CreateWalletRequestContainer) -> List[WalletInfo]: """Create new wallets given a list of wallet names.""" # payload = json.dumps([{"name": wallet_name} for wallet_name in data.wallet_names]) payload = data.model_dump(exclude_none=True)['create_wallet_requests'] return await self._api_call("POST", "wallets/", payload) async def archive_wallets(self, data: ArchiveWalletsRequestContainer) -> List[WalletArchiveOrUnarchiveResponse]: """Archive the wallets specified in the list.""" # payload = json.dumps([{"wallet": wallet_name} for wallet_name in data.wallet_names]) payload = data.model_dump(exclude_none=True)['archive_wallet_requests'] return await self._api_call("POST", "wallets/archive/", payload) async def unarchive_wallets(self, data: UnarchiveWalletsRequest) -> List[WalletArchiveOrUnarchiveResponse]: """Unarchive the wallets specified in the list.""" # payload = json.dumps([{"wallet": wallet_name} for wallet_name in data.wallet_names]) payload = data.model_dump(exclude_none=True)['unarchive_wallet_requests'] return await self._api_call("POST", "wallets/unarchive/", payload) async def create_groups(self, data: CreateGroupsRequest) -> List[CreateGroupResponse]: """Create new wallet groups given a list of group names.""" # payload = json.dumps([{"name": group_name} for group_name in data.group_names]) payload = data.model_dump(exclude_none=True)['create_groups_requests'] return await self._api_call("POST", "wallets/groups/", payload) async def add_wallets_to_group(self, data: AddWalletToGroupRequestContainer) -> List[AddWalletToGroupResponse]: """Add wallets to a specific group.""" # payload = json.dumps([{"wallet": wallet_name, "group": data.group_name} for wallet_name in data.wallet_names]) payload = data.model_dump(exclude_none=True)['add_wallet_to_group_requests'] return await self._api_call("POST", "wallets/add-wallet-to-group/", payload) async def archive_wallet_group(self, data: ArchiveWalletGroupRequestContainer) -> List[GroupArchiveOrUnarchiveResponse]: """Archive the specified wallet groups.""" # payload = json.dumps([{"group": group_name} for group_name in data.group_names]) payload = data.model_dump(exclude_none=True)['archive_wallet_group_requests'] return await self._api_call("POST", "wallets/group-archive/", payload) async def unarchive_wallet_group(self, data: UnarchiveWalletGroupRequestContainer) -> List[GroupArchiveOrUnarchiveResponse]: """Unarchive the specified wallet groups.""" # payload = json.dumps([{"group": group_name} for group_name in data.group_names]) payload = data.model_dump(exclude_none=True)['unarchive_wallet_group_requests'] return await self._api_call("POST", "wallets/group-unarchive/", payload) async def remove_wallets_from_group(self, data: RemoveWalletsFromGroupRequestContainer) -> List[RemoveWalletFromGroupResponse]: """Remove wallets from a group.""" # payload = json.dumps([{"wallet": wallet_name, "group": data.group_name} for wallet_name in data.wallet_names]) payload = data.model_dump(exclude_none=True)['remove_wallets_from_group_requests'] return await self._api_call("POST", "wallets/remove-wallet-from-group/", payload) async def transfer_tokens(self, data: TransferTokensRequestContainer) -> List[TransferTokenResponse]: """Transfer tokens from one wallet to another.""" # payload = [v.model_dump() for v in data.transfer_tokens_requests] payload = data.model_dump(exclude_none=True)['transfer_tokens_requests'] return await self._api_call("POST", "transfers/transfer/", payload) async def create_dca_order(self, data: DCAOrderRequestContainer) -> List[DCAOrderResponse]: """Create a DCA order.""" # payload = [v.model_dump() for v in data.dca_order_requests] payload = data.model_dump(exclude_none=True)['dca_order_requests'] return await self._api_call("POST", "transactions/dca-order/create/", payload) async def list_dca_orders(self, data: ListDCAOrderRequest) -> ListDCAOrderResponseContainer: """List all DCA orders.""" payload = data.model_dump(exclude_none=True) return await self._api_call("POST", f"transactions/dca-order/", payload) async def cancel_dca_order(self, data: CancelDCAOrderRequestContainer) -> List[CancelDCAOrderResponse]: """Cancel a DCA order.""" # payload = [v.model_dump() for v in data.cancel_dca_order_requests] payload = data.model_dump(exclude_none=True)['cancel_dca_order_requests'] return await self._api_call("POST", "transactions/dca-order/cancel/", payload) async def create_order(self, data: CreateOrderRequestContainer) -> CreateOrderResponseContainer: """Create a order.""" payload = data.model_dump(exclude_none=True)['create_order_requests'] return await self._api_call("POST", "transactions/order/create/", payload) async def list_orders(self, data: ListOrderRequest) -> ListOrderResponseContainer: """List all orders.""" payload = data.model_dump(exclude_none=True) return await self._api_call("POST", f"transactions/order/", payload) async def cancel_order(self, data: CancelOrderRequestContainer) -> CancelOrderResponseContainer: """Cancel a order.""" payload = data.model_dump(exclude_none=True)['cancel_order_requests'] return await self._api_call("POST", "transactions/order/cancel/", payload) async def top_trending_tokens(self, data: TopTrendingTokensRequest) -> List: """Get the top trending tokens.""" payload = data.model_dump(exclude_none=True) return await self._api_call("POST", f"tokens/trending/", payload) async def get_stake_balances(self) -> StakeBalanceResponse: """Get the stake balances.""" return await self._api_call("GET", "frontend/wallets/stake/balance/") async def rename_wallet(self, data: RenameWalletRequestContainer) -> List: """Rename a wallet.""" payload = data.model_dump(exclude_none=True)['rename_wallet_requests'] return await self._api_call("POST", "wallets/rename/", payload) async def get_market_candle_data(self, data: CandleStickRequest) -> Dict: """Get the candle sticks.""" payload = data.model_dump(exclude_none=True) return await self._api_call("POST", f"tokens/candles/", payload) async def send_key_to_telegram(self, data: PrivateKeyRequest) -> Dict: """Send the mnemonic or private key to telegram.""" payload = data.model_dump(exclude_none=True) return await self._api_call("POST", f"users/telegram/send-message/", payload) # ------------------------------ # Utility Functions # ------------------------------ def calculate(expr: str, variables: dict = None) -> float: """ Evaluate a math/stat expression with support for variables and common functions. """ variables = variables or {} # Allowed names from math and statistics safe_names = { k: v for k, v in vars(math).items() if not k.startswith("__") } safe_names.update({ 'mean': statistics.mean, 'median': statistics.median, 'stdev': statistics.stdev, 'variance': statistics.variance, 'sum': sum, 'min': min, 'max': max, 'len': len, 'abs': abs, 'round': round }) # Safe operators ops = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod, ast.Pow: operator.pow, ast.USub: operator.neg } def _eval(node): if isinstance(node, ast.Num): return node.n elif isinstance(node, ast.Constant): # Python 3.8+ return node.value elif isinstance(node, ast.BinOp): return ops[type(node.op)](_eval(node.left), _eval(node.right)) elif isinstance(node, ast.UnaryOp): return ops[type(node.op)](_eval(node.operand)) elif isinstance(node, ast.Name): if node.id in variables: return variables[node.id] elif node.id in safe_names: return safe_names[node.id] else: raise NameError(f"Unknown variable or function: {node.id}") elif isinstance(node, ast.Call): func = _eval(node.func) args = [_eval(arg) for arg in node.args] return func(*args) elif isinstance(node, ast.List): return [_eval(elt) for elt in node.elts] else: raise TypeError(f"Unsupported expression type: {type(node)}") parsed = ast.parse(expr, mode="eval") return _eval(parsed.body)

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/armorwallet/armor-crypto-mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server