TA-Lib MCP Server

Instructions

Implementation Reference

• src/mcp_talib/core/server.py:58-100 (handler)

  Primary execution handler for the calculate_ema tool. Registered via @mcp.tool() decorator. Delegates computation to EMAIndicator from registry.

  @mcp.tool()
  async def calculate_ema(
      close: List[float],
      timeperiod: int = 20
  ) -> Dict[str, Any]:
      """Calculate Exponential Moving Average (EMA).
      
      Args:
          close: List of closing prices
          timeperiod: Number of periods for average (default: 20)
          
      Returns:
          Dictionary with EMA values and metadata
      """
      try:
          # Get indicator from registry
          indicator = registry.get_indicator("ema")
          if not indicator:
              raise ValueError("EMA 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/ema.py:9-82 (helper)

  EMAIndicator class containing the core EMA calculation logic, input schema, and metadata handling used by the tool handler.

  class EMAIndicator(BaseIndicator):
      """Exponential Moving Average (EMA) indicator implementation."""
      
      def __init__(self):
          """Initialize EMA indicator."""
          super().__init__(
              name="ema",
              description="Exponential Moving Average (EMA) - gives more weight to recent prices"
          )
      
      @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 for EMA calculation"
                  }
              },
              "required": ["close_prices"]
          }
      
      async def calculate(
          self, 
          market_data: MarketData, 
          options: Dict[str, Any] = None
      ) -> IndicatorResult:
          """Calculate EMA 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 EMA
          # Multiplier: (2 / (timeperiod + 1))
          multiplier = 2.0 / (timeperiod + 1)
          
          # Initialize EMA with SMA of first timeperiod values
          ema_values = []
          initial_sma = sum(close_prices[:timeperiod]) / timeperiod
          ema_values.append(initial_sma)
          
          # Calculate EMA for remaining values
          for i in range(timeperiod, len(close_prices)):
              ema = (close_prices[i] - ema_values[-1]) * multiplier + ema_values[-1]
              ema_values.append(ema)
          
          return IndicatorResult(
              indicator_name=self.name,
              success=True,
              values={"ema": ema_values},
              metadata={
                  "timeperiod": timeperiod,
                  "multiplier": multiplier,
                  "input_points": len(close_prices),
                  "output_points": len(ema_values)
              }
          )

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

  Registration of the EMAIndicator in the global indicator registry, enabling registry.get_indicator('ema') in the handler.

  registry.register("ema", EMAIndicator)

• src/mcp_talib/indicators/ema.py:19-36 (schema)

  JSON schema definition for inputs to the EMA calculation.

  @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 for EMA calculation"
              }
          },
          "required": ["close_prices"]
      }