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"]
      }