alpaca_mcp_server.py•11.5 kB
#!/usr/bin/env python3
"""
Alpaca MCP Server
A Model Context Protocol server for Alpaca Trading API
This server provides tools for interacting with Alpaca's paper trading and live trading APIs.
Tested and verified to work on Apple Silicon M1 Pro laptops.
"""
import json
import os
import sys
import logging
logging.basicConfig(level=logging.INFO, stream=sys.stderr)
try:
import alpaca_trade_api as tradeapi
except ImportError:
print(json.dumps({"error": "alpaca-trade-api not installed. Run: pip install alpaca-trade-api"}), file=sys.stderr)
sys.exit(1)
def get_api():
"""Get Alpaca API client"""
api_key = os.getenv('APCA_API_KEY_ID')
api_secret = os.getenv('APCA_API_SECRET_KEY')
base_url = os.getenv('APCA_API_BASE_URL', 'https://paper-api.alpaca.markets')
if not api_key or not api_secret:
raise ValueError("Missing API credentials. Set APCA_API_KEY_ID and APCA_API_SECRET_KEY environment variables.")
return tradeapi.REST(api_key, api_secret, base_url, api_version='v2')
def handle_request(request):
"""Handle MCP request"""
try:
if request.get("method") == "initialize":
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"protocolVersion": "2024-11-05",
"serverInfo": {"name": "alpaca-mcp-server", "version": "1.0.0"},
"capabilities": {"tools": {"listChanged": False}}
}
}
elif request.get("method") == "tools/list":
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"tools": [
{
"name": "get_account",
"description": "Get Alpaca account information including buying power, cash, equity, and account status",
"inputSchema": {"type": "object", "properties": {}, "required": []}
},
{
"name": "get_positions",
"description": "Get current stock positions in the account",
"inputSchema": {"type": "object", "properties": {}, "required": []}
},
{
"name": "get_quote",
"description": "Get latest quote for a stock symbol",
"inputSchema": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, TSLA, NVDA)"
}
},
"required": ["symbol"]
}
},
{
"name": "get_orders",
"description": "Get recent orders from the account",
"inputSchema": {
"type": "object",
"properties": {
"status": {
"type": "string",
"description": "Order status filter (open, closed, all)",
"default": "all"
},
"limit": {
"type": "integer",
"description": "Maximum number of orders to return",
"default": 10
}
},
"required": []
}
}
]
}
}
elif request.get("method") == "tools/call":
api = get_api()
tool_name = request.get("params", {}).get("name")
arguments = request.get("params", {}).get("arguments", {})
if tool_name == "get_account":
account = api.get_account()
result = {
"account_number": account.account_number,
"status": account.status,
"buying_power": float(account.buying_power),
"cash": float(account.cash),
"equity": float(account.equity),
"portfolio_value": float(account.portfolio_value),
"pattern_day_trader": account.pattern_day_trader,
"day_trade_count": int(account.daytrade_count),
"sma": float(account.sma) if account.sma else 0.0
}
content = [{"type": "text", "text": f"Alpaca Account Information:\n{json.dumps(result, indent=2)}"}]
elif tool_name == "get_positions":
positions = api.list_positions()
if not positions:
content = [{"type": "text", "text": "No open positions found."}]
else:
pos_data = []
total_value = 0
for pos in positions:
pos_info = {
"symbol": pos.symbol,
"qty": float(pos.qty),
"side": pos.side,
"market_value": float(pos.market_value),
"cost_basis": float(pos.cost_basis),
"unrealized_pl": float(pos.unrealized_pl),
"unrealized_plpc": float(pos.unrealized_plpc),
"current_price": float(pos.current_price)
}
pos_data.append(pos_info)
total_value += float(pos.market_value)
summary = {
"total_positions": len(pos_data),
"total_market_value": total_value,
"positions": pos_data
}
content = [{"type": "text", "text": f"Current Positions:\n{json.dumps(summary, indent=2)}"}]
elif tool_name == "get_quote":
symbol = arguments.get("symbol", "").upper()
if not symbol:
content = [{"type": "text", "text": "Error: symbol parameter is required"}]
else:
try:
quote = api.get_latest_quote(symbol)
quote_data = {
"symbol": symbol,
"bid_price": float(quote.bid_price),
"ask_price": float(quote.ask_price),
"bid_size": int(quote.bid_size),
"ask_size": int(quote.ask_size),
"timestamp": str(quote.timestamp),
"spread": round(float(quote.ask_price) - float(quote.bid_price), 4)
}
content = [{"type": "text", "text": f"Quote for {symbol}:\n{json.dumps(quote_data, indent=2)}"}]
except Exception as e:
content = [{"type": "text", "text": f"Error getting quote for {symbol}: {str(e)}"}]
elif tool_name == "get_orders":
status = arguments.get("status", "all")
limit = arguments.get("limit", 10)
try:
if status == "open":
orders = api.list_orders(status='open', limit=limit)
elif status == "closed":
orders = api.list_orders(status='closed', limit=limit)
else:
orders = api.list_orders(status='all', limit=limit)
if not orders:
content = [{"type": "text", "text": f"No {status} orders found."}]
else:
order_data = []
for order in orders:
order_info = {
"id": order.id,
"symbol": order.symbol,
"qty": float(order.qty),
"side": order.side,
"order_type": order.order_type,
"status": order.status,
"submitted_at": str(order.submitted_at),
"filled_qty": float(order.filled_qty) if order.filled_qty else 0,
"filled_avg_price": float(order.filled_avg_price) if order.filled_avg_price else None
}
order_data.append(order_info)
summary = {
"total_orders": len(order_data),
"status_filter": status,
"orders": order_data
}
content = [{"type": "text", "text": f"Orders ({status}):\n{json.dumps(summary, indent=2)}"}]
except Exception as e:
content = [{"type": "text", "text": f"Error getting orders: {str(e)}"}]
else:
content = [{"type": "text", "text": f"Unknown tool: {tool_name}"}]
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {"content": content}
}
else:
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {}
}
except Exception as e:
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {"code": -32603, "message": str(e)}
}
def main():
"""Main loop - simple stdio JSON-RPC for MCP protocol"""
try:
# Test Alpaca connection on startup
api = get_api()
account = api.get_account()
print(f"Connected to Alpaca. Account: {account.account_number}, Status: {account.status}", file=sys.stderr)
# Handle MCP requests via stdin/stdout
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
request = json.loads(line)
response = handle_request(request)
print(json.dumps(response))
sys.stdout.flush()
except json.JSONDecodeError:
continue
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error processing request: {e}", file=sys.stderr)
except Exception as e:
print(f"Fatal error: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()