import logging, datetime
from typing import Any, Optional, Annotated
from pydantic import Field
from mcp_opendart.server import mcp
from mcp.types import TextContent
from mcp_opendart.utils.ctx_helper import with_context, with_context_async, as_json_text
from mcp_opendart.registry.initialize_registry import initialize_registry
logger = logging.getLogger("mcp-opendart")
tool_registry = initialize_registry()
@mcp.tool(
name="get_opendart_tool_info",
description="기업의 심층 분석을 위해 mcp 도구의 이름을 정확히 입력하여 도구의 목적, 활용 흐름, 필수 파라미터, 연관 도구 등을 확인 후 진행, 예: get_opendart_tool_info(tool_name='get_corporation_info')",
tags={"심층분석", "기업분석", "도구흐름"}
)
async def get_opendart_tool_info(
tool_name: str,
) -> TextContent:
tool = tool_registry.get_tool(tool_name)
if not tool:
return TextContent(type="text", text=f"❌ '{tool_name}'이라는 이름의 도구를 찾을 수 없습니다.")
# 파라미터 설명 추출
param_lines = []
props = tool.parameters.get("properties", {})
for param, meta in props.items():
desc = meta.get("description", "")
param_lines.append(f"- {param}: {desc}")
# linked 도구 리스트
linked = tool.linked_tools or []
linked_str = ", ".join(linked) if linked else "(연관 도구 없음)"
# 현재 날짜
today = datetime.date.today().strftime("%Y-%m-%d")
# 응답 생성
text = f"""
📌 {tool.name} ({tool.korean_name or '도구명 미정'})
이 도구는 특정 기업의 "{tool.name}"을(를) 다음과 같은 파라미터를 이용하여 조회할 수 있습니다:
{chr(10).join(param_lines)}
날짜가 필요한 경우, 반드시 시스템 기준일({today})을 기준으로 조회를 시작해야 합니다.
이 정보를 바탕으로, 사용자는 다음과 같은 흐름으로 도구를 활용할 수 있습니다:
1. 먼저 get_corporation_code_by_name 도구를 통해 기업명을 고유번호로 변환합니다.
2. {tool.name}을 사용해 관련 정보를 조회합니다.
3. 연관된 다음 도구를 반드시 확인해야 합니다:
{linked_str}
이 도구는 정보 흐름의 중심 축으로 활용되며, 다른 도구들과 결합되어 기업 리스크 분석에 핵심적인 역할을 합니다.
"""
return TextContent(type="text", text=text.strip())
@mcp.tool(
name="get_corporation_code_by_name",
description="기업명을 이용하여 기업 고유번호 조회, 공시조회를 위해 가장 먼저 실행하여 고유번호를 얻어야 함",
tags={"기업검색", "고유번호", "기업기초정보", "기업식별"}
)
async def get_corporation_code_by_name(
corp_name: Annotated[str, "기업명 (예: 삼성전자, 현대자동차)"],
) -> TextContent:
result = await with_context_async(None, "get_corporation_code_by_name", lambda context: context.ds001.get_corporation_code_by_name(corp_name))
return as_json_text(result)
@mcp.tool(
name="get_disclosure_list",
description="기업의 전체 공시 이력을 날짜별로 조회하여 경영활동, 재무현황, 지배구조 변화를 신속하게 파악. 여러 페이지가 있을 경우 페이지 번호를 바꿔 조회",
tags={"공시", "목록", "접수내역", "이벤트탐지"}
)
def get_disclosure_list(
corp_code: Annotated[str, "공시대상회사의 고유번호 (8자리)"],
bgn_de: Annotated[str, "조회 시작일 (YYYYMMDD)"],
end_de: Annotated[str, "조회 종료일 (YYYYMMDD)"],
page_no: Annotated[int, "페이지 번호 (기본값: 1)"] = 1,
pblntf_ty: Annotated[str, "공시유형 (A:정기공시, B:주요사항, C:발행, D:지분, E:기타 등)"] = "",
) -> TextContent:
"""
공시 목록 조회 - 기업의 전체 공시 이력을 날짜별로 조회하여 경영활동, 재무현황, 지배구조 변화를 신속하게 파악
응답에 total_count와 total_page 정보가 포함됩니다.
여러 페이지가 있을 경우 page_no 파라미터를 사용하여 다른 페이지를 조회하세요.
예: page_no=2, page_no=3 등으로 순차적으로 조회
Args:
corp_code (str): 공시대상회사의 고유번호 (8자리)
bgn_de (str): 조회 시작일 (YYYYMMDD)
end_de (str): 조회 종료일 (YYYYMMDD)
page_no (int): 페이지 번호 (기본값: 1) - total_page를 확인하여 필요한 페이지를 조회
pblntf_ty (str): 공시유형 필터 (기본값: "" - 모든 공시유형)
- "": 모든 공시유형 (기본값)
- "A": 정기공시 (사업보고서, 반기보고서, 분기보고서, 반기보고서 등)
- "B": 주요사항보고 (영업정지, 감자, 합병, 분할, 영업양수도 등)
- "C": 발행공시 (신주인수권부사채, 전환사채, 신주발행 등)
- "D": 지분공시 (대량보유상황보고서, 임원변경 등)
- "E": 기타공시 (기업설명회, 투자설명서 등)
- "F": 외부감사관련 (감사보고서, 회계법인 변경 등)
- "G": 펀드공시 (펀드 관련 공시)
- "H": 자산유동화 (자산유동화 관련 공시)
- "I": 거래소공시 (거래소 관련 공시)
- "J": 공정위공시 (공정거래위원회 관련 공시)
공시유형 선택 가이드:
- 사업보고서, 반기보고서, 분기보고서 조회 시: "A" 사용
- 경영진 변경, 사업 구조 변경 등 중요 사항: "B" 사용
- 신주발행, 사채발행 등 자금조달: "C" 사용
- 주주변동, 대주주 변경: "D" 사용
- IR 활동, 투자설명회: "E" 사용
참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019001
"""
result = with_context(None, "get_disclosure_list", lambda context: context.ds001.get_disclosure_list(
corp_code=corp_code,
bgn_de=bgn_de,
end_de=end_de,
page_no=page_no,
pblntf_ty=pblntf_ty
))
return as_json_text(result)
@mcp.tool(
name="get_corporation_info",
description="대표자, 결산월, 상장상태 등 기업 기본 정보 기반 지배구조 및 공시 일정 분석",
tags={"기업기초정보", "지배구조", "공시일정", "대표자분석", "가족경영"}
)
def get_corporation_info(
corp_code: Annotated[str, "공시대상회사의 고유번호 (8자리)"],
) -> TextContent:
"""
기업 개황정보 조회
Args:
corp_code (str): 공시대상회사의 고유번호(8자리)
참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019002
"""
result = with_context(None, "get_corporation_info", lambda context: context.ds001.get_corporation_info(corp_code))
return as_json_text(result)
@mcp.tool(
name="get_disclosure_document",
description="공시서류 원본파일을 다운로드하고 자동으로 재무제표 주석을 추출하여 캐싱합니다. 재무제표 주석이 없으면 다른 보고서를 찾아보라는 안내를 제공합니다.",
tags={"공시서류", "원본파일", "재무제표주석", "자동추출", "캐싱"}
)
def get_disclosure_document(
rcp_no: Annotated[str, "공시서류의 접수번호 (14자리)"],
) -> TextContent:
"""
공시서류 원본파일 다운로드 및 자동 재무제표 주석 추출
Args:
rcp_no (str): 공시서류의 접수번호 (14자리)
Returns:
TextContent: 공시서류 데이터 및 재무제표 주석 추출 결과
Features:
- XML 파일 다운로드 및 로컬 저장
- 자동 재무제표 주석 추출 (연결재무제표 주석, 재무제표 주석)
- 재무제표 주석이 없으면 다른 보고서를 찾아보라는 안내 제공
- 구조화된 데이터 캐싱 (disclosure_cache 디렉토리)
참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019003
"""
# 1. 공시서류 다운로드
result = with_context(None, "get_disclosure_document", lambda context: context.ds001.get_disclosure_document(rcp_no))
# 2. 재무제표 주석 자동 추출 (다운로드 성공 시)
if result.get("status_code") == "000" or result.get("status") == "000":
try:
financial_notes_result = with_context(None, "extract_financial_notes_document",
lambda context: context.ds001.extract_financial_notes_document(rcp_no))
if financial_notes_result.get("status") == "000":
result["financial_notes_extraction"] = {
"status": "success",
"message": "재무제표 주석이 성공적으로 추출되었습니다",
"summary": financial_notes_result.get("summary", {}),
"sections": financial_notes_result.get("sections", {}),
"save_path": financial_notes_result.get("save_path", ""),
"extraction_date": financial_notes_result.get("extraction_date", ""),
"final_rcp_no": financial_notes_result.get("rcp_no", rcp_no)
}
else:
result["financial_notes_extraction"] = {
"status": "error",
"message": financial_notes_result.get("message", "재무제표 주석 추출 실패"),
"error": financial_notes_result.get("error", "알 수 없는 오류")
}
except Exception as e:
result["financial_notes_extraction"] = {
"status": "error",
"message": "재무제표 주석 추출 중 오류 발생",
"error": str(e)
}
return as_json_text(result)
@mcp.tool(
name="search_financial_notes",
description="추출된 재무제표 주석, 사업의 내용, 회사의 개요 데이터에서 테이블과 문단을 검색합니다. 연결재무제표 주석, 재무제표 주석, 사업의 내용, 회사의 개요에서 특정 키워드를 검색할 수 있습니다.",
tags={"재무제표주석", "검색", "테이블검색", "문단검색", "키워드검색", "사업의내용", "회사의개요"}
)
def search_financial_notes(
rcp_no: Annotated[str, Field(description="공시 접수번호 (예: 20241231000420). get_disclosure_list에서 조회 가능")],
search_term: Annotated[str, Field(description="검색할 키워드 (예: CSM, 계약서비스마진, 보험계약자산)")],
section_type: Annotated[str, Field(description="검색 섹션 (all: 전체, consolidated_notes: 연결재무제표 주석, separate_notes: 재무제표 주석, business_content: 사업의 내용, company_overview: 회사의 개요)")] = "all",
search_in: Annotated[str, Field(description="검색 범위 (both: 테이블과 문단 모두, tables: 테이블만, paragraphs: 문단만)")] = "both",
case_sensitive: Annotated[bool, Field(description="대소문자 구분 여부 (기본값: False)")] = False,
max_results: Annotated[int, Field(description="최대 결과 수 (기본값: 50)")] = 50,
) -> TextContent:
"""
재무제표 주석 데이터 검색
Args:
rcp_no (str): 공시서류의 접수번호 (14자리)
search_term (str): 검색할 키워드
section_type (str): 검색할 섹션 ("all", "consolidated_notes", "separate_notes", "business_content", "company_overview")
search_in (str): 검색 대상 ("both", "tables", "paragraphs")
case_sensitive (bool): 대소문자 구분 여부 (기본값: False)
max_results (int): 최대 결과 수 (기본값: 50)
Returns:
TextContent: 검색 결과 (테이블 및 문단)
"""
from pathlib import Path
import json
import re
from difflib import SequenceMatcher
def is_similar_match(search_term: str, content: str) -> bool:
"""유사도 20% 이상 매칭 확인"""
# 정확한 매칭 먼저 확인
if case_sensitive:
if search_term in content:
return True
else:
if search_term.lower() in content.lower():
return True
# 유사도 매칭 확인 (20% 이상)
similarity = SequenceMatcher(None, search_term.lower(), content.lower()).ratio()
return similarity >= 0.2
try:
# 캐시 디렉토리 확인
project_root = Path(__file__).parent.parent.parent
cache_path = project_root / 'mcp_opendart' / 'utils' / 'data' / 'disclosure_cache' / f'financial_notes_{rcp_no}'
if not cache_path.exists():
return as_json_text({
"status": "404",
"message": "재무제표 주석 데이터를 찾을 수 없습니다. 먼저 get_disclosure_document를 실행하여 데이터를 추출하세요.",
"error": f"캐시 경로가 존재하지 않습니다: {cache_path}"
})
# 검색 결과 저장
search_results = {
"status": "000",
"message": "정상",
"rcp_no": rcp_no,
"search_term": search_term,
"section_type": section_type,
"search_in": search_in,
"results": {
"tables": [],
"paragraphs": []
}
}
# 섹션별 검색 (자동 확장 포함)
sections_to_search = []
if section_type == "all":
sections_to_search = ["consolidated_notes", "separate_notes", "business_content", "company_overview"]
elif section_type in ["consolidated_notes", "separate_notes", "business_content", "company_overview"]:
sections_to_search = [section_type]
# 검색 결과 풍부화를 위한 자동 확장 로직
enriched_sections = set(sections_to_search)
# 매칭된 섹션의 상위 디렉토리 자동 포함
for section in sections_to_search:
if section == "business_content":
# business_content의 모든 하위 섹션 자동 포함
enriched_sections.add("business_content")
elif section == "company_overview":
# company_overview의 모든 하위 섹션 자동 포함
enriched_sections.add("company_overview")
sections_to_search = list(enriched_sections)
for section_name in sections_to_search:
section_path = cache_path / section_name
if not section_path.exists():
continue
# 사업의 내용과 회사의 개요 섹션은 하위 구조가 다름
if section_name in ["business_content", "company_overview"]:
# 하위 섹션들 검색
subsections_path = section_path / 'subsections'
if subsections_path.exists():
for subsection_dir in subsections_path.iterdir():
if subsection_dir.is_dir():
# 하위 섹션 메타데이터 읽기
try:
with open(subsection_dir / 'metadata.json', 'r', encoding='utf-8') as f:
subsection_meta = json.load(f)
subsection_name = subsection_meta.get('original_name', subsection_dir.name)
except:
subsection_name = subsection_dir.name
# 🎯 섹션명 자체도 검색 대상에 포함
section_name_match = False
if case_sensitive:
section_name_match = search_term in subsection_name
else:
section_name_match = search_term.lower() in subsection_name.lower()
# 🎯 섹션명이 매칭되면 해당 섹션의 모든 내용을 포함
if section_name_match:
# 섹션명 매칭 시 모든 테이블과 문단을 결과에 포함
if search_in in ["tables", "both"]:
tables_path = subsection_dir / 'tables'
if tables_path.exists():
for table_file in tables_path.glob('*.json'):
try:
with open(table_file, 'r', encoding='utf-8') as f:
table_data = json.load(f)
search_text = ' '.join(table_data.get('headers', []) +
[cell for row in table_data.get('rows', []) for cell in row])
search_results["results"]["tables"].append({
"section": f"{section_name}/{subsection_name}",
"table_id": table_data.get('table_id'),
"file_path": str(table_file),
"headers": table_data.get('headers', []),
"row_count": len(table_data.get('rows', [])),
"matched_content": search_text,
"context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
"match_type": "section_name"
})
except Exception as e:
continue
if search_in in ["paragraphs", "both"]:
paragraphs_path = subsection_dir / 'paragraphs'
if paragraphs_path.exists():
for para_file in paragraphs_path.glob('*.json'):
try:
with open(para_file, 'r', encoding='utf-8') as f:
para_data = json.load(f)
content = para_data.get('content', '')
search_results["results"]["paragraphs"].append({
"section": f"{section_name}/{subsection_name}",
"para_id": para_data.get('para_id'),
"file_path": str(para_file),
"line_number": para_data.get('line_number'),
"content": content,
"context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
"match_type": "section_name"
})
except Exception as e:
continue
else:
# 섹션명이 매칭되지 않으면 내용에서만 검색
# 테이블 검색
if search_in in ["tables", "both"]:
tables_path = subsection_dir / 'tables'
if tables_path.exists():
for table_file in tables_path.glob('*.json'):
try:
with open(table_file, 'r', encoding='utf-8') as f:
table_data = json.load(f)
# 테이블 내용에서 검색
search_text = ' '.join(table_data.get('headers', []) +
[cell for row in table_data.get('rows', []) for cell in row])
# 내용 매칭 확인 (유사도 80% 포함)
content_match = is_similar_match(search_term, search_text)
if content_match:
search_results["results"]["tables"].append({
"section": f"{section_name}/{subsection_name}",
"table_id": table_data.get('table_id'),
"file_path": str(table_file),
"headers": table_data.get('headers', []),
"row_count": len(table_data.get('rows', [])),
"matched_content": search_text,
"context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
"match_type": "content"
})
except Exception as e:
continue
# 문단 검색
if search_in in ["paragraphs", "both"]:
paragraphs_path = subsection_dir / 'paragraphs'
if paragraphs_path.exists():
for para_file in paragraphs_path.glob('*.json'):
try:
with open(para_file, 'r', encoding='utf-8') as f:
para_data = json.load(f)
content = para_data.get('content', '')
# 내용 매칭 확인 (유사도 80% 포함)
content_match = is_similar_match(search_term, content)
if content_match:
search_results["results"]["paragraphs"].append({
"section": f"{section_name}/{subsection_name}",
"para_id": para_data.get('para_id'),
"file_path": str(para_file),
"line_number": para_data.get('line_number'),
"content": content,
"context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
"match_type": "content"
})
except Exception as e:
continue
else:
# 기존 재무제표 주석 섹션 검색
# 테이블 검색
if search_in in ["tables", "both"]:
tables_path = section_path / 'tables'
if tables_path.exists():
for table_file in tables_path.glob('*.json'):
try:
with open(table_file, 'r', encoding='utf-8') as f:
table_data = json.load(f)
# 테이블 내용에서 검색
search_text = ' '.join(table_data.get('headers', []) +
[cell for row in table_data.get('rows', []) for cell in row])
if is_similar_match(search_term, search_text):
search_results["results"]["tables"].append({
"section": section_name,
"table_id": table_data.get('table_id'),
"file_path": str(table_file),
"headers": table_data.get('headers', []),
"row_count": len(table_data.get('rows', [])),
"matched_content": search_text,
"context_type": "재무정보"
})
except Exception as e:
continue
# 문단 검색
if search_in in ["paragraphs", "both"]:
paragraphs_path = section_path / 'paragraphs'
if paragraphs_path.exists():
for para_file in paragraphs_path.glob('*.json'):
try:
with open(para_file, 'r', encoding='utf-8') as f:
para_data = json.load(f)
content = para_data.get('content', '')
if is_similar_match(search_term, content):
search_results["results"]["paragraphs"].append({
"section": section_name,
"para_id": para_data.get('para_id'),
"file_path": str(para_file),
"line_number": para_data.get('line_number'),
"content": content,
"context_type": "재무정보"
})
except Exception as e:
continue
# 결과 수 제한
search_results["results"]["tables"] = search_results["results"]["tables"][:max_results]
search_results["results"]["paragraphs"] = search_results["results"]["paragraphs"][:max_results]
# 컨텍스트별 그룹핑 정보 추가
context_groups = {}
for result in search_results["results"]["paragraphs"]:
context_type = result.get("context_type", "기타")
if context_type not in context_groups:
context_groups[context_type] = 0
context_groups[context_type] += 1
for result in search_results["results"]["tables"]:
context_type = result.get("context_type", "기타")
if context_type not in context_groups:
context_groups[context_type] = 0
context_groups[context_type] += 1
# 통계 추가
search_results["summary"] = {
"total_tables_found": len(search_results["results"]["tables"]),
"total_paragraphs_found": len(search_results["results"]["paragraphs"]),
"sections_searched": sections_to_search,
"context_groups": context_groups,
"enrichment_applied": True
}
return as_json_text(search_results)
except Exception as e:
return as_json_text({
"status": "500",
"message": "검색 중 오류 발생",
"error": str(e)
})
@mcp.tool(
name="get_corporation_code",
description="OpenDART에서 제공하는 모든 공시대상 회사의 고유번호 전체 목록(XML 파일)을 조회합니다. 기업명 검색 또는 고유번호 매핑에 사용됩니다.",
tags={"기업전체목록", "고유번호전체", "기업식별", "코드매핑"}
)
def get_corporation_code() -> TextContent:
"""
고유번호 목록 조회
Returns:
Dict[str, Any]: 고유번호 목록 (기업명, 고유번호, 종목코드 등 포함)
참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019018
"""
result = with_context(None, "get_corporation_code", lambda context: context.ds001.get_corporation_code())
return as_json_text(result)
