TA-Lib MCP Server

Instructions

Implementation Reference

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

  The main handler function for the 'calculate_ema' MCP tool, decorated with @mcp.tool() for automatic registration. It retrieves the EMAIndicator from the registry, creates MarketData from input close prices, calls the indicator's calculate method with the timeperiod parameter, and returns the result or error in a standardized dictionary format.
  @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)

  The EMAIndicator class provides the core implementation of the Exponential Moving Average calculation using a custom algorithm (initial SMA followed by iterative EMA formula). Includes input schema definition and data validation. This is the component called 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)

  Registers the EMAIndicator class in the IndicatorRegistry instance, allowing tool handlers to retrieve it using registry.get_indicator("ema").
  registry.register("ema", EMAIndicator)
• src/mcp_talib/indicators/ema.py:19-36 (schema)

  JSON Schema definition for the EMA indicator inputs, used internally for validation (close_prices array and optional timeperiod). Note: Tool uses type hints instead.
  @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"] }