Taiwan Stock Agent

Instructions

Implementation Reference

• tw_stock_agent/tools/stock_tools.py:87-110 (handler)

  Core handler function that validates the stock code, fetches data from StockService, and formats the response for MCP.

  @mcp_error_handler("get_stock_data")
  async def get_stock_data(stock_code: str) -> dict[str, Any]:
      """
      取得股票基本資料
      
      Args:
          stock_code: 股票代號，例如：2330
          
      Returns:
          股票資料，包含公司概況、產業別、市值等資訊
          
      Raises:
          InvalidStockCodeError: 股票代號格式錯誤
          StockNotFoundError: 找不到股票
          StockDataUnavailableError: 股票資料無法取得
      """
      # Validate request parameters
      validated_params = validate_stock_request(stock_code=stock_code)
      
      # Fetch stock data using the validated stock code
      result = await stock_service.fetch_stock_data(validated_params["stock_code"])
      
      # Format response for MCP
      return MCPResponseFormatter.format_stock_data_response(result)

• mcp_server.py:77-102 (registration)

  MCP tool registration for 'get_stock_data' with input stock_code: str and output StockDataResponse. Delegates to core handler and handles errors.

  @mcp.tool(name="get_stock_data",
            description="Get detailed information about a specific stock.",
  )
  async def get_stock_data_tool(stock_code: str) -> StockDataResponse:
      """Get detailed information about a specific stock."""
      try:
          raw_data = await get_stock_data(stock_code)
          # 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 StockDataResponse(**clean_data)
      except TwStockAgentError as e:
          # For MCP tools, we need to return a valid response with error information
          return StockDataResponse(
              stock_code=stock_code,
              updated_at=e.context.timestamp.isoformat(),
              error=e.message
          )
      except Exception as e:
          # Handle unexpected errors
          return StockDataResponse(
              stock_code=stock_code,
              updated_at=datetime.now().isoformat(),
              error=f"Unexpected error: {str(e)}"
          )

• tw_stock_agent/models/stock_models.py:177-277 (schema)

  Pydantic schema for the tool's output, defining structure, validation, and computed fields for stock data responses.

  class StockDataResponse(BaseStockResponse):
      """Enhanced response model for stock basic data."""
      
      name: Optional[str] = Field(
          None,
          description="Company name (公司名稱)",
          example="台積電",
          alias="companyName"
      )
      name_en: Optional[str] = Field(
          None,
          description="English company name",
          alias="companyNameEn"
      )
      stock_type: Optional[str] = Field(
          None,
          description="Stock type (股票類型)",
          example="股票",
          alias="type"
      )
      isin_code: Optional[str] = Field(
          None,
          description="ISIN code (ISIN代碼)",
          alias="isin"
      )
      listing_date: Optional[datetime] = Field(
          None,
          description="Listing date (上市日期)",
          alias="startDate"
      )
      market_type: Optional[str] = Field(
          None,
          description="Market type (市場別)",
          example="上市",
          alias="market"
      )
      industry_category: Optional[str] = Field(
          None,
          description="Industry category (產業別)",
          example="半導體業",
          alias="industry"
      )
      market_cap: Optional[TWDAmount] = Field(
          None,
          description="Market capitalization",
          alias="marketCap"
      )
      
      def __init__(self, **data):
          # Handle backward compatibility for field names
          if 'type' in data and 'stock_type' not in data:
              data['stock_type'] = data.pop('type')
          if 'start_date' in data and 'listing_date' not in data:
              start_date = data.pop('start_date')
              if start_date:
                  try:
                      data['listing_date'] = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
                  except (ValueError, AttributeError):
                      pass
          
          super().__init__(**data)
      
      @field_validator("listing_date")
      @classmethod
      def validate_listing_date(cls, v: Optional[Union[str, datetime]]) -> Optional[datetime]:
          """Validate and convert listing date."""
          if v is None:
              return None
          
          if isinstance(v, str):
              try:
                  # Handle various date formats
                  if 'T' in v:  # ISO format
                      return datetime.fromisoformat(v.replace('Z', '+00:00')).astimezone(TAIWAN_TZ)
                  else:  # Date only
                      return datetime.strptime(v, '%Y-%m-%d').replace(tzinfo=TAIWAN_TZ)
              except ValueError as e:
                  raise ValueError(f"Invalid date format: {v}") from e
          
          return v.astimezone(TAIWAN_TZ) if v.tzinfo else v.replace(tzinfo=TAIWAN_TZ)
      
      @computed_field
      @property
      def years_listed(self) -> Optional[int]:
          """Calculate years since listing."""
          if not self.listing_date:
              return None
          
          years = (datetime.now(TAIWAN_TZ) - self.listing_date).days // 365
          return max(0, years)
      
      @computed_field
      @property
      def is_recently_listed(self) -> bool:
          """Check if stock was listed within the last year."""
          if not self.listing_date:
              return False
          
          days_since_listing = (datetime.now(TAIWAN_TZ) - self.listing_date).days
          return days_since_listing <= 365