KIS REST API MCP Server

Instructions

Implementation Reference

• server.py:645-725 (handler)

  The handler function decorated with @mcp.tool(name="order-overseas-stock"), which implements the core logic for ordering overseas stocks. It normalizes inputs, determines TR_ID based on market and order_type, prepares the request data, and sends a POST request to the KIS overseas order API.
  @mcp.tool( name="order-overseas-stock", description="Order overseas stock (buy/sell) from Korea Investment & Securities", ) async def order_overseas_stock(symbol: str, quantity: int, price: float, order_type: str, market: str): """ Order overseas stock (buy/sell) Args: symbol: Stock symbol (e.g. "AAPL") quantity: Order quantity price: Order price (0 for market price) order_type: Order type ("buy" or "sell", case-insensitive) market: Market code ("NASD" for NASDAQ, "NYSE" for NYSE, etc.) Returns: Dictionary containing order information """ # Normalize order_type to lowercase order_type = order_type.lower() if order_type not in ["buy", "sell"]: raise ValueError('order_type must be either "buy" or "sell"') # Normalize market code to uppercase market = market.upper() if market not in MARKET_CODES: raise ValueError(f"Unsupported market: {market}. Supported markets: {', '.join(MARKET_CODES.keys())}") async with httpx.AsyncClient() as client: token = await get_access_token(client) # Get market prefix for TR_ID market_prefix = { "NASD": "us", # 나스닥 "NYSE": "us", # 뉴욕 "AMEX": "us", # 아멕스 "SEHK": "hk", # 홍콩 "SHAA": "sh", # 중국상해 "SZAA": "sz", # 중국심천 "TKSE": "jp", # 일본 "HASE": "vn", # 베트남 하노이 "VNSE": "vn", # 베트남 호치민 }.get(market) if not market_prefix: raise ValueError(f"Unsupported market: {market}") tr_id_key = f"{market_prefix}_{order_type}" tr_id = TrIdManager.get_tr_id(tr_id_key) if not tr_id: raise ValueError(f"Invalid operation type: {tr_id_key}") # Prepare request data request_data = { "CANO": os.environ["KIS_CANO"], # 계좌번호 "ACNT_PRDT_CD": "01", # 계좌상품코드 "OVRS_EXCG_CD": market, # 해외거래소코드 "PDNO": symbol, # 종목코드 "ORD_QTY": str(quantity), # 주문수량 "OVRS_ORD_UNPR": str(price), # 주문단가 "ORD_SVR_DVSN_CD": "0", # 주문서버구분코드 "ORD_DVSN": "00" if price > 0 else "01" # 주문구분 (00: 지정가, 01: 시장가) } response = await client.post( f"{TrIdManager.get_domain(order_type)}{OVERSEAS_ORDER_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": tr_id, }, json=request_data ) if response.status_code != 200: raise Exception(f"Failed to order overseas stock: {response.text}") return response.json()
• server.py:645-648 (registration)

  The @mcp.tool decorator registers the tool with name "order-overseas-stock" and provides the description. The function signature defines the input schema.
  @mcp.tool( name="order-overseas-stock", description="Order overseas stock (buy/sell) from Korea Investment & Securities", )
• server.py:68-165 (helper)

  TrIdManager class provides transaction IDs and domains for different operations and account types (real/virtual), crucial for constructing the API requests used in the tool.
  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:56-66 (helper)

  Dictionary mapping market codes to names, used for validation in the handler.
  MARKET_CODES = { "NASD": "나스닥", "NYSE": "뉴욕", "AMEX": "아멕스", "SEHK": "홍콩", "SHAA": "중국상해", "SZAA": "중국심천", "TKSE": "일본", "HASE": "베트남 하노이", "VNSE": "베트남 호치민" }