TA-Lib MCP Server

Instructions

Implementation Reference

• src/mcp_talib/core/server.py:293-306 (handler)

  MCP tool handler for 'calculate_sarext'. Delegates to the 'sarext' indicator from the registry, passing high/low prices and parameters, and returns the result.

  @mcp.tool()
  async def calculate_sarext(high: List[float], low: List[float], startvalue: Optional[float] = None, offsetonreverse: float = 0.0, acceleration_initlong: float = 0.02, acceleration_long: float = 0.02, acceleration_maxlong: float = 0.2, acceleration_initshort: float = 0.02, acceleration_short: float = 0.02, acceleration_maxshort: float = 0.2) -> Dict[str, Any]:
      try:
          indicator = registry.get_indicator("sarext")
          if not indicator:
              raise ValueError("SAREXT indicator not found")
          market_data = MarketData(close=[0], high=high, low=low)
          opts = {"startvalue": startvalue, "offsetonreverse": offsetonreverse, "acceleration_initlong": acceleration_initlong, "acceleration_long": acceleration_long, "acceleration_maxlong": acceleration_maxlong, "acceleration_initshort": acceleration_initshort, "acceleration_short": acceleration_short, "acceleration_maxshort": acceleration_maxshort}
          result = await indicator.calculate(market_data, opts)
          if result.success:
              return {"success": True, "values": result.values, "metadata": result.metadata}
          return {"success": False, "error": result.error_message}
      except Exception as e:
          return {"success": False, "error": str(e)}

• src/mcp_talib/indicators/sarext.py:16-33 (schema)

  Input schema definition for the SAREXT indicator, used by the tool handler. Defines parameters including high_prices, low_prices, and various acceleration settings.

  @property
  def input_schema(self) -> Dict[str, Any]:
      return {
          "type": "object",
          "properties": {
              "high_prices": {"type": "array", "items": {"type": "number"}},
              "low_prices": {"type": "array", "items": {"type": "number"}},
              "startvalue": {"type": ["number", "null"]},
              "offsetonreverse": {"type": "number", "default": 0.0},
              "acceleration_initlong": {"type": "number", "default": 0.02},
              "acceleration_long": {"type": "number", "default": 0.02},
              "acceleration_maxlong": {"type": "number", "default": 0.2},
              "acceleration_initshort": {"type": "number", "default": 0.02},
              "acceleration_short": {"type": "number", "default": 0.02},
              "acceleration_maxshort": {"type": "number", "default": 0.2},
          },
          "required": ["high_prices", "low_prices"],
      }

• src/mcp_talib/indicators/sarext.py:35-62 (helper)

  Core calculation logic for SAREXT indicator using TA-Lib's SAREXT function. Maps parameters and computes the Parabolic SAR - Extended values.

  async def calculate(self, market_data: MarketData, options: Dict[str, Any] = None) -> IndicatorResult:
      if options is None:
          options = {}
      high = np.asarray(market_data.high or [], dtype=float)
      low = np.asarray(market_data.low or [], dtype=float)
  
      # Map our input option keys (snake_case) to TA-Lib SAREXT parameter names
      key_map = {
          "startvalue": "startvalue",
          "offsetonreverse": "offsetonreverse",
          "acceleration_initlong": "accelerationinitlong",
          "acceleration_long": "accelerationlong",
          "acceleration_maxlong": "accelerationmaxlong",
          "acceleration_initshort": "accelerationinitshort",
          "acceleration_short": "accelerationshort",
          "acceleration_maxshort": "accelerationmaxshort",
      }
  
      params = {}
      for src_key, ta_key in key_map.items():
          if src_key in options and options.get(src_key) is not None:
              params[ta_key] = options.get(src_key)
  
      try:
          out = ta.SAREXT(high, low, **params)
          return IndicatorResult(indicator_name=self.name, success=True, values={"sarext": out.tolist()}, metadata={"input_points": len(high), "output_points": len(out)})
      except Exception as e:
          return IndicatorResult(indicator_name=self.name, success=False, values={}, error_message=str(e))

• src/mcp_talib/indicators/__init__.py:39-39 (registration)

  Registration of the SAREXTIndicator class in the global indicators registry, enabling registry.get_indicator('sarext') in the tool handler.

  registry.register("sarext", SAREXTIndicator)