dart-mcp

Instructions

Implementation Reference

• dart.py:935-1125 (handler)

  Main execution function for the search_detailed_financial_data tool. Handles company lookup, disclosure listing, XBRL retrieval and parsing for detailed financial statements (balance sheet, income statement, cash flow). Formats results in markdown tables.
  @mcp.tool() async def search_detailed_financial_data( company_name: str, start_date: str, end_date: str, ctx: Context, statement_type: Optional[str] = None, ) -> str: """ 회사의 세부적인 재무 정보를 제공하는 도구. XBRL 파일을 파싱하여 상세한 재무 데이터를 추출합니다. Args: company_name: 회사명 (예: 삼성전자, 네이버 등) start_date: 시작일 (YYYYMMDD 형식, 예: 20230101) end_date: 종료일 (YYYYMMDD 형식, 예: 20231231) ctx: MCP Context 객체 statement_type: 재무제표 유형 ("재무상태표", "손익계산서", "현금흐름표" 중 하나 또는 None) None인 경우 모든 유형의 재무제표 정보를 반환합니다. Returns: 선택한 재무제표 유형(들)의 세부 항목 정보가 포함된 텍스트 """ # 결과 문자열 초기화 result = "" api_errors = [] try: # 재무제표 유형 검증 if statement_type is not None and statement_type not in STATEMENT_TYPES: return f"지원하지 않는 재무제표 유형입니다. 지원되는 유형: {', '.join(STATEMENT_TYPES.keys())}" # 모든 재무제표 유형을 처리할 경우 if statement_type is None: all_statement_types = list(STATEMENT_TYPES.keys()) ctx.info(f"{company_name}의 모든 재무제표(재무상태표, 손익계산서, 현금흐름표) 세부 정보를 검색합니다.") else: all_statement_types = [statement_type] ctx.info(f"{company_name}의 {statement_type} 세부 정보를 검색합니다.") # end_date 조정 original_end_date = end_date adjusted_end_date, was_adjusted = adjust_end_date(end_date) if was_adjusted: ctx.info(f"공시 제출 기간을 고려하여 검색 종료일을 {original_end_date}에서 {adjusted_end_date}로 자동 조정했습니다.") end_date = adjusted_end_date # 회사 코드 조회 corp_code, matched_name = await get_corp_code_by_name(company_name) if not corp_code: return f"회사 검색 오류: {matched_name}" ctx.info(f"{matched_name}(고유번호: {corp_code})의 공시를 검색합니다.") # 공시 목록 조회 disclosures, error_msg = await get_disclosure_list(corp_code, start_date, end_date) if error_msg: return error_msg if not disclosures: date_range_msg = f"{start_date}부터 {end_date}까지" if was_adjusted: date_range_msg += f" (원래 요청: {start_date}~{original_end_date}, 공시 제출 기간 고려하여 확장)" return f"{date_range_msg} '{matched_name}'(고유번호: {corp_code})의 정기공시가 없습니다." ctx.info(f"{len(disclosures)}개의 정기공시를 찾았습니다. XBRL 데이터 조회 및 분석을 시도합니다.") # 결과 문자열 초기화 result = f"# {matched_name}의 세부 재무 정보 ({start_date} ~ {end_date})\n\n" # 최대 5개의 공시만 처리 (API 호출 제한 및 시간 고려) disclosure_count = min(5, len(disclosures)) # 각 공시별로 XBRL 데이터 조회 및 저장 processed_disclosures = [] for disclosure in disclosures[:disclosure_count]: try: report_name = disclosure.get('report_nm', '제목 없음') rcept_dt = disclosure.get('rcept_dt', '날짜 없음') rcept_no = disclosure.get('rcept_no', '') # 보고서 코드 결정 reprt_code = determine_report_code(report_name) if not rcept_no or not reprt_code: continue ctx.info(f"공시 분석 중: {report_name} (접수번호: {rcept_no})") # XBRL 데이터 조회 xbrl_text = await get_financial_statement_xbrl(rcept_no, reprt_code) if not xbrl_text.startswith(("DART API 오류:", "API 요청 실패:", "ZIP 파일", "<인코딩 오류:")): processed_disclosures.append({ 'report_name': report_name, 'rcept_dt': rcept_dt, 'rcept_no': rcept_no, 'reprt_code': reprt_code, 'xbrl_text': xbrl_text }) else: error_summary = xbrl_text.split('\n')[0][:100] api_errors.append(f"{report_name}: {error_summary}") ctx.warning(f"XBRL 데이터 조회 오류 ({report_name}): {error_summary}") except Exception as e: api_errors.append(f"{report_name if 'report_name' in locals() else '알 수 없는 보고서'}: {str(e)}") ctx.error(f"공시 데이터 처리 중 예상치 못한 오류 발생: {e}") traceback.print_exc() # 각 재무제표 유형별 처리 for current_statement_type in all_statement_types: result += f"## {current_statement_type}\n\n" # 해당 재무제표 유형에 대한 태그 목록 조회 items_to_extract = DETAILED_TAGS[current_statement_type] # 재무제표 유형별 결과 저장 reports_with_data = 0 # 각 공시별 처리 for disclosure in processed_disclosures: try: report_name = disclosure['report_name'] rcept_dt = disclosure['rcept_dt'] rcept_no = disclosure['rcept_no'] xbrl_text = disclosure['xbrl_text'] # XBRL 파싱 및 데이터 추출 try: financial_data = parse_xbrl_financial_data(xbrl_text, items_to_extract) # 유효한 데이터가 있는지 확인 (최소 1개 항목 이상) valid_items_count = sum(1 for value in financial_data.values() if value not in INVALID_VALUE_INDICATORS and not value.startswith("오류(") and not value.startswith("분석 중")) if valid_items_count >= 1: reports_with_data += 1 # 데이터 결과에 추가 result += f"### {report_name} ({rcept_dt})\n" result += f"접수번호: {rcept_no}\n\n" # 테이블 형식으로 데이터 출력 result += "| 항목 | 값 |\n" result += "|------|------|\n" for item, value in financial_data.items(): if value not in INVALID_VALUE_INDICATORS and not value.startswith("오류(") and not value.startswith("분석 중"): result += f"| {item} | {value} |\n" else: # 매칭되지 않은 항목은 '-'로 표시 result += f"| {item} | - |\n" result += "\n" else: ctx.info(f"[{report_name}] {current_statement_type}의 유효한 데이터가 없습니다.") except Exception as e: ctx.warning(f"XBRL 파싱/분석 중 오류 발생 ({report_name}): {e}") api_errors.append(f"{report_name} 분석 중 오류: {str(e)}") except Exception as e: ctx.error(f"공시 데이터 처리 중 예상치 못한 오류 발생: {e}") api_errors.append(f"공시 데이터 처리 오류: {str(e)}") traceback.print_exc() # 재무제표 유형별 결과 요약 if reports_with_data == 0: result += f"조회된 공시에서 유효한 {current_statement_type} 데이터를 찾지 못했습니다.\n\n" result += "-" * 50 + "\n\n" # 최종 결과 메시지 추가 if api_errors: result += "\n## 처리 중 발생한 오류\n" for error in api_errors: result += f"- {error}\n" result += "\n" if len(disclosures) > disclosure_count: result += f"※ 총 {len(disclosures)}개의 정기공시 중 최신 {disclosure_count}개에 대해 분석을 시도했습니다.\n" if len(processed_disclosures) == 0: result += "※ 모든 공시에서 XBRL 데이터를 추출하는데 실패했습니다. 오류 메시지를 확인해주세요.\n" except Exception as e: return f"세부 재무 정보 검색 중 예상치 못한 오류가 발생했습니다: {str(e)}\n\n{traceback.format_exc()}" result += chat_guideline return result.strip()
• dart.py:76-120 (schema)

  Schema defining XBRL tag mappings for detailed financial items in balance sheet, income statement, and cash flow statement. Used by the tool's parsing logic.
  DETAILED_TAGS = { "재무상태표": { "유동자산": ["ifrs-full:CurrentAssets"], "비유동자산": ["ifrs-full:NoncurrentAssets"], "자산총계": ["ifrs-full:Assets"], "유동부채": ["ifrs-full:CurrentLiabilities"], "비유동부채": ["ifrs-full:NoncurrentLiabilities"], "부채총계": ["ifrs-full:Liabilities"], "자본금": ["ifrs-full:IssuedCapital"], "자본잉여금": ["ifrs-full:SharePremium"], "이익잉여금": ["ifrs-full:RetainedEarnings"], "기타자본항목": ["dart:ElementsOfOtherStockholdersEquity"], "자본총계": ["ifrs-full:Equity"] }, "손익계산서": { "매출액": ["ifrs-full:Revenue"], "매출원가": ["ifrs-full:CostOfSales"], "매출총이익": ["ifrs-full:GrossProfit"], "판매비와관리비": ["dart:TotalSellingGeneralAdministrativeExpenses"], "영업이익": ["dart:OperatingIncomeLoss"], "금융수익": ["ifrs-full:FinanceIncome"], "금융비용": ["ifrs-full:FinanceCosts"], "법인세비용차감전순이익": ["ifrs-full:ProfitLossBeforeTax"], "법인세비용": ["ifrs-full:IncomeTaxExpenseContinuingOperations"], "당기순이익": ["ifrs-full:ProfitLoss"], "기본주당이익": ["ifrs-full:BasicEarningsLossPerShare"] }, "현금흐름표": { "영업활동 현금흐름": ["ifrs-full:CashFlowsFromUsedInOperatingActivities"], "영업에서 창출된 현금": ["ifrs-full:CashFlowsFromUsedInOperations"], "이자수취": ["ifrs-full:InterestReceivedClassifiedAsOperatingActivities"], "이자지급": ["ifrs-full:InterestPaidClassifiedAsOperatingActivities"], "배당금수취": ["ifrs-full:DividendsReceivedClassifiedAsOperatingActivities"], "법인세납부": ["ifrs-full:IncomeTaxesPaidRefundClassifiedAsOperatingActivities"], "투자활동 현금흐름": ["ifrs-full:CashFlowsFromUsedInInvestingActivities"], "유형자산의 취득": ["ifrs-full:PurchaseOfPropertyPlantAndEquipmentClassifiedAsInvestingActivities"], "무형자산의 취득": ["ifrs-full:PurchaseOfIntangibleAssetsClassifiedAsInvestingActivities"], "유형자산의 처분": ["ifrs-full:ProceedsFromSalesOfPropertyPlantAndEquipmentClassifiedAsInvestingActivities"], "재무활동 현금흐름": ["ifrs-full:CashFlowsFromUsedInFinancingActivities"], "배당금지급": ["ifrs-full:DividendsPaidClassifiedAsFinancingActivities"], "현금및현금성자산의순증가": ["ifrs-full:IncreaseDecreaseInCashAndCashEquivalents"], "기초현금및현금성자산": ["dart:CashAndCashEquivalentsAtBeginningOfPeriodCf"], "기말현금및현금성자산": ["dart:CashAndCashEquivalentsAtEndOfPeriodCf"] } }
• dart.py:935-935 (registration)

  MCP tool registration decorator for search_detailed_financial_data.
  @mcp.tool()
• dart.py:411-502 (helper)

  Core helper function that parses XBRL content to extract financial data values using specified tags and context patterns.
  def parse_xbrl_financial_data(xbrl_content: str, items_and_tags: Dict[str, List[str]]) -> Dict[str, str]: """ XBRL 텍스트 내용을 파싱하여 지정된 항목의 재무 데이터를 추출 Args: xbrl_content: XBRL 파일의 전체 텍스트 내용 items_and_tags: 추출할 항목과 태그 리스트 딕셔너리 {'항목명': ['태그1', '태그2', ...]} Returns: 추출된 재무 데이터 딕셔너리 {'항목명': '값'} """ extracted_data = {item_name: "N/A" for item_name in items_and_tags} # 기본 네임스페이스 정의 base_namespaces = { 'ifrs-full': 'http://xbrl.ifrs.org/taxonomy/2021-03-24/ifrs-full', 'dart': 'http://dart.fss.or.kr/xbrl/dte/2019-10-31', 'kor-ifrs': 'http://www.fss.or.kr/xbrl/kor/kor-ifrs/2021-03-24', } try: # XBRL 파싱 root = ET.fromstring(xbrl_content) # 네임스페이스 추출 및 업데이트 namespaces, detected_namespaces = detect_namespaces(xbrl_content, base_namespaces) # 모든 contextRef 값 수집 all_context_refs = set() for elem in root.findall('.//*[@contextRef]'): all_context_refs.add(elem.get('contextRef')) # 회계연도 추출 fiscal_year = extract_fiscal_year(all_context_refs) # 각 항목별 태그 검색 및 값 추출 for item_name, tag_list in items_and_tags.items(): item_found = False for tag in tag_list: if item_found: break # 해당 태그 요소 검색 elements = root.findall(f'.//{tag}', namespaces) if not elements: continue # 항목 유형에 맞는 패턴 선택 patterns = get_pattern_by_item_type(item_name) # 각 보고서 유형별 패턴 시도 for report_type, pattern_code in patterns.items(): if item_found: break # 기존 접두사 로직은 참조용으로만 사용 (실제 패턴 매칭에는 사용하지 않음) # 패턴에서 접두사 부분을 (.): 어떤 한 글자라도 매칭되도록 함 pattern_base = f"CFY{fiscal_year}.{pattern_code}_ifrs-full_ConsolidatedAndSeparateFinancialStatementsAxis_ifrs-full_ConsolidatedMember" # 패턴의 끝에 $ 추가하여 정확히 일치하는 패턴만 매칭 pattern_regex = re.compile(f"^{pattern_base}$") # 패턴과 일치하는 요소 찾기 for elem in elements: context_ref = elem.get('contextRef') # 정규식으로 패턴 매칭 확인 (완전 일치) if context_ref and pattern_regex.match(context_ref): unit_ref = elem.get('unitRef') value_text = elem.text decimals = elem.get('decimals', '0') if value_text and unit_ref: try: formatted_value = format_numeric_value(value_text, decimals) extracted_data[item_name] = f"{formatted_value} ({report_type})" item_found = True break except (ValueError, TypeError) as e: pass if item_found: break except ET.ParseError as e: extracted_data = {key: "XBRL 파싱 오류" for key in items_and_tags} except Exception as e: traceback.print_exc() extracted_data = {key: "데이터 추출 오류" for key in items_and_tags} return extracted_data
• dart.py:243-307 (helper)

  Helper to fetch XBRL financial statement file from DART API given receipt number and report code.
  async def get_financial_statement_xbrl(rcept_no: str, reprt_code: str) -> str: """ 재무제표 원본파일(XBRL)을 다운로드하여 XBRL 텍스트를 반환하는 함수 Args: rcept_no: 공시 접수번호(14자리) reprt_code: 보고서 코드 (11011: 사업보고서, 11012: 반기보고서, 11013: 1분기보고서, 11014: 3분기보고서) Returns: 추출된 XBRL 텍스트 내용, 실패 시 오류 메시지 문자열 """ url = f"{BASE_URL}/fnlttXbrl.xml?crtfc_key={API_KEY}&rcept_no={rcept_no}&reprt_code={reprt_code}" try: async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return f"API 요청 실패: HTTP 상태 코드 {response.status_code}" try: with zipfile.ZipFile(BytesIO(response.content)) as zip_file: xbrl_content = "" for file_name in zip_file.namelist(): if file_name.lower().endswith('.xbrl'): with zip_file.open(file_name) as xbrl_file: # XBRL 파일을 텍스트로 읽기 (UTF-8 시도, 실패 시 EUC-KR) try: xbrl_content = xbrl_file.read().decode('utf-8') except UnicodeDecodeError: try: xbrl_file.seek(0) xbrl_content = xbrl_file.read().decode('euc-kr') except UnicodeDecodeError: xbrl_content = "<인코딩 오류: XBRL 내용을 읽을 수 없습니다>" break if not xbrl_content: return "ZIP 파일 내에서 XBRL 파일을 찾을 수 없습니다." return xbrl_content except zipfile.BadZipFile: # 응답이 ZIP 파일 형식이 아닐 경우 (DART API 오류 메시지 등) try: error_content = response.content.decode('utf-8') try: root = ET.fromstring(error_content) status = root.findtext('status') message = root.findtext('message') if status and message: return f"DART API 오류: {status} - {message}" else: return f"유효하지 않은 ZIP 파일이며, 오류 메시지 파싱 실패: {error_content[:200]}" except ET.ParseError: return f"유효하지 않은 ZIP 파일이며, XML 파싱 불가: {error_content[:200]}" except Exception: return "다운로드한 파일이 유효한 ZIP 파일이 아닙니다 (내용 확인 불가)." except Exception as e: return f"ZIP 파일 처리 중 오류 발생: {str(e)}" except httpx.RequestError as e: return f"API 요청 중 네트워크 오류 발생: {str(e)}" except Exception as e: return f"XBRL 데이터 처리 중 예상치 못한 오류 발생: {str(e)}"