MCP (Model Context Protocol) for Crypto Bots: Connect Your AI Agent to Any Data Source

What Is MCP and Why It Matters for Crypto Agents

Model Context Protocol (MCP) is an open standard created by Anthropic that defines how AI models interact with external tools and data sources. Think of it as a USB standard for AI — any MCP-compatible client (Claude, Cursor, any LLM tool) can connect to any MCP server.

For crypto trading agents, MCP unlocks:

• Live price data: Your agent reads real prices, not training data
• Exchange execution: Agent places real orders through MCP
• On-chain queries: Agent reads blockchain state directly
• Portfolio data: Agent knows your exact positions

Without MCP, you'd write custom integration code for every tool. With MCP, you build one server and any AI agent can use it.

Building a Crypto MCP Server

# Install: pip install mcp anthropic ccxt
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import ccxt
import json
import asyncio

app = Server("crypto-trading-server")

# Initialize exchange
exchange = ccxt.binance({
    'apiKey': 'YOUR_BINANCE_KEY',
    'secret': 'YOUR_BINANCE_SECRET',
    'enableRateLimit': True,
})

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Tell MCP clients what tools this server provides"""
    return [
        Tool(
            name="get_price",
            description="Get the current price of any cryptocurrency pair",
            inputSchema={
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "Trading pair e.g. BTC/USDT"}
                },
                "required": ["symbol"]
            }
        ),
        Tool(
            name="get_portfolio",
            description="Get current portfolio balances across all assets",
            inputSchema={"type": "object", "properties": {}}
        ),
        Tool(
            name="place_order",
            description="Place a market buy or sell order",
            inputSchema={
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "Trading pair"},
                    "side": {"type": "string", "enum": ["buy", "sell"]},
                    "amount_usd": {"type": "number", "description": "Amount in USD to buy/sell"},
                },
                "required": ["symbol", "side", "amount_usd"]
            }
        ),
        Tool(
            name="get_ohlcv",
            description="Get historical OHLCV candlestick data for analysis",
            inputSchema={
                "type": "object",
                "properties": {
                    "symbol": {"type": "string"},
                    "timeframe": {"type": "string", "description": "1m, 5m, 1h, 4h, 1d"},
                    "limit": {"type": "integer", "description": "Number of candles"},
                },
                "required": ["symbol", "timeframe"]
            }
        ),
        Tool(
            name="get_funding_rate",
            description="Get perpetual futures funding rate for a symbol",
            inputSchema={
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "Perp symbol e.g. BTC/USDT:USDT"}
                },
                "required": ["symbol"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Handle tool calls from AI clients"""
    
    try:
        if name == "get_price":
            ticker = exchange.fetch_ticker(arguments["symbol"])
            result = {
                "symbol": arguments["symbol"],
                "price": ticker["last"],
                "change_24h": ticker.get("percentage", 0),
                "volume_24h": ticker.get("quoteVolume", 0),
                "bid": ticker["bid"],
                "ask": ticker["ask"],
            }
            return [TextContent(type="text", text=json.dumps(result, indent=2))]
        
        elif name == "get_portfolio":
            balance = exchange.fetch_balance()
            portfolio = {
                asset: {
                    "free": amounts["free"],
                    "used": amounts["used"],
                    "total": amounts["total"]
                }
                for asset, amounts in balance.items()
                if isinstance(amounts, dict) and amounts.get("total", 0) > 0
                and asset not in ["info", "free", "used", "total", "debt"]
            }
            return [TextContent(type="text", text=json.dumps(portfolio, indent=2))]
        
        elif name == "place_order":
            symbol = arguments["symbol"]
            side = arguments["side"]
            amount_usd = arguments["amount_usd"]
            
            # Safety check: max $500 per AI-initiated order
            if amount_usd > 500:
                return [TextContent(type="text", text=json.dumps({
                    "error": "Order exceeds AI limit of $500. Adjust manually."
                }))]
            
            ticker = exchange.fetch_ticker(symbol)
            price = ticker["last"]
            qty = amount_usd / price
            
            order = exchange.create_market_order(symbol, side, qty)
            
            return [TextContent(type="text", text=json.dumps({
                "status": "filled",
                "order_id": order["id"],
                "side": side,
                "symbol": symbol,
                "qty": order["filled"],
                "avg_price": order["average"],
                "total_usd": order["filled"] * order["average"],
            }, indent=2))]
        
        elif name == "get_ohlcv":
            ohlcv = exchange.fetch_ohlcv(
                arguments["symbol"],
                arguments["timeframe"],
                limit=arguments.get("limit", 50)
            )
            formatted = [
                {"time": c[0], "open": c[1], "high": c[2], "low": c[3], "close": c[4], "volume": c[5]}
                for c in ohlcv[-20:]  # Last 20 candles
            ]
            return [TextContent(type="text", text=json.dumps(formatted, indent=2))]
        
        elif name == "get_funding_rate":
            funding = exchange.fetch_funding_rate(arguments["symbol"])
            return [TextContent(type="text", text=json.dumps({
                "symbol": arguments["symbol"],
                "funding_rate": funding["fundingRate"],
                "annualized_pct": funding["fundingRate"] * 3 * 365 * 100,
                "next_funding": funding.get("nextFundingDatetime"),
            }, indent=2))]
        
        else:
            return [TextContent(type="text", text=f"Unknown tool: {name}")]
    
    except Exception as e:
        return [TextContent(type="text", text=json.dumps({"error": str(e)}))]

async def main():
    async with stdio_server() as streams:
        await app.run(*streams, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Connecting Claude to Your MCP Server

// ~/.cursor/mcp_settings.json or Claude Desktop config
{
  "mcpServers": {
    "crypto-trading": {
      "command": "python",
      "args": ["/path/to/your/crypto_mcp_server.py"],
      "env": {
        "BINANCE_KEY": "your-key",
        "BINANCE_SECRET": "your-secret"
      }
    }
  }
}

Once connected, Claude can answer questions like:

• "What's my current portfolio worth?"
• "Show me the BTC 4h chart and tell me if I should buy"
• "Check the ETH funding rate and if it's above 0.05%, sell 50% of my ETH and short the perp"

Building an On-Chain MCP Server

from web3 import Web3

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_defi_position",
            description="Get current DeFi position on Aave, Uniswap, or Compound",
            inputSchema={
                "type": "object",
                "properties": {
                    "protocol": {"type": "string", "enum": ["aave", "uniswap", "compound"]},
                    "wallet": {"type": "string", "description": "Ethereum address"},
                },
                "required": ["protocol", "wallet"]
            }
        ),
        Tool(
            name="get_gas_price",
            description="Get current Ethereum gas prices",
            inputSchema={"type": "object", "properties": {}}
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_gas_price":
        w3 = Web3(Web3.HTTPProvider('https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY'))
        gas = w3.eth.gas_price
        return [TextContent(type="text", text=json.dumps({
            "gwei": w3.from_wei(gas, 'gwei'),
            "recommended_priority": "2-5 gwei for standard tx"
        }))]

Why MCP Is the Future of AI Agents

Before MCP, giving an AI agent access to tools required:

1. Custom function calling code for each LLM
2. Different integration patterns per model provider
3. Security challenges with direct API key sharing

With MCP:

• One server works with Claude, Cursor, and any MCP client
• Clean separation between AI logic and tool implementation
• Standardized security model

MCP is to AI agents what REST APIs were to web development — it creates a universal interface that enables an ecosystem. Building MCP servers for your trading infrastructure today puts you ahead of 99% of developers.