get_stock_price

Instructions

Implementation Reference

• src/brvm_mcp/server.py:103-129 (handler)

  The main handler for the get_stock_price MCP tool. Decorated with @mcp.tool(), it accepts a ticker string, normalizes it to uppercase, checks the cache, and if not cached, calls scraper.get_quote() to fetch the stock data from BRVM. Returns JSON with price, change, volume, and company details.

  @mcp.tool()
  async def get_stock_price(ticker: str) -> str:
      """
      Get the current price and trading data for a specific BRVM-listed stock.
  
      Args:
          ticker: The BRVM ticker symbol (e.g., SNTS for Sonatel, SGBC for Société Générale CI,
                  ETIT for Ecobank Transnational, ORAC for Orange CI, ONTBF for Onatel).
  
      Returns a JSON object with price, change, volume, and company details.
      """
      assert scraper and cache
      ticker = ticker.upper().strip()
  
      cached = cache.get(f"quote:{ticker}")
      if cached:
          return json.dumps(cached, ensure_ascii=False, indent=2)
  
      quote = await scraper.get_quote(ticker)
      if not quote:
          return json.dumps(
              {"error": f"Ticker '{ticker}' not found on BRVM. Use search_stocks to find valid tickers."}
          )
  
      data = quote.model_dump()
      cache.set(f"quote:{ticker}", data)
      return json.dumps(data, ensure_ascii=False, indent=2)

• src/brvm_mcp/server.py:103-103 (registration)

  The tool is registered via the @mcp.tool() decorator on line 103, which registers it with the FastMCP server instance created on line 69.

  @mcp.tool()

• src/brvm_mcp/scrapers/__init__.py:178-182 (helper)

  The scraper.get_quote() helper function called by the handler. It calls get_all_quotes() to scrape all live quotes from afx.kwayisi.org/brvm/, then filters for the matching ticker.

  async def get_quote(self, ticker: str) -> StockQuote | None:
      """Get a single stock quote by ticker."""
      ticker = ticker.upper().strip()
      quotes = await self.get_all_quotes()
      return next((q for q in quotes if q.ticker == ticker), None)

• src/brvm_mcp/models/__init__.py:30-45 (schema)

  The StockQuote Pydantic model defines the schema for the data returned by get_stock_price, including ticker, name, price, change, volume, previous_close, country, sector, etc.

  class StockQuote(BaseModel):
      """Real-time or end-of-day quote for a single ticker."""
  
      ticker: str = Field(..., description="BRVM ticker symbol, e.g. SNTS, ONTBF")
      name: str = Field(..., description="Full company name")
      price: float = Field(..., description="Last traded price in XOF")
      change: float = Field(0.0, description="Absolute price change vs previous close")
      change_pct: float = Field(0.0, description="Percentage change vs previous close")
      volume: int = Field(0, description="Number of shares traded")
      value_traded: float = Field(0.0, description="Total value traded in XOF")
      previous_close: float = Field(0.0, description="Previous session closing price")
      market_cap: float | None = Field(None, description="Market capitalization in XOF")
      country: str = Field("", description="Country of the listed company")
      sector: str = Field("", description="Industry sector")
      as_of: str = Field("", description="Data timestamp (ISO format or date)")

• src/brvm_mcp/cache.py:22-52 (helper)

  The BRVMCache class used by get_stock_price to cache/fetch quote data with a 5-minute TTL, reducing repeated requests to the BRVM data source.

  class BRVMCache:
      """Simple disk-backed cache with TTL."""
  
      def __init__(self, cache_dir: Path = DEFAULT_CACHE_DIR, ttl: int = DEFAULT_TTL):
          self.ttl = ttl
          self._cache = diskcache.Cache(str(cache_dir))
          logger.info(f"Cache initialized at {cache_dir} with TTL={ttl}s")
  
      def get(self, key: str) -> Any | None:
          """Get a value from cache, returns None if expired or missing."""
          value = self._cache.get(key)
          if value is not None:
              logger.debug(f"Cache HIT: {key}")
              return json.loads(value) if isinstance(value, str) else value
          logger.debug(f"Cache MISS: {key}")
          return None
  
      def set(self, key: str, value: Any) -> None:
          """Set a value in cache with TTL."""
          serialized = json.dumps(value, default=str)
          self._cache.set(key, serialized, expire=self.ttl)
          logger.debug(f"Cache SET: {key} (TTL={self.ttl}s)")
  
      def clear(self) -> None:
          """Clear all cached data."""
          self._cache.clear()
          logger.info("Cache cleared")
  
      def close(self) -> None:
          self._cache.close()