KIS REST API MCP Server

Instructions

Implementation Reference

• server.py:355-406 (handler)

  Main handler function that executes the stock order (buy/sell) using KIS API. Handles input validation, token acquisition, hashkey generation, and POST request to order endpoint.

  async def order_stock(symbol: str, quantity: int, price: int, order_type: str):
      """
      Order stock (buy/sell) from Korea Investment & Securities
      
      Args:
          symbol: Stock symbol (e.g. "005930")
          quantity: Order quantity
          price: Order price (0 for market price)
          order_type: Order type ("buy" or "sell", case-insensitive)
          
      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"')
  
      async with httpx.AsyncClient() as client:
          token = await get_access_token(client)
          
          # Prepare request data
          request_data = {
              "CANO": os.environ["KIS_CANO"],  # 계좌번호
              "ACNT_PRDT_CD": "01",  # 계좌상품코드
              "PDNO": symbol,  # 종목코드
              "ORD_DVSN": "01" if price == 0 else "00",  # 주문구분 (01: 시장가, 00: 지정가)
              "ORD_QTY": str(quantity),  # 주문수량
              "ORD_UNPR": str(price),  # 주문단가
          }
          
          # Get hashkey
          hashkey = await get_hashkey(client, token, request_data)
          
          response = await client.post(
              f"{TrIdManager.get_domain(order_type)}{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": TrIdManager.get_tr_id(order_type),
                  "hashkey": hashkey
              },
              json=request_data
          )
          
          if response.status_code != 200:
              raise Exception(f"Failed to order stock: {response.text}")
          
          return response.json()

• server.py:351-354 (registration)

  Registration of the 'order-stock' tool using FastMCP @mcp.tool decorator.

  @mcp.tool(
      name="order-stock",
      description="Order stock (buy/sell) from Korea Investment & Securities",
  )

• server.py:223-250 (helper)

  Helper function to generate hashkey required for stock order authentication.

  async def get_hashkey(client: httpx.AsyncClient, token: str, body: dict) -> str:
      """
      Get hash key for order request
      
      Args:
          client: httpx client
          token: Access token
          body: Request body
          
      Returns:
          str: Hash key
      """
      response = await client.post(
          f"{TrIdManager.get_domain('buy')}{HASHKEY_PATH}",
          headers={
              "content-type": CONTENT_TYPE,
              "authorization": f"{AUTH_TYPE} {token}",
              "appkey": os.environ["KIS_APP_KEY"],
              "appsecret": os.environ["KIS_APP_SECRET"],
          },
          json=body
      )
      
      if response.status_code != 200:
          raise Exception(f"Failed to get hash key: {response.text}")
      
      return response.json()["HASH"]

• server.py:193-222 (helper)

  Helper function to acquire and cache access token for API authentication.

  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