KIS REST API MCP Server

Instructions

Implementation Reference

• server.py:559-603 (handler)

  The main handler function decorated with @mcp.tool(name="inquery-stock-history"), which executes the tool logic by authenticating with KIS API, preparing request parameters for the stock symbol and date range, and fetching daily price history data via GET request to the inquire-daily-itemchartprice endpoint.
  @mcp.tool( name="inquery-stock-history", description="Get daily stock price history from Korea Investment & Securities", ) async def inquery_stock_history(symbol: str, start_date: str, end_date: str): """ Get daily stock price history from Korea Investment & Securities Args: symbol: Stock symbol (e.g. "005930") start_date: Start date (YYYYMMDD) end_date: End date (YYYYMMDD) Returns: Dictionary containing daily stock price history """ async with httpx.AsyncClient() as client: token = await get_access_token(client) # Prepare request data request_data = { "FID_COND_MRKT_DIV_CODE": "J", # 시장구분 "FID_INPUT_ISCD": symbol, # 종목코드 "FID_INPUT_DATE_1": start_date, # 시작일자 "FID_INPUT_DATE_2": end_date, # 종료일자 "FID_PERIOD_DIV_CODE": "D", # 기간분류코드 "FID_ORG_ADJ_PRC": "0", # 수정주가원구분 } response = await client.get( f"{TrIdManager.get_domain('stock_history')}{STOCK_HISTORY_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("stock_history") }, params=request_data ) if response.status_code != 200: raise Exception(f"Failed to get stock history: {response.text}") return response.json()
• server.py:42-42 (helper)

  Global constant defining the KIS API endpoint path specifically for the stock history (daily item chart price) query used in the tool handler.
  STOCK_HISTORY_PATH = "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice" # 주식일별주가조회
• server.py:68-165 (helper)

  TrIdManager class provides transaction IDs and domains for API calls, including 'stock_history': 'FHKST03010200' used in the tool's header for authentication with real/virtual accounts.
  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