order-stock
Execute buy or sell orders for stocks through Korea Investment & Securities, specifying symbol, quantity, price, and order type to manage investments.
Instructions
Order stock (buy/sell) from Korea Investment & Securities
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | ||
| quantity | Yes | ||
| price | Yes | ||
| order_type | Yes |
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