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