KIS REST API MCP Server

Instructions

Implementation Reference

• server.py:255-297 (handler)

  The main asynchronous handler function that executes the tool logic: authenticates with KIS API, fetches current stock price for given symbol, and returns the price data.

  async def inquery_stock_price(symbol: str):
      """
      Get current stock price information from Korea Investment & Securities
      
      Args:
          symbol: Stock symbol (e.g. "005930" for Samsung Electronics)
          
      Returns:
          Dictionary containing stock price information including:
          - stck_prpr: Current price
          - prdy_vrss: Change from previous day
          - prdy_vrss_sign: Change direction (+/-)
          - prdy_ctrt: Change rate (%)
          - acml_vol: Accumulated volume
          - acml_tr_pbmn: Accumulated trade value
          - hts_kor_isnm: Stock name in Korean
          - stck_mxpr: High price of the day
          - stck_llam: Low price of the day
          - stck_oprc: Opening price
          - stck_prdy_clpr: Previous day's closing price
      """
      async with httpx.AsyncClient() as client:
          token = await get_access_token(client)
          response = await client.get(
              f"{TrIdManager.get_domain('price')}{STOCK_PRICE_PATH}",
              headers={
                  "content-type": CONTENT_TYPE,
                  "authorization": f"{AUTH_TYPE} {token}",
                  "appkey": os.environ["KIS_APP_KEY"],
                  "appsecret": os.environ["KIS_APP_SECRET"],
                  "tr_id": TrIdManager.get_tr_id("price")
              },
              params={
                  "fid_cond_mrkt_div_code": "J",
                  "fid_input_iscd": symbol
              }
          )
          
          if response.status_code != 200:
              raise Exception(f"Failed to get stock price: {response.text}")
          
          return response.json()["output"]

• server.py:251-254 (registration)

  Tool registration decorator that registers the 'inquery-stock-price' tool with the MCP server, including name and description which serves as input schema.

  @mcp.tool(
      name="inquery-stock-price",
      description="Get current stock price information from Korea Investment & Securities",
  )

• server.py:256-275 (schema)

  Docstring providing detailed input (symbol: str) and output schema description for the tool.

  """
  Get current stock price information from Korea Investment & Securities
  
  Args:
      symbol: Stock symbol (e.g. "005930" for Samsung Electronics)
      
  Returns:
      Dictionary containing stock price information including:
      - stck_prpr: Current price
      - prdy_vrss: Change from previous day
      - prdy_vrss_sign: Change direction (+/-)
      - prdy_ctrt: Change rate (%)
      - acml_vol: Accumulated volume
      - acml_tr_pbmn: Accumulated trade value
      - hts_kor_isnm: Stock name in Korean
      - stck_mxpr: High price of the day
      - stck_llam: Low price of the day
      - stck_oprc: Opening price
      - stck_prdy_clpr: Previous day's closing price
  """

• server.py:68-165 (helper)

  TrIdManager class providing transaction IDs and domains for KIS API calls, used by the tool for 'price' operation.

  class TrIdManager:
      """Transaction ID manager for Korea Investment & Securities API"""
      
      # 실전계좌용 TR_ID
      REAL = {
          # 국내주식
          "balance": "TTTC8434R",  # 잔고조회
          "price": "FHKST01010100",  # 현재가조회
          "buy": "TTTC0802U",  # 주식매수
          "sell": "TTTC0801U",  # 주식매도
          "order_list": "TTTC8001R",  # 일별주문체결조회
          "order_detail": "TTTC8036R",  # 주문체결내역조회
          "stock_info": "FHKST01010400",  # 일별주가조회
          "stock_history": "FHKST03010200",  # 주식일별주가조회
          "stock_ask": "FHKST01010200",  # 주식호가조회
          
          # 해외주식
          "us_buy": "TTTT1002U",      # 미국 매수 주문
          "us_sell": "TTTT1006U",     # 미국 매도 주문
          "jp_buy": "TTTS0308U",      # 일본 매수 주문
          "jp_sell": "TTTS0307U",     # 일본 매도 주문
          "sh_buy": "TTTS0202U",      # 상해 매수 주문
          "sh_sell": "TTTS1005U",     # 상해 매도 주문
          "hk_buy": "TTTS1002U",      # 홍콩 매수 주문
          "hk_sell": "TTTS1001U",     # 홍콩 매도 주문
          "sz_buy": "TTTS0305U",      # 심천 매수 주문
          "sz_sell": "TTTS0304U",     # 심천 매도 주문
          "vn_buy": "TTTS0311U",      # 베트남 매수 주문
          "vn_sell": "TTTS0310U",     # 베트남 매도 주문
      }
      
      # 모의계좌용 TR_ID
      VIRTUAL = {
          # 국내주식
          "balance": "VTTC8434R",  # 잔고조회
          "price": "FHKST01010100",  # 현재가조회
          "buy": "VTTC0802U",  # 주식매수
          "sell": "VTTC0801U",  # 주식매도
          "order_list": "VTTC8001R",  # 일별주문체결조회
          "order_detail": "VTTC8036R",  # 주문체결내역조회
          "stock_info": "FHKST01010400",  # 일별주가조회
          "stock_history": "FHKST03010200",  # 주식일별주가조회
          "stock_ask": "FHKST01010200",  # 주식호가조회
          
          # 해외주식
          "us_buy": "VTTT1002U",      # 미국 매수 주문
          "us_sell": "VTTT1001U",     # 미국 매도 주문
          "jp_buy": "VTTS0308U",      # 일본 매수 주문
          "jp_sell": "VTTS0307U",     # 일본 매도 주문
          "sh_buy": "VTTS0202U",      # 상해 매수 주문
          "sh_sell": "VTTS1005U",     # 상해 매도 주문
          "hk_buy": "VTTS1002U",      # 홍콩 매수 주문
          "hk_sell": "VTTS1001U",     # 홍콩 매도 주문
          "sz_buy": "VTTS0305U",      # 심천 매수 주문
          "sz_sell": "VTTS0304U",     # 심천 매도 주문
          "vn_buy": "VTTS0311U",      # 베트남 매수 주문
          "vn_sell": "VTTS0310U",     # 베트남 매도 주문
      }
      
      @classmethod
      def get_tr_id(cls, operation: str) -> str:
          """
          Get transaction ID for the given operation
          
          Args:
              operation: Operation type ('balance', 'price', 'buy', 'sell', etc.)
              
          Returns:
              str: Transaction ID for the operation
          """
          is_real_account = os.environ.get("KIS_ACCOUNT_TYPE", "REAL").upper() == "REAL"
          tr_id_map = cls.REAL if is_real_account else cls.VIRTUAL
          return tr_id_map.get(operation)
      
      @classmethod
      def get_domain(cls, operation: str) -> str:
          """
          Get domain for the given operation
          
          Args:
              operation: Operation type ('balance', 'price', 'buy', 'sell', etc.)
              
          Returns:
              str: Domain URL for the operation
          """
          is_real_account = os.environ.get("KIS_ACCOUNT_TYPE", "REAL").upper() == "REAL"
          
          # 잔고조회는 실전/모의 계좌별로 다른 도메인 사용
          if operation == "balance":
              return DOMAIN if is_real_account else VIRTUAL_DOMAIN
              
          # 조회 API는 실전/모의 동일한 도메인 사용
          if operation in ["price", "stock_info", "stock_history", "stock_ask"]:
              return DOMAIN
              
          # 거래 API는 계좌 타입에 따라 다른 도메인 사용
          return DOMAIN if is_real_account else VIRTUAL_DOMAIN

• server.py:193-222 (helper)

  Helper function to obtain and cache access token for KIS API authentication, called by the handler.

  async def get_access_token(client: httpx.AsyncClient) -> str:
      """
      Get access token with file-based caching
      Returns cached token if valid, otherwise requests new token
      """
      token, expires_at = load_token()
      if token and expires_at and datetime.now() < expires_at:
          return token
      
      token_response = await client.post(
          f"{DOMAIN}{TOKEN_PATH}",
          headers={"content-type": CONTENT_TYPE},
          json={
              "grant_type": "client_credentials",
              "appkey": os.environ["KIS_APP_KEY"],
              "appsecret": os.environ["KIS_APP_SECRET"]
          }
      )
      
      if token_response.status_code != 200:
          raise Exception(f"Failed to get token: {token_response.text}")
      
      token_data = token_response.json()
      token = token_data["access_token"]
      
      expires_at = datetime.now() + timedelta(hours=23)
      save_token(token, expires_at)
      
      return token