Yahoo Finance MCP Server

Instructions

Implementation Reference

• server.py:663-682 (handler)

  The main handler function that implements the search_companies tool. It queries the Financial Modeling Prep API search endpoint with the given query, limit, and optional exchange, returning JSON data of matching companies.

  async def search_companies(query: str, limit: int = 10, exchange: str = "") -> str:
      """根据关键字搜索公司"""
  
      api_key = os.environ.get("FMP_API_KEY")
      if not api_key:
          return "Error: FMP_API_KEY environment variable not set."
  
      params = {"query": query, "limit": limit, "apikey": api_key}
      if exchange:
          params["exchange"] = exchange
  
      url = "https://financialmodelingprep.com/api/v3/search"
      try:
          resp = requests.get(url, params=params, timeout=10)
          resp.raise_for_status()
          data = resp.json()
      except Exception as e:
          return f"Error: searching companies: {e}"
      return json.dumps(data)

• server.py:650-662 (registration)

  The @fmp_server.tool decorator registers the search_companies tool, specifying its name and description including input parameter schema.

  @fmp_server.tool(
      name="search_companies",
      description="""根据关键字搜索公司信息。
  
  参数说明：
      query: str
          搜索关键字
      limit: int
          返回结果数量，默认 10
      exchange: str
          交易所代码，可选
  """,
  )

• server.py:663-682 (schema)

  Type annotations in the function signature define the input schema: query (str), limit (int, default 10), exchange (str, default empty), returning str (JSON).

  async def search_companies(query: str, limit: int = 10, exchange: str = "") -> str:
      """根据关键字搜索公司"""
  
      api_key = os.environ.get("FMP_API_KEY")
      if not api_key:
          return "Error: FMP_API_KEY environment variable not set."
  
      params = {"query": query, "limit": limit, "apikey": api_key}
      if exchange:
          params["exchange"] = exchange
  
      url = "https://financialmodelingprep.com/api/v3/search"
      try:
          resp = requests.get(url, params=params, timeout=10)
          resp.raise_for_status()
          data = resp.json()
      except Exception as e:
          return f"Error: searching companies: {e}"
      return json.dumps(data)