import logging
import sys
import threading
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any, AsyncIterator, Optional
from fastmcp import FastMCP
from mcp.types import TextContent
from .config import OpenDartConfig, MCPConfig
from .apis.client import OpenDartClient
from .apis import ds001, ds002, ds003, ds004, ds005, ds006
from mcp_opendart.registry.initialize_registry import initialize_registry
# 로거 설정
mcp_config = MCPConfig.from_env()
level_name = mcp_config.log_level.upper()
level = getattr(logging, level_name, logging.INFO)
logger = logging.getLogger("mcp-opendart")
# 로깅 설정: 출력 형식과 대상 스트림을 지정
logging.basicConfig(
level=level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr
)
# 전역 OpenDartContext 저장소 (fallback용)
_global_context: Optional['OpenDartContext'] = None
_context_lock = threading.Lock()
def set_global_context(ctx: 'OpenDartContext') -> None:
"""전역 OpenDartContext를 설정합니다."""
global _global_context
with _context_lock:
_global_context = ctx
logger.info("✅ Global OpenDartContext 저장됨")
def get_global_context() -> Optional['OpenDartContext']:
"""전역 OpenDartContext를 가져옵니다."""
global _global_context
with _context_lock:
return _global_context
@dataclass
class OpenDartContext:
"""Context for OpenDART MCP server operations."""
client: OpenDartClient
ds001: Any
ds002: Any
ds003: Any
ds004: Any
ds005: Any
ds006: Any
@asynccontextmanager
async def opendart_lifespan(app: FastMCP) -> AsyncIterator[OpenDartContext]:
"""Lifespan manager for the OpenDART FastMCP server.
Creates and manages the OpenDartClient instance and API modules.
"""
logger.info("Initializing OpenDART FastMCP server...")
# OpenDART 설정 로드
try:
opendart_config = OpenDartConfig.from_env()
mcp_config = MCPConfig.from_env()
logger.info(f"Server Name: {mcp_config.server_name}")
logger.info(f"Host: {mcp_config.host}")
logger.info(f"Port: {mcp_config.port}")
logger.info(f"Log Level: {mcp_config.log_level}")
# OpenDART API 클라이언트 초기화
client = OpenDartClient(config=opendart_config)
# API 모듈 초기화
ctx = OpenDartContext(
client=client,
ds001=ds001.DisclosureAPI(client),
ds002=ds002.PeriodicReportAPI(client),
ds003=ds003.FinancialInfoAPI(client),
ds004=ds004.OwnershipDisclosureAPI(client),
ds005=ds005.MajorReportAPI(client),
ds006=ds006.SecuritiesFilingAPI(client)
)
logger.info("OpenDART client and API modules initialized successfully.")
# 전역 컨텍스트로 저장 (fallback용)
set_global_context(ctx)
yield ctx
except Exception as e:
logger.error(f"Failed to initialize OpenDART client: {e}", exc_info=True)
raise
finally:
# 전역 컨텍스트 정리
global _global_context
with _context_lock:
_global_context = None
logger.info("Shutting down OpenDART FastMCP server...")
tool_registry = initialize_registry()
# Create the main FastMCP instance
mcp = FastMCP(
"OpenDART MCP",
instructions="OpenDART tools and resources for interacting with DART system",
lifespan=opendart_lifespan,
)
import importlib
for module_name in [
"disclosure_tools", "financial_info_tools", "major_report_tools",
"ownership_disclosure_tools", "periodic_report_tools", "securities_filing_tools"
]:
importlib.import_module(f"mcp_opendart.tools.{module_name}")
def main():
"""Main entry point for the OpenDART MCP server."""
logger.info("✅ Initializing OpenDART FastMCP server...")
mcp_config = MCPConfig.from_env()
if mcp_config.transport == "http":
logger.info(f"Starting server with HTTP transport on http://{mcp_config.host}:{mcp_config.port}")
mcp.run(transport="streamable-http", host=mcp_config.host, port=mcp_config.port)
else:
# stdio transport (default)
mcp.run()
if __name__ == "__main__":
main()
# Tool implementations will be added here
