TA-Lib MCP Server

Instructions

Implementation Reference

• src/mcp_talib/core/server.py:102-144 (handler)

  The MCP tool handler function for 'calculate_rsi'. It retrieves the RSI indicator from the registry, creates MarketData from input close prices, calls the indicator's calculate method, and returns the result or error.
  @mcp.tool() async def calculate_rsi( close: List[float], timeperiod: int = 14 ) -> Dict[str, Any]: """Calculate Relative Strength Index (RSI). Args: close: List of closing prices timeperiod: Number of periods for RSI (default: 14) Returns: Dictionary with RSI values and metadata """ try: # Get indicator from registry indicator = registry.get_indicator("rsi") if not indicator: raise ValueError("RSI 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/rsi.py:38-95 (helper)

  Core implementation of RSI calculation in RSIIndicator.calculate(). Computes price deltas, separates gains/losses, applies exponential moving average smoothing for avg gain/loss, computes RS and RSI values.
  async def calculate( self, market_data: MarketData, options: Dict[str, Any] = None ) -> IndicatorResult: """Calculate RSI indicator.""" if options is None: options = {} timeperiod = options.get("timeperiod", 14) close_prices = market_data.close if len(close_prices) < timeperiod + 1: return IndicatorResult( indicator_name=self.name, success=False, values={}, error_message=f"Not enough data points. Need at least {timeperiod + 1}, got {len(close_prices)}" ) # Calculate price changes deltas = [close_prices[i] - close_prices[i-1] for i in range(1, len(close_prices))] # Separate gains and losses gains = [d if d > 0 else 0 for d in deltas] losses = [-d if d < 0 else 0 for d in deltas] # Calculate initial average gain and loss avg_gain = sum(gains[:timeperiod]) / timeperiod avg_loss = sum(losses[:timeperiod]) / timeperiod rsi_values = [] # Calculate subsequent values using exponential smoothing for i in range(timeperiod, len(gains)): avg_gain = (avg_gain * (timeperiod - 1) + gains[i]) / timeperiod avg_loss = (avg_loss * (timeperiod - 1) + losses[i]) / timeperiod # Calculate RS (Relative Strength) if avg_loss == 0: rs = float('inf') else: rs = avg_gain / avg_loss # Calculate RSI: 100 - (100 / (1 + RS)) rsi = 100 - (100 / (1 + rs)) rsi_values.append(min(max(rsi, 0), 100)) # Clamp between 0 and 100 return IndicatorResult( indicator_name=self.name, success=True, values={"rsi": rsi_values}, metadata={ "timeperiod": timeperiod, "input_points": len(close_prices), "output_points": len(rsi_values) } )
• src/mcp_talib/indicators/__init__.py:28-28 (registration)

  Registration of the RSIIndicator class in the global indicator registry.
  registry.register("rsi", RSIIndicator)
• src/mcp_talib/indicators/rsi.py:20-36 (schema)

  Input schema definition for the RSI indicator, defining expected parameters and types.
  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": 14, "description": "Number of periods for RSI calculation" } }, "required": ["close_prices"] }