dart-mcp

Instructions

Implementation Reference

• dart.py:742-932 (handler)

  The primary handler function for the 'search_disclosure' MCP tool. It handles company name to code lookup, retrieves disclosure list in date range, downloads and parses XBRL financial statements for up to 5 recent reports, extracts specified or default financial items, filters relevant reports, and returns formatted summary.
  @mcp.tool() async def search_disclosure( company_name: str, start_date: str, end_date: str, ctx: Context, requested_items: Optional[List[str]] = None, ) -> str: """ 회사의 주요 재무 정보를 검색하여 제공하는 도구. requested_items가 주어지면 해당 항목 관련 데이터가 있는 공시만 필터링합니다. Args: company_name: 회사명 (예: 삼성전자, 네이버 등) start_date: 시작일 (YYYYMMDD 형식, 예: 20230101) end_date: 종료일 (YYYYMMDD 형식, 예: 20231231) ctx: MCP Context 객체 requested_items: 사용자가 요청한 재무 항목 이름 리스트 (예: ["매출액", "영업이익"]). None이면 모든 주요 항목을 대상으로 함. 사용 가능한 항목: 매출액, 영업이익, 당기순이익, 영업활동 현금흐름, 투자활동 현금흐름, 재무활동 현금흐름, 자산총계, 부채총계, 자본총계 Returns: 검색된 각 공시의 주요 재무 정보 요약 텍스트 (요청 항목 관련 데이터가 있는 경우만) """ # 결과 문자열 초기화 result = "" try: # 진행 상황 알림 info_msg = f"{company_name}의" if requested_items: info_msg += f" {', '.join(requested_items)} 관련" info_msg += " 재무 정보를 검색합니다." ctx.info(info_msg) # 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 f"공시 목록 조회 오류: {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 데이터 조회 및 분석을 시도합니다.") # 추출할 재무 항목 및 가능한 태그 리스트 정의 all_items_and_tags = { "매출액": ["ifrs-full:Revenue"], "영업이익": ["dart:OperatingIncomeLoss"], "당기순이익": ["ifrs-full:ProfitLoss"], "영업활동 현금흐름": ["ifrs-full:CashFlowsFromUsedInOperatingActivities"], "투자활동 현금흐름": ["ifrs-full:CashFlowsFromUsedInInvestingActivities"], "재무활동 현금흐름": ["ifrs-full:CashFlowsFromUsedInFinancingActivities"], "자산총계": ["ifrs-full:Assets"], "부채총계": ["ifrs-full:Liabilities"], "자본총계": ["ifrs-full:Equity"] } # 사용자가 요청한 항목만 추출하도록 구성 if requested_items: items_to_extract = {item: tags for item, tags in all_items_and_tags.items() if item in requested_items} if not items_to_extract: unsupported_items = [item for item in requested_items if item not in all_items_and_tags] return f"요청하신 항목 중 지원되지 않는 항목이 있습니다: {', '.join(unsupported_items)}. 지원 항목: {', '.join(all_items_and_tags.keys())}" else: items_to_extract = all_items_and_tags # 결과 문자열 초기화 result = f"# {matched_name} 주요 재무 정보 ({start_date} ~ {end_date})\n" if requested_items: result += f"({', '.join(requested_items)} 관련)\n" result += "\n" # 최대 5개의 공시만 처리 (API 호출 제한 및 시간 고려) disclosure_count = min(5, len(disclosures)) processed_count = 0 relevant_reports_found = 0 api_errors = [] # 각 공시별 처리 for disclosure in disclosures[:disclosure_count]: 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 # 진행 상황 보고 processed_count += 1 await ctx.report_progress(processed_count, disclosure_count) ctx.info(f"공시 {processed_count}/{disclosure_count} 분석 중: {report_name} (접수번호: {rcept_no})") # XBRL 데이터 조회 try: xbrl_text = await get_financial_statement_xbrl(rcept_no, reprt_code) # XBRL 파싱 및 데이터 추출 financial_data = {} parse_error = None if not xbrl_text.startswith(("DART API 오류:", "API 요청 실패:", "ZIP 파일", "<인코딩 오류:")): try: financial_data = parse_xbrl_financial_data(xbrl_text, items_to_extract) except Exception as e: parse_error = e ctx.warning(f"XBRL 파싱/분석 중 오류 발생 ({report_name}): {e}") financial_data = {key: "분석 중 예외 발생" for key in items_to_extract} elif xbrl_text.startswith("DART API 오류: 013"): financial_data = {key: "데이터 없음(API 013)" for key in items_to_extract} else: error_summary = xbrl_text.split('\n')[0][:100] financial_data = {key: f"오류({error_summary})" for key in items_to_extract} api_errors.append(f"{report_name}: {error_summary}") # 요청된 항목 관련 데이터가 있는지 확인 is_relevant = True if requested_items: is_relevant = any( item in financial_data and financial_data[item] not in INVALID_VALUE_INDICATORS and not financial_data[item].startswith("오류(") and not financial_data[item].startswith("분석 중") for item in requested_items ) # 관련 데이터가 있는 공시만 결과에 추가 if is_relevant: relevant_reports_found += 1 result += f"## {report_name} ({rcept_dt})\n" result += f"접수번호: {rcept_no}\n\n" if financial_data: for item, value in financial_data.items(): result += f"- {item}: {value}\n" elif parse_error: result += f"- XBRL 분석 중 오류 발생: {parse_error}\n" else: result += "- 주요 재무 정보를 추출하지 못했습니다.\n" result += "\n" + "-" * 50 + "\n\n" else: ctx.info(f"[{report_name}] 건너뜀: 요청하신 항목({', '.join(requested_items) if requested_items else '전체'}) 관련 유효 데이터 없음.") except Exception as e: ctx.error(f"공시 처리 중 예상치 못한 오류 발생 ({report_name}): {e}") api_errors.append(f"{report_name}: {str(e)}") traceback.print_exc() # 최종 결과 메시지 추가 if api_errors: result += "\n## 처리 중 발생한 오류\n" for error in api_errors: result += f"- {error}\n" result += "\n" if relevant_reports_found == 0 and processed_count > 0: no_data_reason = "요청하신 항목 관련 유효한 데이터를 찾지 못했거나" if requested_items else "주요 재무 데이터를 찾지 못했거나" result += f"※ 처리된 공시에서 {no_data_reason}, 데이터가 제공되지 않는 보고서일 수 있습니다.\n" elif processed_count == 0 and disclosures: result += "조회된 정기공시가 있으나, XBRL 데이터를 포함하는 보고서 유형(사업/반기/분기)이 아니거나 처리 중 오류가 발생했습니다.\n" if len(disclosures) > disclosure_count: result += f"※ 총 {len(disclosures)}개의 정기공시 중 최신 {disclosure_count}개에 대해 분석을 시도했습니다.\n" if relevant_reports_found > 0 and requested_items: result += f"\n※ 요청하신 항목({', '.join(requested_items)}) 관련 정보가 있는 {relevant_reports_found}개의 보고서를 표시했습니다.\n" except Exception as e: return f"재무 정보 검색 중 예상치 못한 오류가 발생했습니다: {str(e)}\n\n{traceback.format_exc()}" result += chat_guideline return result.strip()
• dart.py:750-762 (schema)

  Input/output schema and description for the search_disclosure tool, defining parameters like company_name, dates, ctx, and optional requested_items with supported financial metrics.
  """ 회사의 주요 재무 정보를 검색하여 제공하는 도구. requested_items가 주어지면 해당 항목 관련 데이터가 있는 공시만 필터링합니다. Args: company_name: 회사명 (예: 삼성전자, 네이버 등) start_date: 시작일 (YYYYMMDD 형식, 예: 20230101) end_date: 종료일 (YYYYMMDD 형식, 예: 20231231) ctx: MCP Context 객체 requested_items: 사용자가 요청한 재무 항목 이름 리스트 (예: ["매출액", "영업이익"]). None이면 모든 주요 항목을 대상으로 함. 사용 가능한 항목: 매출액, 영업이익, 당기순이익, 영업활동 현금흐름, 투자활동 현금흐름, 재무활동 현금흐름, 자산총계, 부채총계, 자본총계 Returns: 검색된 각 공시의 주요 재무 정보 요약 텍스트 (요청 항목 관련 데이터가 있는 경우만)
• dart.py:742-742 (registration)

  MCP tool registration decorator that registers the search_disclosure function as a tool in the FastMCP server.
  @mcp.tool()
• dart.py:201-241 (helper)

  Helper function to retrieve the list of regular (semi-annual/quarterly/annual) disclosures for a company within a date range using DART API.
  async def get_disclosure_list(corp_code: str, start_date: str, end_date: str) -> Tuple[List[Dict[str, Any]], Optional[str]]: """ 기업의 정기공시 목록을 조회하는 함수 Args: corp_code: 회사 고유번호(8자리) start_date: 시작일(YYYYMMDD) end_date: 종료일(YYYYMMDD) Returns: (공시 목록 리스트, 오류 메시지) 튜플. 성공 시 (목록, None), 실패 시 (빈 리스트, 오류 메시지) """ # 정기공시(A) 유형만 조회 url = f"{BASE_URL}/list.json?crtfc_key={API_KEY}&corp_code={corp_code}&bgn_de={start_date}&end_de={end_date}&pblntf_ty=A&page_count=100" try: async with httpx.AsyncClient() as client: try: response = await client.get(url) if response.status_code != 200: return [], f"API 요청 실패: HTTP 상태 코드 {response.status_code}" try: result = response.json() if result.get('status') != '000': status = result.get('status', '알 수 없음') msg = result.get('message', '알 수 없는 오류') return [], f"DART API 오류: {status} - {msg}" return result.get('list', []), None except Exception as e: return [], f"응답 JSON 파싱 오류: {str(e)}" except httpx.RequestError as e: return [], f"API 요청 중 네트워크 오류 발생: {str(e)}" except Exception as e: return [], f"공시 목록 조회 중 예상치 못한 오류 발생: {str(e)}" return [], "알 수 없는 오류로 공시 목록을 조회할 수 없습니다."
• dart.py:411-503 (helper)

  Core helper for parsing XBRL financial statements, detecting namespaces, extracting fiscal year, matching context patterns for different statements (BS/IS/CF), and formatting numeric values.
  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