import logging import os import time import threading from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field, ValidationError from starlette.middleware.authentication import AuthenticationMiddleware import httpx from dotenv import load_dotenv from fastmcp import FastMCP from fastmcp.prompts.prompt import PromptMessage, TextContent from starlette.requests import Request from starlette.responses import PlainTextResponse from fastmcp.server.middleware.rate_limiting import ( RateLimitingMiddleware, ) from fastmcp.server.middleware.error_handling import ( ErrorHandlingMiddleware, RetryMiddleware ) load_dotenv() # Import validation utilities from app.utils.validator import validate_stock_symbol, find_symbol_by_company_name, find_company_name_by_symbol BASE_URL = os.environ.get("BASE_URL", "http://localhost:8000") # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) PORT = int(os.environ.get("PORT", 9099)) mcp = FastMCP( name="nepseapi-mcp-server", on_duplicate_tools="error", on_duplicate_resources="warn", on_duplicate_prompts="replace", stateless_http=True ) # --- Rate Limiting Middleware --- mcp.add_middleware(RateLimitingMiddleware( max_requests_per_second=60/60, # 60 per minute # change rate limit from here burst_capacity=6 )) mcp.add_middleware(ErrorHandlingMiddleware( include_traceback=True, transform_errors=True, error_callback=lambda request, exc: logger.error(f"Error in MCP server: {exc}"), )) mcp.add_middleware(RetryMiddleware( max_retries=3, retry_exceptions=(ConnectionError, TimeoutError) )) @mcp.prompt( name="stock-quick-lookup", description="Get a quick summary of a stock's current price, volume, and latest trades." ) def stock_quick_lookup(symbol: str) -> PromptMessage: """Get a quick summary of a stock's current price, volume, and latest trades.""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Show me a quick summary for {symbol}.")) @mcp.prompt( name="market-sentiment-snapshot", description="Get a snapshot of today's top gainers, losers, and overall market mood." ) def market_sentiment_snapshot() -> PromptMessage: """Get a snapshot of today's top gainers, losers, and overall market mood.""" return PromptMessage(role="user", content=TextContent(type="text", text="Give me today's top gainers, losers, and a market summary.")) @mcp.prompt( name="sector-performance", description="Analyze the performance of a specific sector today." ) def sector_performance(sector: str) -> PromptMessage: """Analyze the performance of a specific sector today.""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Analyze today's performance for the {sector} sector.")) @mcp.prompt( name="company-deep-dive", description="Get a detailed report on a company: profile, price history, and recent trades." ) def company_deep_dive(symbol: str) -> PromptMessage: """Get a detailed report on a company: profile, price history, and recent trades.""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Give me a detailed report for {symbol}, including profile, price history, and recent trades.")) @mcp.prompt( name="live-market-watchlist", description="Monitor live prices and volumes for a custom list of stocks." ) def live_market_watchlist(symbols: str) -> PromptMessage: """Monitor live prices and volumes for a custom list of stocks.""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Show me live prices and volumes for: {symbols}")) @mcp.prompt( name="market-depth-analyzer", description="Analyze the current bid/ask depth for a stock (only when market is open)." ) def market_depth_analyzer(symbol: str) -> PromptMessage: """Analyze the current bid/ask depth for a stock (only when market is open).""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Analyze the current market depth for {symbol}.")) @mcp.prompt( name="post-market-trade-explorer", description="Explore all trades for a stock after market close (floorsheet)." ) def post_market_trade_explorer(symbol: str) -> PromptMessage: """Explore all trades for a stock after market close (floorsheet).""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Show me all trades for {symbol} after market close.")) @mcp.prompt( name="validate-stock-symbol", description="Check if a stock symbol is valid and get suggestions if not." ) def validate_stock_symbol_prompt(symbol: str) -> PromptMessage: """Check if a stock symbol is valid and get suggestions if not.""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Validate the stock symbol: {symbol}")) @mcp.prompt( name="market-open-status", description="Check if the NEPSE market is currently open or closed." ) def market_open_status() -> PromptMessage: """Check if the NEPSE market is currently open or closed.""" return PromptMessage(role="user", content=TextContent(type="text", text="Is the NEPSE market currently open or closed?")) @mcp.prompt( name="setup-alert", description="Set up a price or volume alert for a stock (UI clients can use this to trigger notifications)." ) def setup_alert(symbol: str, type: str, threshold: str) -> PromptMessage: """Set up a price or volume alert for a stock (UI clients can use this to trigger notifications).""" return PromptMessage(role="user", content=TextContent(type="text", text=f"Set up an alert for {symbol} when {type} crosses {threshold}.")) # Pydantic Models for Type Validation class Summary(BaseModel): totalTurnoverRs: float = Field(..., alias="Total Turnover Rs:") totalTradedShares: float = Field(..., alias="Total Traded Shares") totalTransactions: float = Field(..., alias="Total Transactions") totalScripsTraded: float = Field(..., alias="Total Scrips Traded") class PriceVolumeItem(BaseModel): securityId: str securityName: str symbol: str indexId: int totalTradeQuantity: int lastTradedPrice: float percentageChange: float previousClose: float closePrice: Optional[float] = None class SupplyDemand(BaseModel): symbol: str totalOrder: int totalQuantity: int securityName: str securityId: Optional[str] = None class SupplyDemandData(BaseModel): supplyList: List[SupplyDemand] demandList: List[SupplyDemand] class TopGainerLoser(BaseModel): symbol: str ltp: float pointChange: float percentageChange: float securityName: str securityId: int class TopTradeScrip(BaseModel): symbol: str shareTraded: int closingPrice: float securityName: str securityId: int class TopTurnover(BaseModel): symbol: str turnover: float closingPrice: float securityName: str securityId: int class TopTraders(BaseModel): securityId: int totalTrades: int lastTradedPrice: float securityName: str symbol: str class TopTransactions(BaseModel): securityId: int totalTrades: int lastTradedPrice: float securityName: str symbol: str class TopTransaction(BaseModel): securityId: int totalTrades: int lastTradedPrice: int securityName: str symbol: str class MarketStatus(BaseModel): isOpen: str asOf: str id: int class CompanyInfo(BaseModel): id: int companyName: str symbol: str securityName: str status: str companyEmail: str website: str sectorName: str regulatoryBody: str instrumentType: str class LiveMarketItem(BaseModel): securityId: str securityName: str symbol: str indexId: int openPrice: float highPrice: float lowPrice: float totalTradeQuantity: int totalTradeValue: float lastTradedPrice: float percentageChange: float lastUpdatedDateTime: str lastTradedVolume: int previousClose: float averageTradedPrice: float # These fields are sometimes missing in the API response, so make them optional totalTradedVolume: Optional[int] = None numberOfTrades: Optional[int] = None # Additional Pydantic Models for remaining tools class MarketIndex(BaseModel): id: int auditId: Optional[int] = None exchangeIndexId: Optional[int] = None generatedTime: str index: str close: float high: float low: float previousClose: float change: float perChange: float fiftyTwoWeekHigh: float fiftyTwoWeekLow: float currentValue: float class SubIndex(BaseModel): id: int index: str change: float perChange: float currentValue: float class TradeContract(BaseModel): contractId: int stockSymbol: str buyerMemberId: str sellerMemberId: str contractQuantity: int contractRate: float contractAmount: float businessDate: str tradeBookId: int stockId: int buyerBrokerName: str sellerBrokerName: str tradeTime: str securityName: str class HistoricalTradeEntry(BaseModel): businessDate: str totalTrades: int totalTradedQuantity: int totalTradedValue: float highPrice: float lowPrice: float closePrice: float class MarketDepthItem(BaseModel): stockId: int orderBookOrderPrice: float quantity: int orderCount: int isBuy: int # 1 or 2 buy: bool sell: bool class MarketDepthData(BaseModel): buyMarketDepthList: List[MarketDepthItem] sellMarketDepthList: List[MarketDepthItem] class MarketDepthResponse(BaseModel): symbol: str totalBuyQty: int marketDepth: MarketDepthData totalSellQty: int timeStamp: Optional[int] = None class SecurityDailyTradeDto(BaseModel): securityId: str openPrice: float highPrice: float lowPrice: float totalTradeQuantity: int totalTrades: int lastTradedPrice: float previousClose: float businessDate: str closePrice: float fiftyTwoWeekHigh: float fiftyTwoWeekLow: float lastUpdatedDateTime: str class InstrumentType(BaseModel): id: int code: str description: str activeStatus: str class ShareGroup(BaseModel): id: int name: str description: str capitalRangeMin: int modifiedBy: Optional[str] = None modifiedDate: Optional[str] = None activeStatus: str isDefault: str class SectorMaster(BaseModel): id: int sectorDescription: str activeStatus: str regulatoryBody: str class CompanyId(BaseModel): id: int companyShortName: str companyName: str email: str companyWebsite: str companyContactPerson: str sectorMaster: SectorMaster companyRegistrationNumber: str activeStatus: str class Security(BaseModel): id: int symbol: str isin: str permittedToTrade: str listingDate: str creditRating: Optional[str] = None tickSize: float instrumentType: InstrumentType capitalGainBaseDate: str faceValue: float highRangeDPR: float issuerName: Optional[str] = None meInstanceNumber: int parentId: Optional[int] = None recordType: int schemeDescription: Optional[str] = None schemeName: Optional[str] = None secured: Optional[str] = None series: Optional[str] = None shareGroupId: ShareGroup activeStatus: str divisor: int cdsStockRefId: int securityName: str tradingStartDate: str networthBasePrice: float securityTradeCycle: int isPromoter: str companyId: CompanyId class SecurityOverview(BaseModel): securityDailyTradeDto: SecurityDailyTradeDto security: Security stockListedShares: float paidUpCapital: float issuedCapital: float marketCapitalization: float publicShares: int publicPercentage: float promoterShares: float promoterPercentage: float updatedDate: str securityId: int class TurnoverIndex(BaseModel): id: int index: str change: float perChange: float currentValue: float class ScripDetail(BaseModel): symbol: str sector: str Turnover: float transaction: int volume: int previousClose: float lastUpdatedDateTime: int name: str category: str pointChange: float percentageChange: float ltp: float class SectorDetail(BaseModel): transaction: int volume: int totalTurnover: float turnover: TurnoverIndex sectorName: str class MarketSummary(BaseModel): scripsDetails: Dict[str, ScripDetail] sectorsDetails: Dict[str, SectorDetail] class IndexData(BaseModel): id: int auditId: Optional[int] exchangeIndexId: Optional[int] generatedTime: Optional[str] index: str close: float high: float low: float previousClose: float change: float perChange: float fiftyTwoWeekHigh: float fiftyTwoWeekLow: float currentValue: float from pydantic import RootModel class NepseIndex(RootModel[Dict[str, IndexData]]): pass class AllIndices(RootModel[Dict[str, IndexData]]): pass class TimeValue(BaseModel): timestamp: int value: float class TimeSeriesData(BaseModel): data: List[TimeValue] @classmethod def from_list(cls, raw: List[List[float]]) -> "TimeSeriesData": return cls(data=[TimeValue(timestamp=t[0], value=t[1]) for t in raw]) # Global cache for endpoint responses (thread-safe) _endpoint_cache = {} _endpoint_cache_lock = threading.Lock() _ENDPOINT_CACHE_TTL = 600 # 10 minutes def fetch_nepse_api(endpoint: str) -> Dict[str, Any]: """Fetch data from the NEPSE API and return parsed JSON, with endpoint-level caching.""" now = time.time() cache_key = endpoint with _endpoint_cache_lock: cached = _endpoint_cache.get(cache_key) if cached: data, expires_at = cached if now < expires_at: return data else: del _endpoint_cache[cache_key] url = f"{BASE_URL}{endpoint}" response = httpx.get(url, timeout=30.0) response.raise_for_status() data = response.json() with _endpoint_cache_lock: _endpoint_cache[cache_key] = (data, now + _ENDPOINT_CACHE_TTL) return data def validate_and_return(data: Any, model_class: BaseModel, is_list: bool = False): """Validate data against Pydantic model and return validated result.""" try: if is_list: return [model_class(**item) for item in data] else: return model_class(**data) except ValidationError as e: logger.error(f"Validation error for {model_class.__name__}: {e}") return data # Return raw data if validation fails @mcp.tool() def ping() -> Dict[str, bool]: return {"pong": True} @mcp.tool() def get_market_status() -> Dict[str, Any]: """ Get the current status of the NEPSE market. Returns: Dict with keys: - isOpen: "OPEN" or "CLOSED" - asOf: Timestamp of the status - id: Status identifier Use this tool to check if the NEPSE market is currently open or closed before making live market or trading queries. """ try: market_status_response = fetch_nepse_api("/IsNepseOpen") market_status = validate_and_return(market_status_response, MarketStatus) return market_status.model_dump() if hasattr(market_status, 'model_dump') else market_status_response except Exception as e: logger.error(f"Error fetching market status: {e}") return {"error": str(e)} def check_market_open() -> bool: """ Returns True if the NEPSE market is currently open, False if closed. Use this tool to programmatically check market status before calling live market or trading tools. """ try: market_status_response = fetch_nepse_api("/IsNepseOpen") market_status = validate_and_return(market_status_response, MarketStatus) # Check if market is open using both validated model and raw response if hasattr(market_status, 'isOpen'): return market_status.isOpen == 'OPEN' else: return market_status_response.get('isOpen') == 'OPEN' except Exception as e: logger.error(f"Error checking market status: {e}") return False # Assume market is closed if we can't check @mcp.tool() def get_market_summary() -> Dict[str, float]: """ Get the latest live NEPSE market summary including key metrics. Returns: Dict with keys: - totalTurnoverRs: Total turnover in rupees - totalTradedShares: Total number of shares traded - totalTransactions: Total number of transactions - totalScripsTraded: Total number of scrips traded Use this tool for a quick overview of the day's market activity. """ try: response = fetch_nepse_api("/Summary") validated_data = validate_and_return(response, Summary) return validated_data.model_dump() if hasattr(validated_data, 'model_dump') else response except Exception as e: logger.error(f"Error fetching market summary: {e}") return {"error": str(e)} @mcp.tool() def get_nepse_subindex() -> Dict: """ Get all NEPSE subindices (sector indices). Use this to get the live performance of indexes like Development Bank Index Manufacturing And Processing Microfinance Index Life Insurance Mutual Fund Banking SubIndex Hotels And Tourism Index Others Index HydroPower Index Non Life Insurance Finance Index Trading Index Investment Index Returns: Dict mapping subindex names to their data: - id: Index ID - index: Name of the subindex - change: Absolute change in index value - perChange: Percentage change in index value - currentValue: Current value of the subindex Use this tool to get the latest values for all sector indices (e.g., Banking, HydroPower, Finance, etc.). """ try: response = fetch_nepse_api("/NepseSubIndices") validated = validate_and_return(response, AllIndices) return validated.__root__ if hasattr(validated, "__root__") else response except Exception as e: logger.error(f"Error fetching NEPSE subindices: {e}") return {"error": str(e)} @mcp.tool() def get_nepse_index() -> Dict: """ Get the NEPSE index and related indices. Provides detailed live performance data for the these index. Sensitive Float Index Float Index Sensitive Index NEPSE Index Returns: Dict mapping index names to their data: - id: Index ID - auditId: (optional) Audit ID - exchangeIndexId: (optional) Exchange Index ID - generatedTime: (optional) Time the index was generated - index: Name of the index - close: Closing value - high: Highest value - low: Lowest value - previousClose: Previous closing value - change: Absolute change in index value - perChange: Percentage change in index value - fiftyTwoWeekHigh: 52-week high value - fiftyTwoWeekLow: 52-week low value - currentValue: Current value of the index Use this tool to get the latest NEPSE index and related index values. """ try: response = fetch_nepse_api("/NepseIndex") validated = validate_and_return(response, NepseIndex) return validated.__root__ if hasattr(validated, "__root__") else response except Exception as e: logger.error(f"Error fetching NEPSE index: {e}") return {"error": str(e)} def _get_index_graph(endpoint: str,limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Helper to fetch and paginate index graph data from NEPSE API. Returns paginated time series data. """ try: raw_data = fetch_nepse_api(endpoint) # The API returns a list of [timestamp, value] pairs parsed = TimeSeriesData.from_list(raw_data) items = [tv.model_dump() if hasattr(tv, 'model_dump') else tv for tv in parsed.data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching {endpoint}: {e}") return {"error": str(e)} @mcp.tool() def get_daily_nepse_index_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily NEPSE index graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyNepseIndexGraph", limit, page) @mcp.tool() def get_daily_sensitive_index_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Sensitive index graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailySensitiveIndexGraph", limit, page) @mcp.tool() def get_daily_float_index_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Float index graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyFloatIndexGraph", limit, page) @mcp.tool() def get_daily_sensitive_float_index_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Sensitive Float index graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailySensitiveFloatIndexGraph", limit, page) @mcp.tool() def get_daily_bank_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Bank subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyBankSubindexGraph", limit, page) @mcp.tool() def get_daily_development_bank_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Development Bank subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyDevelopmentBankSubindexGraph", limit, page) @mcp.tool() def get_daily_finance_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Finance subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyFinanceSubindexGraph", limit, page) @mcp.tool() def get_daily_hotel_tourism_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Hotel & Tourism subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyHotelTourismSubindexGraph", limit, page) @mcp.tool() def get_daily_hydropower_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Hydropower subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyHydroPowerSubindexGraph", limit, page) @mcp.tool() def get_daily_investment_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Investment subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyInvestmentSubindexGraph", limit, page) @mcp.tool() def get_daily_life_insurance_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Life Insurance subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyLifeInsuranceSubindexGraph", limit, page) @mcp.tool() def get_daily_manufacturing_processing_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Manufacturing & Processing subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyManufacturingProcessingSubindexGraph", limit, page) @mcp.tool() def get_daily_microfinance_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Microfinance subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyMicrofinanceSubindexGraph", limit, page) @mcp.tool() def get_daily_mutual_fund_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Mutual Fund subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyMutualFundSubindexGraph", limit, page) @mcp.tool() def get_daily_non_life_insurance_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Non-Life Insurance subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyNonLifeInsuranceSubindexGraph", limit, page) @mcp.tool() def get_daily_others_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Others subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyOthersSubindexGraph", limit, page) @mcp.tool() def get_daily_trading_subindex_graph(limit: Optional[int] = None, page: Optional[int] = 1) -> dict: """ Get daily Trading subindex graph (time series data). Supports pagination. Returns paginated list of {timestamp, value}. """ return _get_index_graph("/DailyTradingSubindexGraph", limit, page) @mcp.tool() def get_live_market(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get real-time live market data for all securities with pagination support. Returns: Dict with: - results: List of securities, each with fields: - securityId, securityName, symbol, indexId, openPrice, highPrice, lowPrice, totalTradeQuantity, totalTradeValue, lastTradedPrice, percentageChange, lastUpdatedDateTime, lastTradedVolume, previousClose, averageTradedPrice, totalTradedVolume (optional), numberOfTrades (optional) - total: Total number of securities - page: Current page number - limit: Number of results per page Use this tool to monitor live prices and volumes for all stocks when the market is open. """ try: # Check if market is open first if not check_market_open(): return {"error": "Market is closed. This tool only works when the market is open."} response = fetch_nepse_api("/LiveMarket") validated_data = validate_and_return(response, LiveMarketItem, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching live market: {e}") return {"error": str(e)} def paginate_list(items, limit: Optional[int], page: Optional[int]): """Paginate a list of items. Returns (paged_items, total, page, limit).""" total = len(items) if limit is None or not isinstance(limit, int) or limit <= 0: limit = 10 if page is None or not isinstance(page, int) or page <= 0: page = 1 start = (page - 1) * limit end = start + limit paged_items = items[start:end] return paged_items, total, page, limit @mcp.tool() def get_price_volume(company: str = "", limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get price and volume data for all stocks, or filter by company name or symbol. Supports pagination. Args: company: (optional) Company name or symbol to filter (case-insensitive, partial match allowed). Leave empty for all stocks. limit: (optional) Number of results per page (default: 10). page: (optional) Page number for pagination (default: 1). Returns: Dict with: - results: List of stocks, each with fields: - securityId, securityName, symbol, indexId, totalTradeQuantity, lastTradedPrice, percentageChange, previousClose, closePrice (optional) - total: Total number of stocks matching the filter - page: Current page number - limit: Number of results per page Use this tool to get price/volume for all stocks or search by company/symbol. """ try: # Handle MCP framework parameter issues - convert various empty/null representations to None if company in ('None', 'null', '', 'undefined', None): company = None response = fetch_nepse_api("/PriceVolume") validated_data = validate_and_return(response, PriceVolumeItem, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] # Filter by company if provided if company is not None and company.strip(): company_clean = company.strip() logger.info(f"Filtering by company: '{company_clean}'") # Step 1: Check if it's a valid symbol first validation_result = validate_stock_symbol(company_clean) if validation_result.get("valid"): validated_symbol = validation_result["symbol"] logger.info(f"Found valid symbol: {validated_symbol}") filtered = [item for item in items if item.get("symbol", "").strip().upper() == validated_symbol.upper()] if filtered: items = filtered logger.info(f"Filtered to {len(items)} items by symbol") else: return {"error": f"No price/volume data found for symbol '{validated_symbol}'."} else: # Step 2: Try to find symbol by company name logger.info(f"Not a valid symbol, trying company name lookup") symbol_lookup = find_symbol_by_company_name(company_clean) if symbol_lookup.get("found"): matching_symbols = [match["symbol"] for match in symbol_lookup.get("matches", [])] logger.info(f"Found symbols from company name lookup: {matching_symbols}") filtered = [item for item in items if item.get("symbol", "").strip().upper() in [s.upper() for s in matching_symbols]] if filtered: items = filtered logger.info(f"Filtered to {len(items)} items by company name") else: return {"error": f"No price/volume data found for company '{company_clean}'."} else: # Step 3: No match found, return all companies with pagination logger.info(f"No exact match found for '{company_clean}', returning all companies with pagination") else: logger.info("No company filter provided, returning paginated results") paged_items, total, page, limit = paginate_list(items, limit, page) logger.info(f"Returning {len(paged_items)} items out of {total} total") return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching price volume: {e}") return {"error": str(e)} @mcp.tool() def get_top_gainers(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get list of top gaining stocks with pagination support. Returns: Dict with: - results: List of stocks, each with fields: - symbol, ltp (last traded price), pointChange, percentageChange, securityName, securityId - total: Total number of top gainers - page: Current page number - limit: Number of results per page Use this tool to find which stocks have gained the most (by percentage) today. """ try: response = fetch_nepse_api("/TopGainers") validated_data = validate_and_return(response, TopGainerLoser, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching top gainers: {e}") return {"error": str(e)} @mcp.tool() def get_top_losers(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get list of top losing stocks with pagination support. Returns: Dict with: - results: List of stocks, each with fields: - symbol, ltp, pointChange, percentageChange, securityName, securityId - total: Total number of top losers - page: Current page number - limit: Number of results per page Use this tool to find which stocks have lost the most (by percentage) today. """ try: response = fetch_nepse_api("/TopLosers") validated_data = validate_and_return(response, TopGainerLoser, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching top losers: {e}") return {"error": str(e)} @mcp.tool() def get_company_list(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get list of all companies listed in NEPSE with pagination support. Returns: Dict with: - results: List of companies, each with fields: - id, companyName, symbol, securityName, status, companyEmail, website, sectorName, regulatoryBody, instrumentType - total: Total number of companies - page: Current page number - limit: Number of results per page Use this tool to browse or search all listed companies. """ try: response = fetch_nepse_api("/CompanyList") validated_data = validate_and_return(response, CompanyInfo, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching company list: {e}") return {"error": str(e)} @mcp.tool() def get_top_turnover(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get top companies by turnover with pagination support. Returns: Dict with: - results: List of companies, each with fields: - symbol, turnover, closingPrice, securityName, securityId - total: Total number of top turnover companies - page: Current page number - limit: Number of results per page Use this tool to find companies with the highest trading turnover today. """ try: response = fetch_nepse_api("/TopTenTurnoverScrips") validated_data = validate_and_return(response, TopTurnover, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching top turnover: {e}") return {"error": str(e)} @mcp.tool() def get_top_traders(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get top traders by volume of Nepse securities with pagination support. Returns: Dict with: - results: List of securities, each with fields: - securityId, totalTrades, lastTradedPrice, securityName, symbol - total: Total number of top traders - page: Current page number - limit: Number of results per page Use this tool to find which securities had the most trades today. """ try: response = fetch_nepse_api("/TopTenTradeScrips") validated_data = validate_and_return(response, TopTraders, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching top traders: {e}") return {"error": str(e)} @mcp.tool() def get_top_transactions(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get top transactions by value for Nepse securities with pagination support. Returns: Dict with: - results: List of securities, each with fields: - securityId, totalTrades, lastTradedPrice, securityName, symbol - total: Total number of top transactions - page: Current page number - limit: Number of results per page Use this tool to find which securities had the highest transaction values today. """ try: response = fetch_nepse_api("/TopTenTransactionScrips") validated_data = validate_and_return(response, TopTransactions, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching top transactions: {e}") return {"error": str(e)} @mcp.tool() def get_floorsheet(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get today's floorsheet data (all transactions) with pagination support. Returns: Dict with: - results: List of trade contracts, each with fields: - contractId, stockSymbol, buyerMemberId, sellerMemberId, contractQuantity, contractRate, contractAmount, businessDate, tradeBookId, stockId, buyerBrokerName, sellerBrokerName, tradeTime, securityName - total: Total number of trades - page: Current page number - limit: Number of results per page Use this tool to explore all trades executed today after market close. """ try: if check_market_open(): return {"error": "Market is open. This tool works when the market is closed, it does not work when the market is open."} response = fetch_nepse_api("/Floorsheet") validated_data = validate_and_return(response, TradeContract, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching floorsheet: {e}") return {"error": str(e)} @mcp.tool() def get_company_floorsheet(symbol: str, limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get floorsheet data for a specific company with pagination support. Args: symbol: Stock symbol to filter trades. limit: (optional) Number of results per page (default: 10). page: (optional) Page number for pagination (default: 1). Returns: Dict with: - results: List of trade contracts for the company (see get_floorsheet for fields) - total: Total number of trades for the company - page: Current page number - limit: Number of results per page - symbol: The validated stock symbol Use this tool to see all trades for a specific company after market close. """ try: if check_market_open(): return {"error": "Market is open. This tool works when the market is closed, it does not work when the market is open."} validation_result = validate_stock_symbol(symbol) if not validation_result["valid"]: return {"error": validation_result["error"]} validated_symbol = validation_result["symbol"] response = fetch_nepse_api(f"/FloorsheetOf?symbol={validated_symbol}") validated_data = validate_and_return(response, TradeContract, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit, "symbol": validated_symbol } except Exception as e: logger.error(f"Error fetching company floorsheet for {symbol}: {e}") return {"error": str(e)} @mcp.tool() def get_price_history(symbol: str, limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get historical price and volume data for a company with pagination support. Args: symbol: Stock symbol to get history for. limit: (optional) Number of results per page (default: 10). page: (optional) Page number for pagination (default: 1). Returns: Dict with: - results: List of historical trade entries, each with fields: - businessDate, totalTrades, totalTradedQuantity, totalTradedValue, highPrice, lowPrice, closePrice - total: Total number of historical entries - page: Current page number - limit: Number of results per page - symbol: The validated stock symbol Use this tool to analyze a company's price and volume history. """ try: validation_result = validate_stock_symbol(symbol) if not validation_result["valid"]: return {"error": validation_result["error"]} validated_symbol = validation_result["symbol"] response = fetch_nepse_api(f"/PriceVolumeHistory?symbol={validated_symbol}") validated_data = validate_and_return(response, HistoricalTradeEntry, is_list=True) items = [item.model_dump() if hasattr(item, 'model_dump') else item for item in validated_data] paged_items, total, page, limit = paginate_list(items, limit, page) return { "results": paged_items, "total": total, "page": page, "limit": limit, "symbol": validated_symbol } except Exception as e: logger.error(f"Error fetching price history for {symbol}: {e}") return {"error": str(e)} @mcp.tool() def get_market_depth(symbol: str) -> Dict: """ Get market depth (bid/ask) for a specific stock. Args: symbol: Stock symbol to get market depth for. Returns: Dict with: - symbol: The stock symbol - totalBuyQty: Total buy quantity - totalSellQty: Total sell quantity - marketDepth: Object with buyMarketDepthList and sellMarketDepthList (each a list of MarketDepthItem) - timeStamp: (optional) Timestamp of the data Use this tool to analyze the current bid/ask depth for a stock when the market is open. """ try: # Check if market is open first if not check_market_open(): return {"error": "Market is closed. This tool only works when the market is open."} # Validate symbol validation_result = validate_stock_symbol(symbol) if not validation_result["valid"]: return {"error": validation_result["error"]} validated_symbol = validation_result["symbol"] response = fetch_nepse_api(f"/MarketDepth?symbol={validated_symbol}") validated_data = validate_and_return(response, MarketDepthResponse) return validated_data.model_dump() if hasattr(validated_data, 'model_dump') else response except Exception as e: logger.error(f"Error fetching market depth for {symbol}: {e}") return {"error": str(e)} @mcp.tool() def get_supply_demand(limit: Optional[int] = None, page: Optional[int] = 1) -> Dict: """ Get the current supply and demand data for the NEPSE market, with pagination support. Args: limit: (optional) Number of results per page for both supply and demand lists (default: 10). page: (optional) Page number for pagination (default: 1). Returns: Dict with: - supplyList: Paginated list of supply (sell) orders, each with fields: - symbol, securityName, totalQuantity, totalOrder, securityId - demandList: Paginated list of demand (buy) orders, each with fields: - symbol, securityName, totalQuantity, totalOrder, securityId - totalSupply: Total number of supply entries before pagination - totalDemand: Total number of demand entries before pagination - page: Current page number - limit: Number of results per page Use this tool to analyze the current supply and demand (order book) for all stocks. Pagination is applied independently to both lists. """ try: supply_demand_response = fetch_nepse_api("/SupplyDemand") # Use the server response directly as you provided supply_items = supply_demand_response["supplyList"] demand_items = supply_demand_response["demandList"] # Paginate both lists independently paged_supply, total_supply, page, limit = paginate_list(supply_items, limit, page) paged_demand, total_demand, _, _ = paginate_list(demand_items, limit, page) return { "supplyList": paged_supply, "demandList": paged_demand, "totalSupply": total_supply, "totalDemand": total_demand, "page": page, "limit": limit } except Exception as e: logger.error(f"Error fetching supply and demand data: {e}") return {"error": str(e)} @mcp.tool() def validate_stock_symbol_tool(symbol: str) -> Dict: """ Validate if a stock symbol exists in NEPSE. Args: symbol: Stock symbol to validate. Returns: Dict with: - validation_result: Dict with keys 'valid', 'symbol', 'info' (if valid), 'error' (if invalid), and 'suggestions' (if invalid) Use this tool to check if a symbol is valid and get suggestions if not. """ try: validation_result = validate_stock_symbol(symbol) return {"validation_result": validation_result} except Exception as e: logger.error(f"Error validating stock symbol {symbol}: {e}") return {"error": str(e)} @mcp.tool() def get_company_symbol(company_name: str) -> Dict: """ Find stock symbol by company name. Use the first significant word of the company name for best results. Args: company_name: Name of the company to search for. Returns: Dict with: - search_result: Dict with keys 'found', 'query', 'matches' (list of symbol/company_name/match_type), 'total_matches', or 'error' Use this tool to find the NEPSE symbol for a company by its name. """ try: result = find_symbol_by_company_name(company_name) return {"search_result": result} except Exception as e: logger.error(f"Error finding symbol for company name {company_name}: {e}") return {"error": str(e)} @mcp.tool() def get_company_name_from_symbol(symbol: str) -> Dict: """ Find company name by stock symbol. Args: symbol: Stock symbol to look up. Returns: Dict with: - search_result: Dict with keys 'found', 'symbol', 'company_name', 'full_info', or 'error' Use this tool to get the full company name and info for a given NEPSE symbol. """ try: result = find_company_name_by_symbol(symbol) return {"search_result": result} except Exception as e: logger.error(f"Error finding company name for symbol {symbol}: {e}") return {"error": str(e)} @mcp.custom_route("/health", methods=["GET"]) async def health_check(request: Request) -> PlainTextResponse: return PlainTextResponse("OK") # Update the transport to 'stdio' for local testing, 'http' for production / remote if __name__ == "__main__": # for local testing # mcp.run(transport="stdio") # for production / remote mcp.run(transport="http", host="0.0.0.0", port=PORT)