加密货币投资组合管理服务器

Get wallet portfolio across all chains:
Use getWalletPortfolio with:
- address: "0x42b9dF65B219B3dD36FF330A4dD8f327A6Ada990"
- currency: "usd"

Get positions across all chains:
Use listWalletPositions with:
- address: "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
- filter[positions]: "only_complex"
- currency: "usd"

Get all DeFi positions with protocol data:
Use listWalletPositions with:
- address: "0x42b9dF65B219B3dD36FF330A4dD8f327A6Ada990"
- filter[positions]: "only_complex"
- filter[trash]: "only_non_trash"
- currency: "usd"

Filter positions by specific protocol:
Use listWalletPositions with:
- address: "0x..."
- filter[dapp_ids]: "aave-v3,compound-v2"
- filter[positions]: "only_complex"

Get NFT with full metadata:
Use getNFTById with:
- nft_id: "ethereum:0x74ee68a33f6c9f113e22b3b77418b75f85d07d22:10"
- currency: "usd"

Get wallet NFT positions:
Use listWalletNFTPositions with:
- address: "0x..."
- filter[trash]: "only_non_trash"

# Clone the repository
git clone https://github.com/SAK1337/myzerionmcp.git
cd myzerionmcp

# Create .env file with your API key
cp .env.example .env
# Edit .env and set ZERION_API_KEY

# Start with Docker Compose
docker-compose up -d

# View logs
docker-compose logs -f

# Clone the repository
git clone https://github.com/SAK1337/myzerionmcp.git
cd myzerionmcp

# Install the package
pip install -e .

# For development (includes testing dependencies)
pip install -e ".[dev]"

pip install zerion-mcp-server

export ZERION_API_KEY="Bearer your-api-key-here"

zerion-mcp-server

{
  "mcpServers": {
    "zerion": {
      "command": "zerion-mcp-server",
      "env": {
        "ZERION_API_KEY": "Bearer your-api-key-here"
      }
    }
  }
}

# Server configuration
name: "Zerion API"
base_url: "https://api.zerion.io"
oas_url: "https://raw.githubusercontent.com/smart-mcp-proxy/zerion-mcp-server/main/zerion_mcp_server/openapi_zerion.yaml"

# API authentication
api_key: "${ZERION_API_KEY}"  # Environment variable substitution

# Logging configuration
logging:
  level: "INFO"      # DEBUG, INFO, WARN, ERROR
  format: "text"     # text or json

Get the portfolio balance for wallet 0x1234...

Show me the DeFi positions for address 0xabcd...

List NFTs owned by 0x5678...

┌──────────────┐
│  AI Client   │ (Claude Desktop)
└──────┬───────┘
       │ stdio (MCP protocol)
       ▼
┌──────────────┐
│  MCP Server  │ - Manages subscriptions via API calls
└──────┬───────┘
       │ HTTPS
       ▼
┌──────────────┐
│  Zerion API  │
└──────┬───────┘
       │ Webhooks (HTTP POST)
       ▼
┌──────────────┐
│ Your HTTP    │ - Receives webhook payloads (separate service)
│ Receiver     │
└──────────────┘

Use the createTxSubscription tool with:
- addresses: ["0x42b9dF65B219B3dD36FF330A4dD8f327A6Ada990"]
- callback_url: "https://webhook.site/your-unique-url"
- chain_ids: ["ethereum", "base"]

Use the createTxSubscription tool with:
- addresses: ["8BH9pjtgyZDC4iAQH5ZiYDZ1MDWC98xki2V8NzqqKW3K"]
- callback_url: "https://your-server.com/webhooks/zerion"
- chain_ids: ["solana"]

{
  "data": {
    "type": "callback",
    "id": "bf300927-3f57-4d00-a01a-f7b75bd9b8de",
    "attributes": {
      "address": "0x42b9dF65B219B3dD36FF330A4dD8f327A6Ada990",
      "callback_url": "https://webhook.site/test",
      "timestamp": "2025-10-16T10:56:50Z"
    },
    "relationships": {
      "subscription": {
        "id": "61f13641-443e-4068-932b-c28edeaefd85",
        "type": "tx-subscriptions"
      }
    }
  },
  "included": [
    {
      "type": "transactions",
      "id": "13de850a-bfa4-54c7-a7bb-fd6371d98894",
      "attributes": {
        "operation_type": "trade",
        "hash": "0x...",
        "mined_at": "2025-10-16T10:56:48Z",
        "status": "confirmed",
        "fee": { ... },
        "transfers": [ ... ]
      },
      "relationships": {
        "chain": { "id": "ethereum", "type": "chains" },
        "dapp": { "id": "uniswap", "type": "dapps" }
      }
    }
  ]
}

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhooks/zerion', methods=['POST'])
def handle_zerion_webhook():
    payload = request.json

    # Extract transaction data
    callback_data = payload['data']
    transactions = payload.get('included', [])

    for tx in transactions:
        if tx['type'] == 'transactions':
            print(f"New transaction: {tx['attributes']['hash']}")
            print(f"Operation: {tx['attributes']['operation_type']}")
            print(f"Chain: {tx['relationships']['chain']['id']}")

    return jsonify({"status": "received"}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

const express = require('express');
const app = express();

app.use(express.json());

app.post('/webhooks/zerion', (req, res) => {
  const { data, included } = req.body;

  const transactions = included.filter(item => item.type === 'transactions');

  transactions.forEach(tx => {
    console.log(`New ${tx.attributes.operation_type}: ${tx.attributes.hash}`);
  });

  res.status(200).json({ status: 'received' });
});

app.listen(8080, () => console.log('Webhook receiver running on port 8080'));

Use listWalletPositions with:
- address: "0x42b9dF65B219B3dD36FF330A4dD8f327A6Ada990"
- filter[positions]: "only_complex"
- filter[trash]: "only_non_trash"
- currency: "usd"

Use listWalletTransactions with:
- address: "0x..."
- filter[chain_ids]: "ethereum,base"
- filter[trash]: "only_non_trash"

Use listWalletPositions with:
- address: "0x..."
- filter[chain_ids]: "optimism"

Use listWalletTransactions with:
- address: "0x..."
- filter[trash]: "only_non_trash"
- page[size]: 100

Use listWalletTransactions with:
- address: "0x..."
- filter[operation_types]: "trade"
- filter[chain_ids]: "ethereum,base"

Use listWalletPositions with:
- address: "0x..."
- filter[positions]: "only_complex"
- filter[position_types]: "staked,reward"

- Get all positions for address
- Filter out spam in your app
- Filter to Ethereum-only in your app

Use listWalletPositions with:
- filter[chain_ids]: "ethereum"
- filter[trash]: "only_non_trash"

Use listWalletPositions with:
- filter[positions]: "only_complex"
- filter[chain_ids]: "ethereum"
- filter[trash]: "only_non_trash"
- currency: "usd"
- sort: "value"

Use listWalletTransactions with:
- page[size]: 50
- filter[trash]: "only_non_trash"

Use listWalletTransactions with:
- address: "0x..."
- page[size]: 100
- filter[trash]: "only_non_trash"

{
  "data": [...],
  "links": {
    "next": "https://api.zerion.io/v1/wallets/.../transactions?page[after]=cursor123"
  }
}

Use listWalletTransactions with:
- address: "0x..."
- page[size]: 100
- page[after]: "cursor123"

from zerion_mcp_server.pagination import fetch_all_pages
from zerion_mcp_server import RetryAsyncClient

# Create client
client = RetryAsyncClient(
    base_url="https://api.zerion.io",
    headers={"Authorization": "Bearer your-key"}
)

# Define your API call
async def get_transactions(address, **kwargs):
    response = await client.get(
        f"/v1/wallets/{address}/transactions",
        params=kwargs
    )
    return response.json()

# Fetch all transactions (up to max_pages)
all_transactions = await fetch_all_pages(
    api_call=lambda **kw: get_transactions("0x123...", **kw),
    max_pages=50,  # Safety limit
    page_size=100
)

print(f"Total transactions: {len(all_transactions)}")

pagination:
  # Default page size (max: 100)
  default_page_size: 100

  # Safety limit for auto-pagination
  max_auto_pages: 50

listWalletTransactions:
  - address: "0x..."
  - page[size]: 100

listWalletTransactions:
  - address: "0x..."
  - filter[chain_ids]: "ethereum"
  - filter[trash]: "only_non_trash"
  - page[size]: 100

INFO: Fetched 10 pages - quota impact may be significant
WARNING: Fetched 25 pages - high quota usage
WARNING: Reached max page limit (50) - results may be incomplete

Request → 429 (rate limit)
 ↓ Wait 1 second
Retry 1 → 429
 ↓ Wait 2 seconds (exponential backoff)
Retry 2 → 429
 ↓ Wait 4 seconds
Retry 3 → 200 OK (success!)

retry_policy:
  # Maximum retry attempts before raising error
  max_attempts: 5

  # Base delay for exponential backoff (seconds)
  base_delay: 1

  # Maximum delay between retries (seconds)
  max_delay: 60

  # Exponential backoff multiplier
  exponential_base: 2

RateLimitError: Rate limit exceeded after 5 retry attempts.
Retry after 60 seconds. Consider upgrading tier or reducing
request frequency. See: https://zerion.io/pricing

# Check for new transactions every 10 seconds
while True:
    transactions = get_transactions(address)
    time.sleep(10)

# Webhook delivers new transactions as they happen
# 0 polling requests, only create subscription once
create_webhook_subscription(addresses=[address])

from functools import lru_cache
import time

@lru_cache(maxsize=100)
def get_portfolio(address, ttl_hash):
    return fetch_wallet_portfolio(address)

# Cache for 5 minutes
def get_ttl_hash(seconds=300):
    return round(time.time() / seconds)

# Usage
portfolio = get_portfolio(address, get_ttl_hash())

Use listWalletTransactions with:
- filter[chain_ids]: "ethereum"
- filter[trash]: "only_non_trash"
- page[size]: 50

WARNING: Rate limit exceeded (url=/v1/wallets/.../transactions, retry_after=30s)
INFO: Retrying request after rate limit (attempt=2)
INFO: Request succeeded after 2 retry attempts

logging:
  level: "DEBUG"

retry_policy:
  max_attempts: 0  # Disable retries

Request → 202 Accepted (indexing...)
 ↓ Wait 3 seconds
Retry 1 → 202 Accepted (still indexing...)
 ↓ Wait 3 seconds
Retry 2 → 200 OK (indexing complete!)

wallet_indexing:
  # Delay between retries (seconds)
  retry_delay: 3

  # Maximum retry attempts
  max_retries: 3

  # Automatically retry (recommended: true)
  auto_retry: true

WalletIndexingError: Wallet is still being indexed by Zerion.
Tried 3 times over 9 seconds. Please retry in 30-60 seconds.

wallet_indexing:
  auto_retry: false

WalletIndexingError: Wallet indexing in progress. This is a
new wallet address for Zerion. Enable auto_retry in
configuration or wait and retry manually.

INFO: Wallet indexing in progress, will retry (retry_delay=3s, max_retries=3)
INFO: Retrying wallet indexing request (attempt=1/3)
INFO: Wallet indexing completed successfully (attempts=2, total_wait=6s)

WARNING: Wallet indexing timeout (attempts=3, total_wait=9s)

Use listWalletPositions with:
- address: "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb"
- X-Env: "testnet"
- currency: "usd"

# Test on Sepolia first
Use getWalletPortfolio with:
- address: "0x..."
- X-Env: "testnet"
- currency: "usd"

# Then move to mainnet
Use getWalletPortfolio with:
- address: "0x..."
- currency: "usd"

Use getWalletPortfolio with:
- address: "0x..."
- currency: "usd"

Use getWalletPortfolio with:
- address: "0x..."
- currency: "eth"

Use listWalletPositions with:
- address: "0x..."
- currency: "eur"
- filter[chain_ids]: "ethereum"

Use getWalletChart with:
- address: "0x..."
- currency: "btc"
- period: "month"

Use listWalletPositions with:
- address: "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
- filter[positions]: "only_complex"
- currency: "eth"

Use getWalletPortfolio with:
- address: "0x..."
- currency: "eur"

Use getWalletChart with:
- address: "0x..."
- currency: "btc"
- period: "year"

# Clone and install with dev dependencies
git clone https://github.com/SAK1337/myzerionmcp.git
cd myzerionmcp
pip install -e ".[dev]"

# Set up API key
export ZERION_API_KEY="Bearer your-test-key"

# Run all tests
pytest

# Run with coverage
pytest --cov=zerion_mcp_server --cov-report=html

# Run specific test file
pytest tests/test_config.py

# Run with verbose output
pytest -v

# Type checking (if mypy is installed)
mypy zerion_mcp_server

# Linting (if ruff is installed)
ruff check zerion_mcp_server

export ZERION_API_KEY="Bearer your-api-key-here"

export ZERION_API_KEY="Bearer your-actual-key"

export LOG_LEVEL="DEBUG"
zerion-mcp-server

logging:
  level: "DEBUG"
  format: "json"  # Structured logs for analysis

┌─────────────────┐
│   MCP Client    │ (e.g., Claude Desktop)
│  (AI Assistant) │
└────────┬────────┘
         │ MCP Protocol
         │
┌────────▼────────┐
│  FastMCP Server │
│                 │
│  ┌───────────┐  │
│  │ Config    │  │ (YAML + Env)
│  │ Manager   │  │
│  └───────────┘  │
│                 │
│  ┌───────────┐  │
│  │  Logger   │  │ (Structured)
│  └───────────┘  │
│                 │
│  ┌───────────┐  │
│  │  Error    │  │ (Custom Exceptions)
│  │  Handler  │  │
│  └───────────┘  │
│                 │
│  ┌───────────┐  │
│  │ HTTP      │  │ (httpx AsyncClient)
│  │ Client    │  │
│  └─────┬─────┘  │
└────────┼────────┘
         │
         │ HTTPS
         │
┌────────▼────────┐
│   Zerion API    │
│                 │
│  - Portfolios   │
│  - DeFi         │
│  - NFTs         │
│  - Transactions │
└─────────────────┘