nbulian/bitcoin-mcp-server

bitcoin-mcp-server/
├── src/                              # Main source code directory
│   ├── __init__.py                   # Package initialization
│   ├── main.py                       # FastAPI application entry point and MCP routing
│   ├── config.py                     # Configuration management with Pydantic
│   ├── bitcoin_client.py             # Bitcoin RPC client with connection management
│   ├── mcp_protocol.py               # MCP Protocol implementation
│   ├── tools/                        # MCP tool implementations
│   │   ├── __init__.py              # Tools package initialization
│   │   ├── blockchain.py            # Blockchain query tools (blocks, transactions)
│   │   ├── network.py               # Network monitoring tools (mempool, mining)
│   │   ├── address.py               # Address analysis tools (balance, UTXOs)
│   │   └── market.py                # Market data tools (price, stats)
│   └── utils/                        # Utility modules
│       ├── __init__.py              # Utils package initialization
│       ├── validation.py            # Input validation utilities
│       └── errors.py                # Custom error classes and handling
├── requirements.txt                  # Python dependencies
├── Dockerfile                        # Docker container configuration
├── docker-compose.yml               # Docker Compose for easy deployment
├── .env.example                      # Environment variables template
└── README.md                         # Project documentation

{
  "jsonrpc": "2.0",
  "error": {
    "code": -32002,
    "message": "Invalid Bitcoin address format",
    "data": {
      "field": "address"
    }
  },
  "id": 1
}

{
  "jsonrpc": "2.0",
  "result": {
    // Method-specific data
  },
  "id": 1
}

{
  "jsonrpc": "2.0",
  "result": {
    "chain": "main",
    "blocks": 800000,
    "headers": 800000,
    "bestblockhash": "0000000000000000000...",
    "difficulty": 123456789.123456789,
    "mediantime": 1640995200,
    "verificationprogress": 0.999999,
    "initialblockdownload": false,
    "chainwork": "0000000000000000000...",
    "size_on_disk": 123456789,
    "pruned": false,
    "pruneheight": 0,
    "automatic_pruning": false,
    "prune_target_size": 0,
    "softforks": {...},
    "bip9_softforks": {...},
    "warnings": ""
  },
  "id": 1
}

{
  "jsonrpc": "2.0",
  "result": {
    "address": "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa",
    "is_valid": true,
    "is_script": false,
    "is_witness": false,
    "witness_version": null,
    "witness_program": null,
    "script_type": "pubkeyhash",
    "address_type": "P2PKH"
  },
  "id": 2
}

{
  "jsonrpc": "2.0",
  "result": {
    "currency": "usd",
    "price": 45000.50,
    "change_24h": 2.5,
    "change_percentage_24h": 5.88,
    "market_cap": 850000000000,
    "volume_24h": 25000000000,
    "last_updated": "2024-01-01T12:00:00Z"
  },
  "id": 3
}

async def new_method(self, param1: str, param2: int) -> Dict[str, Any]:
    """Method description."""
    # Validate inputs
    if not param1:
        raise ValidationError("Missing parameter", "param1")
    
    # Implementation
    result = await self.client.some_rpc_call(param1, param2)
    return result

elif method == "new_method":
    param1 = params.get("param1")
    param2 = params.get("param2", default_value)
    if not param1:
        raise ValidationError("Missing 'param1' parameter", "param1")
    return await tool_instance.new_method(param1, param2)

"available_methods": [
    # ... existing methods
    "new_method",
]

# Always use custom exceptions for consistent error responses
try:
    result = await some_operation()
    return result
except SomeSpecificError as e:
    raise ValidationError(f"Operation failed: {str(e)}", "field_name")
except Exception as e:
    raise BitcoinRPCError(f"Unexpected error: {str(e)}")

# Use context managers for resource cleanup
class NewTool:
    async def __aenter__(self):
        self.client = httpx.AsyncClient()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if hasattr(self, 'client'):
            await self.client.aclose()

# Add new config parameters in config.py
class Config(BaseSettings):
    NEW_API_KEY: Optional[str] = None
    NEW_API_URL: str = "https://api.example.com"
    
    @validator('NEW_API_URL')
    def validate_new_api_url(cls, v):
        if not v.startswith('http'):
            raise ValueError('API URL must start with http')
        return v

# Test blockchain info
curl -X POST http://localhost:8000/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "get_blockchain_info",
    "params": {},
    "id": 1
  }'

# Test address validation
curl -X POST http://localhost:8000/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "validate_address",
    "params": {"address": "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa"},
    "id": 2
  }'

# Test rate limiting
for i in {1..65}; do
  curl -X POST http://localhost:8000/ \
    -H "Content-Type: application/json" \
    -d '{"jsonrpc": "2.0", "method": "get_blockchain_info", "params": {}, "id": '$i'}'
done

# Example test structure (add to tests/ directory)
import pytest
import httpx
from src.main import app

@pytest.fixture
async def client():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client

async def test_get_blockchain_info(client):
    response = await client.post("/", json={
        "jsonrpc": "2.0",
        "method": "get_blockchain_info",
        "params": {},
        "id": 1
    })
    assert response.status_code == 200
    data = response.json()
    assert data["jsonrpc"] == "2.0"
    assert "result" in data

async def test_invalid_method(client):
    response = await client.post("/", json={
        "jsonrpc": "2.0",
        "method": "invalid_method",
        "params": {},
        "id": 1
    })
    assert response.status_code == 200
    data = response.json()
    assert data["error"]["code"] == -32601

# Add caching for expensive operations (future enhancement)
from functools import lru_cache
from datetime import datetime, timedelta

class CachedTool:
    def __init__(self):
        self.cache = {}
        self.cache_ttl = timedelta(minutes=5)
    
    async def cached_method(self, key: str):
        now = datetime.utcnow()
        if key in self.cache:
            data, timestamp = self.cache[key]
            if now - timestamp < self.cache_ttl:
                return data
        
        # Fetch new data
        result = await self.expensive_operation(key)
        self.cache[key] = (result, now)
        return result

# Never commit sensitive data
BITCOIN_RPC_PASSWORD=your_secure_password
API_KEYS=your_api_keys

# Use different credentials for different environments
BITCOIN_RPC_URL_MAINNET=https://mainnet.node.com/
BITCOIN_RPC_URL_TESTNET=https://testnet.node.com/

# Always validate and sanitize inputs
def sanitize_input(value: str, max_length: int = 100) -> str:
    if not isinstance(value, str):
        raise ValidationError("Input must be string")
    
    # Remove potentially dangerous characters
    sanitized = re.sub(r'[^\w\-.]', '', value)
    
    if len(sanitized) > max_length:
        raise ValidationError(f"Input too long, max {max_length} characters")
    
    return sanitized

git clone <repository-url>
cd bitcoin-mcp-server
cp .env.example .env

pip install -r requirements.txt

# Edit .env file with your Bitcoin RPC credentials
BITCOIN_RPC_URL=https://bitcoin.two.nono.casa/
BITCOIN_RPC_USER=user
BITCOIN_RPC_PASSWORD=pass

python -m uvicorn src.main:app --reload --host 0.0.0.0 --port 8000

docker-compose up -d

curl http://localhost:8000/health

docker-compose logs -f bitcoin-mcp-server

services:
  - type: web
    name: bitcoin-mcp-server
    env: docker
    dockerfilePath: ./Dockerfile
    plan: starter
    envVars:
      - key: PORT
        value: 10000
      - key: BITCOIN_RPC_URL
        value: https://bitcoin.two.nono.casa/
      - key: BITCOIN_RPC_USER
        value: user
      - key: BITCOIN_RPC_PASSWORD
        value: pass
    healthCheckPath: /health

{
  "jsonrpc": "2.0",
  "method": "get_current_price",
  "params": {"currency": "usd"},
  "id": 1
}

{
  "jsonrpc": "2.0",
  "method": "analyze_address_activity",
  "params": {"address": "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa"},
  "id": 2
}

{
  "jsonrpc": "2.0",
  "method": "get_latest_blocks",
  "params": {"count": 5},
  "id": 3
}

# Check server health
curl http://localhost:8000/health

# Response format
{
  "status": "healthy",
  "bitcoin_connected": true,
  "timestamp": "2024-01-01T12:00:00.000Z"
}

# Logs are structured with timestamps and levels
2024-01-01 12:00:00,000 - bitcoin_mcp - INFO - Bitcoin MCP Server started on 0.0.0.0:8000
2024-01-01 12:00:01,000 - bitcoin_mcp - ERROR - Bitcoin RPC Error: Connection refused
2024-01-01 12:00:02,000 - bitcoin_mcp - DEBUG - Rate limit check passed for client

# View real-time logs
docker-compose logs -f bitcoin-mcp-server

# View last 100 lines
docker-compose logs --tail=100 bitcoin-mcp-server

# Check if Bitcoin node is accessible
curl --user user:pass --data-binary '{"jsonrpc":"1.0","id":"test","method":"getblockchaininfo","params":[]}' \
  -H 'content-type: text/plain;' https://bitcoin.two.nono.casa/

# If this fails, check:
# 1. Network connectivity
# 2. RPC credentials
# 3. Node availability

{
  "jsonrpc": "2.0",
  "error": {
    "code": -32004,
    "message": "Rate limit of 60 requests per minute exceeded"
  },
  "id": 1
}
// Solution: Increase RATE_LIMIT_PER_MINUTE or implement client-side throttling

# Common causes:
# 1. Invalid address format
# 2. Wrong network (mainnet address on testnet)
# 3. Unsupported address type

# Debug with validate_address method
{
  "jsonrpc": "2.0",
  "method": "validate_address",
  "params": {"address": "your_address_here"},
  "id": 1
}

# In .env file
DEBUG=true

# Or via environment
DEBUG=true python -m uvicorn src.main:app --reload

import logging
logger = logging.getLogger(__name__)

async def your_method(self):
    logger.debug(f"Processing request with params: {params}")
    try:
        result = await some_operation()
        logger.info(f"Operation successful: {result}")
        return result
    except Exception as e:
        logger.error(f"Operation failed: {str(e)}", exc_info=True)
        raise

# Add IP whitelisting for production
ALLOWED_IPS = ["127.0.0.1", "10.0.0.0/8", "192.168.0.0/16"]

@app.middleware("http")
async def ip_whitelist_middleware(request: Request, call_next):
    client_ip = request.client.host
    if client_ip not in ALLOWED_IPS:
        return JSONResponse(
            status_code=403,
            content={"error": "Access denied"}
        )
    return await call_next(request)

# Implement comprehensive input validation
def validate_amount(amount: Union[str, int, float]) -> float:
    try:
        amount_float = float(amount)
        if amount_float < 0:
            raise ValidationError("Amount cannot be negative")
        if amount_float > 21_000_000:  # Max Bitcoin supply
            raise ValidationError("Amount exceeds maximum Bitcoin supply")
        return amount_float
    except (ValueError, TypeError):
        raise ValidationError("Invalid amount format")

# Sanitize string inputs
def sanitize_string(value: str, max_length: int = 255) -> str:
    if not isinstance(value, str):
        raise ValidationError("Value must be a string")
    
    # Remove potentially dangerous characters
    sanitized = re.sub(r'[<>"\']', '', value.strip())
    
    if len(sanitized) > max_length:
        raise ValidationError(f"String too long (max {max_length} chars)")
    
    return sanitized

# Validate pagination parameters
def validate_pagination(offset: int = 0, limit: int = 25) -> tuple:
    if offset < 0:
        raise ValidationError("Offset cannot be negative")
    if limit <= 0 or limit > 100:
        raise ValidationError("Limit must be between 1 and 100")
    return offset, limit

# Advanced rate limiting with Redis (optional enhancement)
import redis
from datetime import datetime, timedelta

class AdvancedRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def check_rate_limit(self, client_id: str, limit: int = 60, window: int = 60):
        key = f"rate_limit:{client_id}"
        current_time = datetime.utcnow()
        
        # Sliding window rate limiting
        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(key, 0, current_time.timestamp() - window)
        pipe.zcard(key)
        pipe.zadd(key, {str(current_time.timestamp()): current_time.timestamp()})
        pipe.expire(key, window)
        
        results = pipe.execute()
        request_count = results[1]
        
        if request_count >= limit:
            raise RateLimitError(f"Rate limit exceeded: {request_count}/{limit} requests")