TA-Lib MCP Server

Instructions

Implementation Reference

• src/mcp_talib/core/server.py:58-100 (handler)

  Primary execution handler for the calculate_ema tool. Registered via @mcp.tool() decorator. Delegates computation to EMAIndicator from registry.
  @mcp.tool() async def calculate_ema( close: List[float], timeperiod: int = 20 ) -> Dict[str, Any]: """Calculate Exponential Moving Average (EMA). Args: close: List of closing prices timeperiod: Number of periods for average (default: 20) Returns: Dictionary with EMA values and metadata """ try: # Get indicator from registry indicator = registry.get_indicator("ema") if not indicator: raise ValueError("EMA indicator not found") # Create market data market_data = MarketData(close=close) # Calculate indicator result = await indicator.calculate(market_data, {"timeperiod": timeperiod}) if result.success: return { "success": True, "values": result.values, "metadata": result.metadata, } else: return { "success": False, "error": result.error_message, } except Exception as e: return { "success": False, "error": str(e), }
• src/mcp_talib/indicators/ema.py:9-82 (helper)

  EMAIndicator class containing the core EMA calculation logic, input schema, and metadata handling used by the tool handler.
  class EMAIndicator(BaseIndicator): """Exponential Moving Average (EMA) indicator implementation.""" def __init__(self): """Initialize EMA indicator.""" super().__init__( name="ema", description="Exponential Moving Average (EMA) - gives more weight to recent prices" ) @property def input_schema(self) -> Dict[str, Any]: return { "type": "object", "properties": { "close_prices": { "type": "array", "items": {"type": "number"}, "description": "List of closing prices" }, "timeperiod": { "type": "integer", "default": 20, "description": "Number of periods for EMA calculation" } }, "required": ["close_prices"] } async def calculate( self, market_data: MarketData, options: Dict[str, Any] = None ) -> IndicatorResult: """Calculate EMA indicator.""" if options is None: options = {} timeperiod = options.get("timeperiod", 20) close_prices = market_data.close if len(close_prices) < timeperiod: return IndicatorResult( indicator_name=self.name, success=False, values={}, error_message=f"Not enough data points. Need at least {timeperiod}, got {len(close_prices)}" ) # Calculate EMA # Multiplier: (2 / (timeperiod + 1)) multiplier = 2.0 / (timeperiod + 1) # Initialize EMA with SMA of first timeperiod values ema_values = [] initial_sma = sum(close_prices[:timeperiod]) / timeperiod ema_values.append(initial_sma) # Calculate EMA for remaining values for i in range(timeperiod, len(close_prices)): ema = (close_prices[i] - ema_values[-1]) * multiplier + ema_values[-1] ema_values.append(ema) return IndicatorResult( indicator_name=self.name, success=True, values={"ema": ema_values}, metadata={ "timeperiod": timeperiod, "multiplier": multiplier, "input_points": len(close_prices), "output_points": len(ema_values) } )
• src/mcp_talib/indicators/__init__.py:27-27 (registration)

  Registration of the EMAIndicator in the global indicator registry, enabling registry.get_indicator('ema') in the handler.
  registry.register("ema", EMAIndicator)
• src/mcp_talib/indicators/ema.py:19-36 (schema)

  JSON schema definition for inputs to the EMA calculation.
  @property def input_schema(self) -> Dict[str, Any]: return { "type": "object", "properties": { "close_prices": { "type": "array", "items": {"type": "number"}, "description": "List of closing prices" }, "timeperiod": { "type": "integer", "default": 20, "description": "Number of periods for EMA calculation" } }, "required": ["close_prices"] }