Taiwan Stock Agent

Instructions

Implementation Reference

• mcp_server.py:104-131 (handler)

  MCP tool handler registered with @mcp.tool(name="get_price_history"). Calls core get_price_history helper, handles exceptions, formats response with PriceHistoryResponse Pydantic model.

  @mcp.tool(name="get_price_history",
            description="Get historical price data for a specific stock.",
  )
  async def get_price_history_tool(
      stock_code: str, period: str = "1mo"
  ) -> PriceHistoryResponse:
      """Get historical price data for a specific stock.""" 
      try:
          raw_data = await get_price_history(stock_code, period)
          # Extract clean data without _metadata for Pydantic model
          from tw_stock_agent.utils.mcp_error_handler import MCPResponseFormatter
          clean_data = MCPResponseFormatter.extract_metadata_for_model(raw_data)
          return PriceHistoryResponse(**clean_data)
      except TwStockAgentError as e:
          # Return error response in proper format
          return PriceHistoryResponse(
              stock_code=stock_code,
              period=period,
              data=[],
              error=e.message
          )
      except Exception as e:
          return PriceHistoryResponse(
              stock_code=stock_code,
              period=period,
              data=[],
              error=f"Unexpected error: {str(e)}"
          )

• tw_stock_agent/tools/stock_tools.py:112-172 (helper)

  Core implementation of get_price_history: validates parameters, maps period to days, fetches historical price data from StockService, constructs and formats the response dictionary.

  @mcp_error_handler("get_price_history")
  async def get_price_history(stock_code: str, period: str = "1mo") -> dict[str, Any]:
      """
      取得股票的歷史價格資料
      
      Args:
          stock_code: 股票代號，例如：2330
          period: 時間區間，可選值：'1d', '5d', '1mo', '3mo', '6mo', '1y', 'ytd', 'max'
          
      Returns:
          歷史價格資料
          
      Raises:
          InvalidStockCodeError: 股票代號格式錯誤
          ParameterValidationError: period參數錯誤
          StockDataUnavailableError: 價格資料無法取得
      """
      # Validate request parameters
      validated_params = validate_stock_request(
          stock_code=stock_code,
          period=period
      )
      
      # 處理period參數，轉換為天數
      days_map = {
          "1d": 1,
          "5d": 5,
          "1mo": 30,
          "3mo": 90,
          "6mo": 180,
          "1y": 365,
          "ytd": max(1, (datetime.now() - datetime(datetime.now().year, 1, 1)).days),
          "max": 3650  # 約10年
      }
      
      validated_period = validated_params["period"]
      days = days_map.get(validated_period, 30)
      
      if days <= 0:
          raise ParameterValidationError(
              parameter_name="period",
              parameter_value=period,
              message="Invalid period resulting in zero or negative days"
          )
      
      # Fetch price data
      price_data = await stock_service.fetch_price_data(
          validated_params["stock_code"], 
          days
      )
      
      result = {
          "stock_code": validated_params["stock_code"],
          "period": validated_period,
          "data": price_data,
          "requested_days": days,
          "actual_records": len(price_data) if isinstance(price_data, list) else 0
      }
      
      # Format response for MCP with enhanced structure  
      return MCPResponseFormatter.format_price_data_response(result)

• tw_stock_agent/models/stock_models.py:417-517 (schema)

  Pydantic model PriceHistoryResponse defining the structure, validation, and computed fields for the tool's output response.

  class PriceHistoryResponse(BaseStockResponse):
      """Enhanced response model for price history data."""
      
      period: str = Field(
          ...,
          description="Time period (時間區間)",
          example="1mo"
      )
      data: List[PriceDataPoint] = Field(
          default_factory=list,
          description="Price data points (價格資料)"
      )
      requested_days: Optional[int] = Field(
          None,
          description="Number of days requested",
          alias="requestedDays"
      )
      actual_records: Optional[int] = Field(
          None,
          description="Actual number of records returned",
          alias="actualRecords"
      )
      
      def __init__(self, **data):
          super().__init__(**data)
          self.metadata.data_type = "price_history"
          self.metadata.record_count = len(self.data)
      
      @field_validator("period")
      @classmethod
      def validate_period(cls, v: str) -> str:
          """Validate period format."""
          valid_periods = {'1d', '5d', '1mo', '3mo', '6mo', '1y', 'ytd', 'max'}
          if v not in valid_periods:
              raise ValueError(f"Invalid period. Must be one of: {', '.join(valid_periods)}")
          return v
      
      @computed_field
      @property
      def price_summary(self) -> Optional[Dict[str, Any]]:
          """Calculate price summary statistics."""
          if not self.data:
              return None
          
          prices = []
          volumes = []
          
          for point in self.data:
              if point.close_price:
                  prices.append(point.close_price.amount)
              if point.volume:
                  volumes.append(point.volume)
          
          if not prices:
              return None
          
          return {
              "min_price": float(min(prices)),
              "max_price": float(max(prices)),
              "avg_price": float(sum(prices) / len(prices)),
              "total_volume": sum(volumes) if volumes else None,
              "avg_volume": sum(volumes) / len(volumes) if volumes else None,
              "trading_days": len(self.data)
          }
      
      @computed_field
      @property
      def overall_change(self) -> Optional[Dict[str, Any]]:
          """Calculate overall change for the period."""
          if len(self.data) < 2:
              return None
          
          first_price = None
          last_price = None
          
          # Find first valid price
          for point in self.data:
              if point.close_price:
                  first_price = point.close_price.amount
                  break
          
          # Find last valid price
          for point in reversed(self.data):
              if point.close_price:
                  last_price = point.close_price.amount
                  break
          
          if not first_price or not last_price or first_price == 0:
              return None
          
          change_amount = last_price - first_price
          change_percentage = float((change_amount / first_price) * 100)
          
          return {
              "period_start_price": float(first_price),
              "period_end_price": float(last_price),
              "change_amount": float(change_amount),
              "change_percentage": change_percentage,
              "is_positive": change_amount > 0
          }