Taiwan Stock Agent

Instructions

Implementation Reference

• tw_stock_agent/tools/stock_tools.py:225-270 (handler)

  Core handler function that implements the get_market_overview logic by fetching TSMC (2330) realtime data as market proxy and formatting the response.
  @mcp_error_handler("get_market_overview") async def get_market_overview() -> dict[str, Any]: """ 取得市場概況 Returns: 大盤指數、成交量、漲跌家數等資訊 Raises: StockDataUnavailableError: 市場資料無法取得 """ try: # 使用台積電(2330)作為市場指標 realtime_data = await stock_service.get_realtime_data("2330") # Check if we got valid data if "error" in realtime_data: from tw_stock_agent.exceptions import StockDataUnavailableError raise StockDataUnavailableError( stock_code="market", data_type="market overview", message="Unable to fetch market overview data" ) result = { "date": datetime.now().isoformat(), "taiex": realtime_data.get("current_price"), "volume": realtime_data.get("volume"), "updated_at": realtime_data.get("updated_at"), "market_status": realtime_data.get("market_status", "unknown"), "reference_stock": "2330" # TSMC as market reference } # Format response for MCP with enhanced market overview structure return MCPResponseFormatter.format_market_overview_response(result) except TwStockAgentError: # Re-raise our custom errors raise except Exception as e: from tw_stock_agent.exceptions import StockDataUnavailableError raise StockDataUnavailableError( stock_code="market", data_type="market overview", message=f"Failed to fetch market overview: {str(e)}" )
• mcp_server.py:181-202 (registration)

  MCP tool registration for get_market_overview, wraps the core handler and returns typed MarketOverviewResponse.
  @mcp.tool(name="get_market_overview", description="Get market overview information.", ) async def get_market_overview_tool() -> MarketOverviewResponse: """Get market overview information.""" try: raw_data = await get_market_overview() # 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 MarketOverviewResponse(**clean_data) except TwStockAgentError as e: return MarketOverviewResponse( date=e.context.timestamp.isoformat(), error=e.message ) except Exception as e: from datetime import datetime return MarketOverviewResponse( date=datetime.now().isoformat(), error=f"Unexpected error: {str(e)}" )
• tw_stock_agent/models/stock_models.py:797-962 (schema)

  Pydantic model defining the input/output schema for MarketOverviewResponse used by the tool.
  class MarketOverviewResponse(BaseModel): """Enhanced response model for market overview.""" model_config = ConfigDict( str_strip_whitespace=True, validate_assignment=True, populate_by_name=True, json_encoders={ datetime: lambda v: v.isoformat(), Decimal: lambda v: float(v) } ) trading_date: datetime = Field( ..., description="Trading date (交易日期)", alias="date" ) taiex_index: Optional[MarketIndexData] = Field( None, description="TAIEX index data (台股加權指數)", alias="taiex" ) total_volume: Optional[int] = Field( None, description="Total market volume (總成交量)", alias="volume", ge=0 ) total_turnover: Optional[TWDAmount] = Field( None, description="Total market turnover (總成交金額)", alias="turnover" ) advancing_stocks: Optional[int] = Field( None, description="Number of advancing stocks (上漲家數)", ge=0 ) declining_stocks: Optional[int] = Field( None, description="Number of declining stocks (下跌家數)", ge=0 ) unchanged_stocks: Optional[int] = Field( None, description="Number of unchanged stocks (平盤家數)", ge=0 ) updated_at: datetime = Field( default_factory=lambda: datetime.now(TAIWAN_TZ), description="Last update timestamp (更新時間)", alias="updatedAt" ) market_status: Optional[str] = Field( None, description="Current market status (市場狀態)" ) reference_stock: Optional[str] = Field( None, description="Reference stock for market data" ) error: Optional[str] = Field( None, description="Error message if any (錯誤訊息)" ) metadata: ResponseMetadata = Field( default_factory=lambda: ResponseMetadata(data_type="market_overview") ) def __init__(self, **data): # Handle backward compatibility if 'date' in data and isinstance(data['date'], str): try: data['trading_date'] = datetime.fromisoformat(data['date'].replace('Z', '+00:00')) data.pop('date', None) except ValueError: try: data['trading_date'] = datetime.strptime(data['date'], '%Y-%m-%d') data.pop('date', None) except ValueError: pass # Handle TAIEX value conversion if 'taiex' in data and not isinstance(data['taiex'], dict): if data['taiex'] is not None: data['taiex_index'] = { 'index_name': 'TAIEX', 'current_value': data['taiex'] } data.pop('taiex', None) super().__init__(**data) @field_validator("trading_date") @classmethod def validate_trading_date(cls, v: Union[str, datetime]) -> datetime: """Validate and convert trading date.""" if isinstance(v, str): try: if 'T' in v: return datetime.fromisoformat(v.replace('Z', '+00:00')).astimezone(TAIWAN_TZ) else: 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) @field_validator("updated_at") @classmethod def validate_updated_at(cls, v: datetime) -> datetime: """Ensure updated_at is in Taiwan timezone.""" if v.tzinfo is None: return v.replace(tzinfo=TAIWAN_TZ) return v.astimezone(TAIWAN_TZ) @model_validator(mode='after') def update_metadata(self) -> 'MarketOverviewResponse': """Update metadata based on response state.""" if self.error: self.metadata.has_error = True return self @computed_field @property def total_trading_stocks(self) -> Optional[int]: """Calculate total number of stocks that traded.""" counts = [self.advancing_stocks, self.declining_stocks, self.unchanged_stocks] valid_counts = [c for c in counts if c is not None] if not valid_counts: return None return sum(valid_counts) @computed_field @property def market_sentiment(self) -> Optional[str]: """Determine market sentiment based on advancing/declining stocks.""" if not self.advancing_stocks or not self.declining_stocks: return None if self.advancing_stocks > self.declining_stocks * 1.5: return "very_bullish" elif self.advancing_stocks > self.declining_stocks: return "bullish" elif self.declining_stocks > self.advancing_stocks * 1.5: return "very_bearish" elif self.declining_stocks > self.advancing_stocks: return "bearish" else: return "neutral" @computed_field @property def is_trading_day(self) -> bool: """Check if the date is a trading day (weekday).""" return self.trading_date.weekday() < 5 # Monday = 0, Friday = 4 @computed_field @property def formatted_timestamp(self) -> str: """ISO formatted timestamp string.""" return self.updated_at.isoformat()
• mcp_server.py:236-240 (registration)

  MCP resource registration for market://overview which indirectly uses get_market_overview via resource_manager.
  @mcp.resource("market://overview") async def get_market_overview_resource() -> str: """Get Taiwan stock market overview.""" return await resource_manager.get_resource("market://overview")
• tw_stock_agent/services/mcp_resource_service.py:208-214 (helper)

  Helper function in resource manager that calls get_market_overview for market://overview resources.
  elif parsed.scheme == "market": # For market:// URIs, the format is market://resource_type resource_type = parsed.netloc if resource_type == "overview": data = await get_market_overview() else: