# Directory Structure
```
├── .github
│ └── workflows
│ └── python-publish.yml
├── armor_crypto_mcp
│ ├── __init__.py
│ ├── armor_client.py
│ └── armor_mcp.py
├── CODE_OF_CONDUCT.md
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README_prompts.md
├── README.md
├── requirements.txt
└── smithery.yaml
```
# Files
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Armor Crypto MCP
*Alpha Test version 0.1.24*
A single source for integrating AI Agents with the Crypto ecosystem. This includes Wallet creation and management, swaps, transfers, event-based trades like DCA, stop loss and take profit, and much more. The Armor MCP supports Solana in Alpha and, when in beta, will support more than a dozen blockchains, including Ethereum. Base, Avalanche, Bitcoin, Sui, Berachain, megaETH, Optimism, Ton, BNB, and Arbitrum, among others. Using Armor's MCP you can bring all of crypto into your AI Agent with unified logic and a complete set of tools.
<br />
<br />
<br />
<br />
<br />
<br />
# Features
🧠 AI Native
📙 Wallet Management
🔃 Swaps
🌈 Specialized trades (DCA, Stop Loss etc.)
⛓️ Multi-chain
↔️ Cross-chain transations
🥩 Staking
🤖 Fast intergration to Agentic frameworks
👫 Social Sentiment
🔮 Prediction
<br />
<br />
<br />
<br />
<br />
<br />
<br />
<br />
# Requirements
### 1. Make sure you have python installed
<br />
### 2. Install `uv`
*Linux / Windows*
```sh
pip install uv
```
*Mac*
```sh
brew install uv
```
<br />
### 3. Claude Desktop or your AI Agent will run the MCP
See [Usage & Configuration](#usage--configuration) for details.
<br />
<br />
<br />
<br />
<br />
# Alpha Testing
We are currently in pre-alpha, and we are testing the capabilities of various agents and agentic frameworks like Claude Desktop, Cline, Cursor, n8n, etc.
## Current Features & Tools
- Wallet Management
- Grouping & Organization
- Archiving
- Swap & Trades
- Normal swap
- DCA (place / list / cancel)
- Scheduled Orders
- Limit Orders (place / list / cancel)
- Staking and Unstaking
- Token Search and Trending Tokens
- Statistical Calculator for accurate Analysis
- Supports Solana blockchain
## Coming Soon
- More Blockchain Support
- Minting
- Armor Agents as a Tool (or A2A)
## MCP Setup
Currently you need to have the Armor NFT to get an API Key.
Get it [here](https://codex.armorwallet.ai/)
## Usage & Configuration
To use the Armor MCP with your agent, you need the following configuration, replace `<PUT-YOUR-KEY-HERE>` with your API key:
```json
{
"mcpServers": {
"armor-crypto-mcp": {
"command": "uvx",
"args": ["armor-crypto-mcp@latest", "--version"],
"env": {
"ARMOR_API_KEY": "<PUT-YOUR-KEY-HERE>"
}
}
}
}
```
<br />
<br />
<br />
<br />
<br />
<br />
# Use in Claude Desktop
1. Must have Developer Mode enabled
2. Open Claude Desktop's File Menu top left of the window.
3. Go to File > Settings
4. Under Developer, click Edit Configuration
5. In the config file, insert the `armor-wallet-mcp` section from above
6. Make sure to replace the placeholder with your API key
7. Save the file and start a new Chat in Claude Desktop
## Use in Cline
1. Click on the `MCP Servers` button in the Cline tab in VSCode on the left panel
2. Scroll to the bottom of the left panel and click on `Configure MCP Servers`
3. In the config file, insert `armor-wallet-mcp` section from above
4. Make sure to replace the placeholder with your API key
5. Save the file, click `Done` under the `MCP Servers` tab and start chatting with Cline
## Use in n8n
1. Open the n8n app
2. Bottom-left of screen click `...` next to your username and click `Settings`
3. On the left panel, click `Community nodes` and then `Install a Community Node` button
4. In the search field for `npm Package Name` type in *mcp*
5. Install `MCP Nodes`
6. Add any MCP node, for example: `List Tools`
7. In the MCP Client `Parameters` tab, click `Select Credential` and click `Create new credential`
8. Under `Command` enter `uvx`
9. Under `Arguments` enter `armor-crypto-mcp`
10. Under `Environments` enter `ARMOR_API_KEY=eyJhbGciOiJIUzI1NiIsIn...` paste the full API Key value after the `=`
11. Back in the `Parameters` tab you can choose the MCP `Operation` for that Node
<br />
<br />
<br />
<br />
<br />
<br />
# Using Armor MCP
Once you have setup the Armor MCP [here are some prompts you can use to get started](https://github.com/armorwallet/armor-crypto-mcp/blob/main/README_prompts.md)
<br />
<br />
<br />
```
--------------------------------------------------------------------------------
/CODE_OF_CONDUCT.md:
--------------------------------------------------------------------------------
```markdown
# Contributor Covenant Code of Conduct
## Our Pledge
We as members, contributors, and leaders pledge to make participation in our
community a harassment-free experience for everyone, regardless of age, body
size, visible or invisible disability, ethnicity, sex characteristics, gender
identity and expression, level of experience, education, socio-economic status,
nationality, personal appearance, race, caste, color, religion, or sexual
identity and orientation.
We pledge to act and interact in ways that contribute to an open, welcoming,
diverse, inclusive, and healthy community.
## Our Standards
Examples of behavior that contributes to a positive environment for our
community include:
- Demonstrating empathy and kindness toward other people
- Being respectful of differing opinions, viewpoints, and experiences
- Giving and gracefully accepting constructive feedback
- Accepting responsibility and apologizing to those affected by our mistakes,
and learning from the experience
- Focusing on what is best not just for us as individuals, but for the overall
community
Examples of unacceptable behavior include:
- The use of sexualized language or imagery, and sexual attention or advances of
any kind
- Trolling, insulting or derogatory comments, and personal or political attacks
- Public or private harassment
- Publishing others' private information, such as a physical or email address,
without their explicit permission
- Other conduct which could reasonably be considered inappropriate in a
professional setting
## Enforcement Responsibilities
Community leaders are responsible for clarifying and enforcing our standards of
acceptable behavior and will take appropriate and fair corrective action in
response to any behavior that they deem inappropriate, threatening, offensive,
or harmful.
Community leaders have the right and responsibility to remove, edit, or reject
comments, commits, code, wiki edits, issues, and other contributions that are
not aligned to this Code of Conduct, and will communicate reasons for moderation
decisions when appropriate.
## Scope
This Code of Conduct applies within all community spaces, and also applies when
an individual is officially representing the community in public spaces.
Examples of representing our community include using an official email address,
posting via an official social media account, or acting as an appointed
representative at an online or offline event.
## Enforcement
Instances of abusive, harassing, or otherwise unacceptable behavior may be
reported to the community leaders responsible for enforcement at
[email protected].
All complaints will be reviewed and investigated promptly and fairly.
All community leaders are obligated to respect the privacy and security of the
reporter of any incident.
## Enforcement Guidelines
Community leaders will follow these Community Impact Guidelines in determining
the consequences for any action they deem in violation of this Code of Conduct:
### 1. Correction
**Community Impact**: Use of inappropriate language or other behavior deemed
unprofessional or unwelcome in the community.
**Consequence**: A private, written warning from community leaders, providing
clarity around the nature of the violation and an explanation of why the
behavior was inappropriate. A public apology may be requested.
### 2. Warning
**Community Impact**: A violation through a single incident or series of
actions.
**Consequence**: A warning with consequences for continued behavior. No
interaction with the people involved, including unsolicited interaction with
those enforcing the Code of Conduct, for a specified period of time. This
includes avoiding interactions in community spaces as well as external channels
like social media. Violating these terms may lead to a temporary or permanent
ban.
### 3. Temporary Ban
**Community Impact**: A serious violation of community standards, including
sustained inappropriate behavior.
**Consequence**: A temporary ban from any sort of interaction or public
communication with the community for a specified period of time. No public or
private interaction with the people involved, including unsolicited interaction
with those enforcing the Code of Conduct, is allowed during this period.
Violating these terms may lead to a permanent ban.
### 4. Permanent Ban
**Community Impact**: Demonstrating a pattern of violation of community
standards, including sustained inappropriate behavior, harassment of an
individual, or aggression toward or disparagement of classes of individuals.
**Consequence**: A permanent ban from any sort of public interaction within the
community.
## Attribution
This Code of Conduct is adapted from the [Contributor Covenant][homepage],
version 2.1, available at
[https://www.contributor-covenant.org/version/2/1/code_of_conduct.html][v2.1].
Community Impact Guidelines were inspired by
[Mozilla's code of conduct enforcement ladder][Mozilla CoC].
For answers to common questions about this code of conduct, see the FAQ at
[https://www.contributor-covenant.org/faq][FAQ]. Translations are available at
[https://www.contributor-covenant.org/translations][translations].
[homepage]: https://www.contributor-covenant.org
[v2.1]: https://www.contributor-covenant.org/version/2/1/code_of_conduct.html
[Mozilla CoC]: https://github.com/mozilla/diversity
[FAQ]: https://www.contributor-covenant.org/faq
[translations]: https://www.contributor-covenant.org/translations
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
python-dotenv>=1.0.0
requests>=2.31.0
mcp>=1.0.0
uvicorn>=0.32.1
httpx
```
--------------------------------------------------------------------------------
/armor_crypto_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Armor MCP Package
This package provides an agentic interface for interoperating with multiple blockchains,
staking, DeFi operations, swaps, bridging, wallet management, and for developing crypto
trading strategies through dynamic DCA.
"""
__version__ = "0.2.1"
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
FROM python:3.11-slim
WORKDIR /app
# Copy local code to the container. We assume the MCP base path contains the project files.
COPY . /app
# Install dependencies using pip. Use pip install . to install the package defined by pyproject.toml.
RUN pip install --no-cache-dir .
# Expose port if necessary (optional)
EXPOSE 8000
# Run the MCP server. The entry point is defined by the package's script, so we use the installed command.
CMD ["armor-crypto-mcp"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "armor-crypto-mcp"
version = "0.2.1"
description = "MCP to interface with multiple blockchains, staking, DeFi, swap, bridging, wallet management, DCA, Limit Orders, Coin Lookup, Tracking and more"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"mcp>=1.1.0",
"python-dotenv>=1.0.0",
"requests>=2.31.0",
"uvicorn>=0.32.1",
"httpx"
]
[[project.authors]]
name = "Armor Wallet"
email = "[email protected]"
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[project.scripts]
armor-crypto-mcp = "armor_crypto_mcp.armor_mcp:main"
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
required:
- armorApiKey
properties:
armorApiKey:
type: string
description: API Key for Armor API authentication.
armorApiUrl:
type: string
default: https://app.armorwallet.ai/api/v1
description: The base URL for the Armor API.
commandFunction:
# A JS function that produces the CLI command based on the given config to start the MCP on stdio.
|-
(config) => ({
command: 'armor-crypto-mcp',
args: [],
env: {
ARMOR_API_KEY: config.armorApiKey,
ARMOR_API_URL: config.armorApiUrl || 'https://app.armorwallet.ai/api/v1'
}
})
exampleConfig:
armorApiKey: example_api_key
armorApiUrl: https://app.armorwallet.ai/api/v1
```
--------------------------------------------------------------------------------
/README_prompts.md:
--------------------------------------------------------------------------------
```markdown
# Example Prompts for Armor MCP
## How Armor Crypto MCP Might be Used
Here are some example prompts that we have been testing with our internal Armor Agents, but could use testing with other agents. Think of this MCP as the bridge to a large number of cryptocurrency ecosystems. Each tool we provide your agent can be combined with other tools to form powerful chains of action.
### General Questions
```
What can I do in Armor?
```
```
What tools does Armor make available?
```
### Wallet Management
- Creating wallets
```
Create a wallet named test2 and transfer 0.2 SOL to it from test1
```
- Wallet organization
```
Put wallets test1 and test2 into a new group called testing
```
```
List my wallet groups
```
- Archiving wallets
```
Move all of my assets from test3 to test1 and archive test3
```
### DCA and Swaps
- Simple DCA
```
DCA into SOL from 20% of my USDc
```
- Specific DCA
```
Buy SOL with all of my USD in test1 wallet over a period of 3 months, place the orders at midnight every monday and thursday
```
- Placing Orders
```
Buy 0.12 BTC with my SOL at 10% below current market price
```
```
Get out of SOL now!
```
```
Put a stop loss on all my altcoin positions in test2 wallet
```
- Cancelling Orders
```
Cancel all my open orders
```
```
Cancel all my buy orders below 5% of the current market price in SOL
```
### Helpful Notes
- The more specific you are, the more control you can have over whatever strategy you want.
- It will help if you ask for the current state of your assets to better plan what to do.
- All agents are not created equally, and won't use tools in the same way.
- If your agent has Thinking mode or capability, try using that for a boost.
- Talk to your agent about strategy before commanding it to do something.
- None of this is financial advice.
```
--------------------------------------------------------------------------------
/.github/workflows/python-publish.yml:
--------------------------------------------------------------------------------
```yaml
# This workflow will upload a Python Package to PyPI when a release is created
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries
# This workflow uses actions that are not certified by GitHub.
# They are provided by a third-party and are governed by
# separate terms of service, privacy policy, and support
# documentation.
name: Upload Python Package
on:
release:
types: [published]
permissions:
contents: read
jobs:
release-build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.x"
- name: Build release distributions
run: |
# NOTE: put your own distribution build steps here.
python -m pip install build
python -m build
- name: Upload distributions
uses: actions/upload-artifact@v4
with:
name: release-dists
path: dist/
pypi-publish:
runs-on: ubuntu-latest
needs:
- release-build
permissions:
# IMPORTANT: this permission is mandatory for trusted publishing
id-token: write
# Dedicated environments with protections for publishing are strongly recommended.
# For more information, see: https://docs.github.com/en/actions/deployment/targeting-different-environments/using-environments-for-deployment#deployment-protection-rules
environment:
name: pypi
# OPTIONAL: uncomment and update to include your PyPI project URL in the deployment status:
# url: https://pypi.org/p/YOURPROJECT
#
# ALTERNATIVE: if your GitHub Release name is the PyPI project version string
# ALTERNATIVE: exactly, uncomment the following line instead:
url: https://pypi.org/project/armor-crypto-mcp/${{ github.event.release.name }}
steps:
- name: Retrieve release distributions
uses: actions/download-artifact@v4
with:
name: release-dists
path: dist/
- name: Publish release distributions to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
packages-dir: dist/
```
--------------------------------------------------------------------------------
/armor_crypto_mcp/armor_mcp.py:
--------------------------------------------------------------------------------
```python
import os
import asyncio
from typing import List, Any, Dict
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP, Context
# Import base models
from .armor_client import (
ArmorWalletAPIClient,
calculate,
WalletTokenBalance,
ConversionResponse,
SwapQuoteResponse,
SwapTransactionResponse,
Wallet,
TokenDetailsResponseContainer,
GroupInfo,
SingleGroupInfo,
WalletInfo,
WalletArchiveOrUnarchiveResponse,
CreateGroupResponse,
AddWalletToGroupResponse,
GroupArchiveOrUnarchiveResponse,
RemoveWalletFromGroupResponse,
TransferTokenResponse,
DCAOrderResponse,
CancelDCAOrderResponse,
ListSingleGroupRequest,
TopTrendingTokensRequest,
CandleStickRequest,
StakeBalanceResponse,
ListWalletsRequest,
ListDCAOrderRequest,
ListOrderRequest,
PrivateKeyRequest,
WalletTokenPairsContainer,
ConversionRequestContainer,
SwapQuoteRequestContainer,
SwapTransactionRequestContainer,
TokenDetailsRequestContainer,
TokenSearchRequest,
TokenSearchResponseContainer,
TransferTokensRequestContainer,
DCAOrderRequestContainer,
CancelDCAOrderRequestContainer,
CreateWalletRequestContainer,
ArchiveWalletsRequestContainer,
UnarchiveWalletRequestContainer,
CreateGroupsRequestContainer,
AddWalletToGroupRequestContainer,
ArchiveWalletGroupRequestContainer,
UnarchiveWalletGroupRequestContainer,
RemoveWalletsFromGroupRequestContainer,
CreateOrderRequestContainer,
CancelOrderRequestContainer,
CreateOrderResponseContainer,
CancelOrderResponseContainer,
StakeQuoteRequestContainer,
UnstakeQuoteRequestContainer,
StakeTransactionRequestContainer,
UnstakeTransactionRequestContainer,
RenameWalletRequestContainer,
ListDCAOrderResponseContainer,
ListOrderResponseContainer,
)
# Load environment variables (e.g. BASE_API_URL, etc.)
load_dotenv()
# Create an MCP server instance with FastMCP
mcp = FastMCP("Armor Crypto MCP")
# Global variable to hold the authenticated Armor API client
ACCESS_TOKEN = os.getenv('ARMOR_API_KEY') or os.getenv('ARMOR_ACCESS_TOKEN')
BASE_API_URL = os.getenv('ARMOR_API_URL') or 'https://app.armorwallet.ai/api/v1'
armor_client = ArmorWalletAPIClient(ACCESS_TOKEN, base_api_url=BASE_API_URL) #, log_path='armor_client.log')
# Include version endpoint
from armor_crypto_mcp import __version__
@mcp.tool()
async def get_armor_mcp_version():
"""Get the current Armor Wallet version"""
return {'armor_version': __version__}
@mcp.tool()
async def wait_a_moment(seconds:float):
"""Wait for some short amount of time, no more than 10 seconds"""
await asyncio.sleep(seconds)
return {"waited": seconds}
from datetime import datetime, timezone
@mcp.tool()
async def get_current_time() -> Dict:
"""Gets the current time and date"""
return {"current_time": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")}
@mcp.tool()
async def calculator(expression:str, variables:dict[str, Any]):
"""
Safely evaluates a mathematical or statistical expression string using Python syntax.
Supports arithmetic operations (+, -, *, /, **, %, //), list expressions, and a range of math and statistics functions:
abs, round, min, max, len, sum, mean, median, stdev, variance, sin, cos, tan, sqrt, log, exp, floor, ceil, etc.
Custom variables can be passed via the 'variables' dict, including lists for time series data.
"""
return {'result': calculate(expression, variables)}
@mcp.tool()
async def get_wallet_token_balance(wallet_token_pairs: WalletTokenPairsContainer) -> List[WalletTokenBalance]:
"""
Get the balance for a list of wallet/token pairs.
Expects a WalletTokenPairsContainer, returns a list of WalletTokenBalance.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[WalletTokenBalance] = await armor_client.get_wallet_token_balance(wallet_token_pairs)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def calculate_token_conversion(conversion_requests: ConversionRequestContainer) -> List[ConversionResponse]:
"""
Perform token conversion quote between two tokens. Good for quickly calculating market prices.
Expects a ConversionRequestContainer, returns a list of ConversionResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[ConversionResponse] = await armor_client.conversion_api(conversion_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def swap_quote(swap_quote_requests: SwapQuoteRequestContainer) -> List[SwapQuoteResponse]:
"""
Retrieve a swap quote. Be sure to add slippage!
Expects a SwapQuoteRequestContainer, returns a list of SwapQuoteResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[SwapQuoteResponse] = await armor_client.swap_quote(swap_quote_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def swap_transaction(swap_transaction_requests: SwapTransactionRequestContainer) -> List[SwapTransactionResponse]:
"""
Execute a swap transaction.
Expects a SwapTransactionRequestContainer, returns a list of SwapTransactionResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[SwapTransactionResponse] = await armor_client.swap_transaction(swap_transaction_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def get_all_wallets(get_all_wallets_requests: ListWalletsRequest) -> List[Wallet]:
"""
Retrieve all wallets with balances.
Returns a list of Wallets and asssets
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[Wallet] = await armor_client.get_all_wallets(get_all_wallets_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def get_all_orders(get_all_orders_requests: ListOrderRequest) -> ListOrderResponseContainer:
"""
Retrieve all limit, take profit and stop loss orders.
Returns a list of orders.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: ListOrderResponseContainer = await armor_client.list_orders(get_all_orders_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def search_official_token_address(token_details_requests: TokenDetailsRequestContainer) -> TokenDetailsResponseContainer:
"""
Get the official token address and symbol for a token symbol or token address.
Try to use this first to get address and symbol of coin. If not found, use search_token_details to get details.
Expects a TokenDetailsRequestContainer, returns a TokenDetailsResponseContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: TokenDetailsResponseContainer = await armor_client.get_official_token_address(token_details_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def search_token_details(token_search_requests: TokenSearchRequest) -> TokenSearchResponseContainer:
"""
Search and retrieve details about single token.
If only address or symbol is needed, use get_official_token_address first.
Expects a TokenSearchRequest, returns a list of TokenDetailsResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: TokenSearchResponseContainer = await armor_client.search_token(token_search_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def list_groups() -> List[GroupInfo]:
"""
List all wallet groups.
Returns a list of GroupInfo.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[GroupInfo] = await armor_client.list_groups()
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def list_single_group(list_single_group_requests: ListSingleGroupRequest) -> SingleGroupInfo:
"""
Retrieve details for a single wallet group.
Expects the group name as a parameter, returns SingleGroupInfo.
"""
if not armor_client:
return {"error": "Not logged in"}
try:
result: SingleGroupInfo = await armor_client.list_single_group(list_single_group_requests)
return result
except Exception as e:
return {"error": str(e)}
@mcp.tool()
async def create_wallet(create_wallet_requests: CreateWalletRequestContainer) -> List[WalletInfo]:
"""
Create new wallets.
Expects a list of wallet names, returns a list of WalletInfo.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[WalletInfo] = await armor_client.create_wallet(create_wallet_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def archive_wallets(archive_wallet_requests: ArchiveWalletsRequestContainer) -> List[WalletArchiveOrUnarchiveResponse]:
"""
Archive wallets.
Expects a list of wallet names, returns a list of WalletArchiveOrUnarchiveResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[WalletArchiveOrUnarchiveResponse] = await armor_client.archive_wallets(archive_wallet_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def unarchive_wallets(unarchive_wallet_requests: UnarchiveWalletRequestContainer) -> List[WalletArchiveOrUnarchiveResponse]:
"""
Unarchive wallets.
Expects a list of wallet names, returns a list of WalletArchiveOrUnarchiveResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[WalletArchiveOrUnarchiveResponse] = await armor_client.unarchive_wallets(unarchive_wallet_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def create_groups(create_groups_requests: CreateGroupsRequestContainer) -> List[CreateGroupResponse]:
"""
Create new wallet groups.
Expects a list of group names, returns a list of CreateGroupResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[CreateGroupResponse] = await armor_client.create_groups(create_groups_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def add_wallets_to_group(add_wallet_to_group_requests: AddWalletToGroupRequestContainer) -> List[AddWalletToGroupResponse]:
"""
Add wallets to a specified group.
Expects the group name and a list of wallet names, returns a list of AddWalletToGroupResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[AddWalletToGroupResponse] = await armor_client.add_wallets_to_group(add_wallet_to_group_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def archive_wallet_group(archive_wallet_group_requests: ArchiveWalletGroupRequestContainer) -> List[GroupArchiveOrUnarchiveResponse]:
"""
Archive wallet groups.
Expects a list of group names, returns a list of GroupArchiveOrUnarchiveResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[GroupArchiveOrUnarchiveResponse] = await armor_client.archive_wallet_group(archive_wallet_group_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def unarchive_wallet_group(unarchive_wallet_group_requests: UnarchiveWalletGroupRequestContainer) -> List[GroupArchiveOrUnarchiveResponse]:
"""
Unarchive wallet groups.
Expects a list of group names, returns a list of GroupArchiveOrUnarchiveResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[GroupArchiveOrUnarchiveResponse] = await armor_client.unarchive_wallet_group(unarchive_wallet_group_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def remove_wallets_from_group(remove_wallets_from_group_requests: RemoveWalletsFromGroupRequestContainer) -> List[RemoveWalletFromGroupResponse]:
"""
Remove wallets from a specified group.
Expects the group name and a list of wallet names, returns a list of RemoveWalletFromGroupResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[RemoveWalletFromGroupResponse] = await armor_client.remove_wallets_from_group(remove_wallets_from_group_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def transfer_tokens(transfer_tokens_requests: TransferTokensRequestContainer) -> List[TransferTokenResponse]:
"""
Transfer tokens from one wallet to another.
Expects a TransferTokensRequestContainer, returns a list of TransferTokenResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[TransferTokenResponse] = await armor_client.transfer_tokens(transfer_tokens_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def create_dca_order(dca_order_requests: DCAOrderRequestContainer) -> List[DCAOrderResponse]:
"""
Create a DCA order.
Expects a DCAOrderRequestContainer, returns a list of DCAOrderResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[DCAOrderResponse] = await armor_client.create_dca_order(dca_order_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def list_dca_orders(list_dca_order_requests: ListDCAOrderRequest) -> ListDCAOrderResponseContainer:
"""
List all DCA orders.
Returns a list of DCAOrderResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: ListDCAOrderResponseContainer = await armor_client.list_dca_orders(list_dca_order_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def cancel_dca_order(cancel_dca_order_requests: CancelDCAOrderRequestContainer) -> List[CancelDCAOrderResponse]:
"""
Create a DCA order.
Note: Make a single or multiple dca_order_requests
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List[CancelDCAOrderResponse] = await armor_client.cancel_dca_order(cancel_dca_order_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def create_order(create_order_requests: CreateOrderRequestContainer) -> CreateOrderResponseContainer:
"""
Create a order. Can be a limit, take profit or stop loss order.
Expects a CreateOrderRequestContainer, returns a CreateOrderResponseContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: CreateOrderResponseContainer = await armor_client.create_order(create_order_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def cancel_order(cancel_order_requests: CancelOrderRequestContainer) -> CancelOrderResponseContainer:
"""
Cancel a limit, take profit or stop loss order.
Expects a CancelOrderRequestContainer, returns a CancelOrderResponseContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: CancelOrderResponseContainer = await armor_client.cancel_order(cancel_order_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def stake_quote(stake_quote_requests: StakeQuoteRequestContainer) -> SwapQuoteRequestContainer:
"""
Retrieve a stake quote.
Expects a StakeQuoteRequestContainer, returns a SwapQuoteRequestContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: StakeQuoteRequestContainer = await armor_client.stake_quote(stake_quote_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def unstake_quote(unstake_quote_requests: UnstakeQuoteRequestContainer) -> SwapQuoteRequestContainer:
"""
Retrieve an unstake quote.
Expects a UnstakeQuoteRequestContainer, returns a SwapQuoteRequestContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: UnstakeQuoteRequestContainer = await armor_client.unstake_quote(unstake_quote_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def stake_transaction(stake_transaction_requests: StakeTransactionRequestContainer) -> SwapTransactionRequestContainer:
"""
Execute a stake transaction.
Expects a StakeTransactionRequestContainer, returns a SwapTransactionRequestContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: SwapTransactionRequestContainer = await armor_client.stake_transaction(stake_transaction_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def unstake_transaction(unstake_transaction_requests: UnstakeTransactionRequestContainer) -> SwapTransactionRequestContainer:
"""
Execute an unstake transaction.
Expects a UnstakeTransactionRequestContainer, returns a SwapTransactionRequestContainer.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: SwapTransactionRequestContainer = await armor_client.unstake_transaction(unstake_transaction_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def get_top_trending_tokens(top_trending_tokens_requests: TopTrendingTokensRequest) -> List:
"""
Get the top trending tokens in a particular time frame. Great for comparing market cap or volume.
Expects a TopTrendingTokensRequest, returns a list of tokens with their details.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List = await armor_client.top_trending_tokens(top_trending_tokens_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def get_stake_balances() -> StakeBalanceResponse:
"""
Get the balance of staked SOL (jupSOL).
Returns a StakeBalanceResponse.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: StakeBalanceResponse = await armor_client.get_stake_balances()
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def rename_wallets(rename_wallet_requests: RenameWalletRequestContainer) -> List:
"""
Rename wallets.
Expects a RenameWalletRequestContainer, returns a list.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List = await armor_client.rename_wallet(rename_wallet_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.tool()
async def get_token_candle_data(candle_stick_requests: CandleStickRequest) -> List:
"""
Get candle data about any token for analysis.
Expects a CandleStickRequest, returns a list of candle sticks.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: List = await armor_client.get_market_candle_data(candle_stick_requests)
return result
except Exception as e:
return [{"error": str(e)}]
@mcp.prompt()
def login_prompt(email: str) -> str:
"""
A sample prompt to ask the user for their access token after providing an email.
"""
return f"Please enter the Access token for your account {email}."
@mcp.tool()
async def send_key_to_telegram(private_key_request: PrivateKeyRequest) -> Dict:
"""
Send the mnemonic or private key to telegram.
"""
if not armor_client:
return [{"error": "Not logged in"}]
try:
result: Dict = await armor_client.send_key_to_telegram(private_key_request)
return result
except Exception as e:
return [{"error": str(e)}]
def main():
mcp.run()
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/armor_crypto_mcp/armor_client.py:
--------------------------------------------------------------------------------
```python
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)
```