inquery-stock-price
Retrieve current stock price data from Korea Investment & Securities to monitor market values and inform investment decisions.
Instructions
Get current stock price information from Korea Investment & Securities
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes |
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