TA-Lib MCP Server

Instructions

Implementation Reference

• src/mcp_talib/core/server.py:14-56 (handler)

  MCP tool handler for 'calculate_sma': decorated with @mcp.tool(), delegates to SMAIndicator via registry.
  @mcp.tool() async def calculate_sma( close: List[float], timeperiod: int = 20 ) -> Dict[str, Any]: """Calculate Simple Moving Average (SMA). Args: close: List of closing prices timeperiod: Number of periods to average (default: 20) Returns: Dictionary with SMA values and metadata """ try: # Get indicator from registry indicator = registry.get_indicator("sma") if not indicator: raise ValueError("SMA 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/sma.py:9-81 (helper)

  SMAIndicator class: core computation logic for SMA, including input validation, schema definition, and pure Python SMA calculation.
  class SMAIndicator(BaseIndicator): """Simple Moving Average (SMA) indicator implementation.""" def __init__(self): """Initialize SMA indicator.""" super().__init__( name="sma", description="Simple Moving Average (SMA) - calculates the arithmetic mean of prices over a specified time period" ) @property def name(self) -> str: return "sma" @property def description(self) -> str: return "Simple Moving Average (SMA) - calculates the average price over a specified period" @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 to average" } }, "required": ["close_prices"] } async def calculate( self, market_data: MarketData, options: Dict[str, Any] = None ) -> IndicatorResult: """Calculate SMA 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 SMA sma_values = [] for i in range(timeperiod - 1, len(close_prices)): avg = sum(close_prices[i - timeperiod + 1:i + 1]) / timeperiod sma_values.append(avg) return IndicatorResult( indicator_name=self.name, success=True, values={"sma": sma_values}, metadata={ "timeperiod": timeperiod, "input_points": len(close_prices), "output_points": len(sma_values) } )
• src/mcp_talib/indicators/__init__.py:26-26 (registration)

  Registration of SMAIndicator class in the global indicator registry.
  registry.register("sma", SMAIndicator)
• src/mcp_talib/indicators/sma.py:28-44 (schema)

  JSON schema definition for SMA indicator inputs (used indirectly by the tool).
  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 to average" } }, "required": ["close_prices"] }