main.py•17 kB
"""
LexLink MCP Server
MCP server for accessing Korean National Law Information Center API
"""
import os
import json
from typing import Optional
from pydantic import BaseModel, Field
from fastmcp import FastMCP
from mcp.server.fastmcp import Context
from smithery.decorators import smithery
from lexlink_client import LexLinkClient
class LexLinkConfig(BaseModel):
"""
Session configuration schema for LexLink MCP server.
Users provide their personal API key from the Korean Law API portal.
"""
openlaw_api_key: Optional[str] = Field(
default=None,
description="Your personal Korean Law API key (OC) from https://open.law.go.kr",
json_schema_extra={"secret": True}
)
def _get_client(ctx: Optional[Context] = None) -> LexLinkClient:
"""
Get LexLinkClient instance with API key from environment or context.
Defers API key validation to runtime to avoid import-time failures.
Args:
ctx: FastMCP context (may contain session config)
Returns:
LexLinkClient instance
Raises:
ValueError: If API key is not found
"""
# Environment variables supported:
# - OPENLAW_API_KEY (can be issued from https://open.law.go.kr)
api_key = os.getenv("OPENLAW_API_KEY")
# Try to get from session config if context is provided
if not api_key and ctx is not None:
session_cfg = getattr(ctx, 'session_config', None)
if session_cfg:
# Handle both dict and Pydantic model (Smithery may pass as dict)
if hasattr(session_cfg, 'openlaw_api_key'):
api_key = session_cfg.openlaw_api_key
elif isinstance(session_cfg, dict):
api_key = session_cfg.get('openlaw_api_key')
if not api_key:
raise ValueError(
"OpenLaw API key (OC) is required. "
"Provide via session config field 'openlaw_api_key' or env 'OPENLAW_API_KEY'."
)
return LexLinkClient(api_key)
@smithery.server(config_schema=LexLinkConfig)
def create_server() -> FastMCP:
"""
Create and return the FastMCP server instance.
Returns:
FastMCP server with all tools registered
"""
mcp = FastMCP("LexLink", dependencies=["httpx"])
@mcp.tool()
async def search_laws(
query: str,
display: int = 20,
page: int = 1,
ctx: Optional[Context] = None
) -> str:
"""
Search Korean laws and regulations (법령 검색).
Use this tool to find Korean laws by keyword. Returns a list of matching laws
with their names, serial numbers (MST), enactment dates, and governing ministries.
Args:
query: Search keyword in Korean or English (e.g., "민법", "형법", "근로기준법")
display: Number of results per page, maximum 100 (default: 20)
page: Page number for pagination (default: 1)
ctx: FastMCP context (optional)
Returns:
JSON string containing search results with law names, MST numbers, dates,
governing ministries, and links to full text
"""
try:
client = _get_client(ctx)
result = await client.search_laws(query, display, page)
# Extract key information
total_count = result.get('totalCnt', 0)
laws = result.get('law', [])
# Ensure laws is a list
if not isinstance(laws, list):
laws = [laws] if laws else []
# Format response
response = {
'total_count': total_count,
'page': page,
'results_count': len(laws),
'laws': []
}
for law in laws:
response['laws'].append({
'mst': law.get('법령일련번호', ''),
'name': law.get('법령명한글', ''),
'abbreviation': law.get('법령약칭명', ''),
'law_id': law.get('법령ID', ''),
'promulgation_date': law.get('공포일자', ''),
'enforcement_date': law.get('시행일자', ''),
'ministry': law.get('소관부처명', ''),
'law_type': law.get('법령구분명', ''),
'status': law.get('현행연혁코드', '')
})
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to search laws'
}, ensure_ascii=False)
@mcp.tool()
async def get_law_details(
mst: str,
effective_date: Optional[str] = None,
ctx: Optional[Context] = None
) -> str:
"""
Get full text and details of a specific Korean law (법령 상세 조회).
Use this tool to retrieve the complete text of a law including all articles,
provisions, and metadata. You need the MST number from search_laws results.
Args:
mst: Law serial number (법령일련번호/MST) from search results
effective_date: Optional effective date in YYYYMMDD format to get historical version
ctx: FastMCP context (optional)
Returns:
JSON string with complete law text including all articles and provisions.
Note: Response can be very large for comprehensive laws.
"""
try:
client = _get_client(ctx)
result = await client.get_law_details(mst, effective_date)
# Extract basic info
basic_info = result.get('기본정보', {})
articles = result.get('조문', {})
response = {
'basic_info': {
'law_id': basic_info.get('법령ID', ''),
'law_name': basic_info.get('법령명_한글', ''),
'promulgation_date': basic_info.get('공포일자', ''),
'enforcement_date': basic_info.get('시행일자', ''),
'ministry': basic_info.get('소관부처', ''),
'law_type': basic_info.get('법종구분', ''),
'amendment_type': basic_info.get('제개정구분', '')
},
'articles': []
}
# Parse articles
article_units = articles.get('조문단위', [])
if not isinstance(article_units, list):
article_units = [article_units] if article_units else []
for article in article_units[:50]: # Limit to first 50 articles for readability
article_data = {
'number': article.get('조문번호', ''),
'title': article.get('조문제목', ''),
'content': article.get('조문내용', ''),
'type': article.get('조문여부', '')
}
response['articles'].append(article_data)
if len(article_units) > 50:
response['note'] = f'Showing first 50 of {len(article_units)} articles'
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to get law details'
}, ensure_ascii=False)
@mcp.tool()
async def search_case_law(
query: str,
display: int = 20,
page: int = 1,
search_scope: int = 2,
court_type: Optional[str] = None,
date_range: Optional[str] = None,
ctx: Optional[Context] = None
) -> str:
"""
Search Korean court precedents and case law (판례 검색).
Use this tool to find court decisions by keyword. Returns case names, numbers,
courts, judgment dates, and case IDs for retrieving full text.
Args:
query: Search keyword (e.g., "손해배상", "계약", "불법행위")
display: Number of results per page, maximum 100 (default: 20)
page: Page number for pagination (default: 1)
search_scope: 1=case name only, 2=full text search (default: 2)
court_type: "400201" for Supreme Court, "400202" for lower courts (optional)
date_range: Date range in format "YYYYMMDD~YYYYMMDD" (optional)
ctx: FastMCP context (optional)
Returns:
JSON string with case list including case IDs, names, numbers, courts, and dates
"""
try:
client = _get_client(ctx)
result = await client.search_case_law(
query, search_scope, display, page, court_type, date_range
)
# Extract key information
total_count = result.get('totalCnt', 0)
cases = result.get('prec', [])
# Ensure cases is a list
if not isinstance(cases, list):
cases = [cases] if cases else []
response = {
'total_count': total_count,
'page': page,
'results_count': len(cases),
'cases': []
}
for case in cases:
response['cases'].append({
'case_id': case.get('판례일련번호', ''),
'case_name': case.get('사건명', ''),
'case_number': case.get('사건번호', ''),
'judgment_date': case.get('선고일자', ''),
'court': case.get('법원명', ''),
'case_type': case.get('사건종류명', ''),
'judgment_type': case.get('판결유형', ''),
'data_source': case.get('데이터출처명', '')
})
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to search case law'
}, ensure_ascii=False)
@mcp.tool()
async def get_case_details(
case_id: str,
ctx: Optional[Context] = None
) -> str:
"""
Get full text of a specific court case (판례 상세 조회).
Use this tool to retrieve the complete judgment text including holdings,
reasoning, and referenced statutes. You need the case ID from search_case_law results.
Args:
case_id: Case serial number (판례일련번호) from search results
ctx: FastMCP context (optional)
Returns:
JSON string with full judgment text, holdings, reasoning, and case metadata
"""
try:
client = _get_client(ctx)
result = await client.get_case_details(case_id)
# Note: The exact structure depends on the API response
# This is a simplified version - you may need to adjust based on actual response
response = {
'case_id': case_id,
'data': result
}
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to get case details'
}, ensure_ascii=False)
@mcp.tool()
async def search_legal_interpretations(
query: str,
display: int = 20,
page: int = 1,
search_scope: int = 1,
date_range: Optional[str] = None,
ctx: Optional[Context] = None
) -> str:
"""
Search official legal interpretations (법령해석례 검색).
Use this tool to find official interpretations of laws issued by the Ministry
of Government Legislation and other agencies.
Args:
query: Search keyword (e.g., "계약", "공무원", "세금")
display: Number of results per page, maximum 100 (default: 20)
page: Page number for pagination (default: 1)
search_scope: 1=title only, 2=full text (default: 1)
date_range: Date range in format "YYYYMMDD~YYYYMMDD" (optional)
ctx: FastMCP context (optional)
Returns:
JSON string with interpretation list including case numbers, agencies, and dates
"""
try:
client = _get_client(ctx)
result = await client.search_legal_interpretations(
query, display, page, search_scope, date_range
)
# Extract key information
total_count = result.get('totalCnt', 0)
interpretations = result.get('expc', [])
# Ensure interpretations is a list
if not isinstance(interpretations, list):
interpretations = [interpretations] if interpretations else []
response = {
'total_count': total_count,
'page': page,
'results_count': len(interpretations),
'interpretations': []
}
for interp in interpretations:
response['interpretations'].append({
'id': interp.get('법령해석례일련번호', ''),
'case_name': interp.get('안건명', ''),
'case_number': interp.get('안건번호', ''),
'inquiry_agency': interp.get('질의기관명', ''),
'response_agency': interp.get('회신기관명', ''),
'response_date': interp.get('회신일자', '')
})
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to search legal interpretations'
}, ensure_ascii=False)
@mcp.tool()
async def search_local_ordinances(
query: str,
display: int = 20,
page: int = 1,
ctx: Optional[Context] = None
) -> str:
"""
Search local government ordinances (자치법규 검색).
Use this tool to find ordinances and regulations issued by local governments
(provinces, cities, counties) in Korea.
Args:
query: Search keyword
display: Number of results per page, maximum 100 (default: 20)
page: Page number for pagination (default: 1)
ctx: FastMCP context (optional)
Returns:
JSON string with ordinance list
"""
try:
client = _get_client(ctx)
result = await client.search_local_ordinances(query, display, page)
# Extract key information
total_count = result.get('totalCnt', 0)
ordinances = result.get('ordin', [])
# Ensure ordinances is a list
if not isinstance(ordinances, list):
ordinances = [ordinances] if ordinances else []
response = {
'total_count': total_count,
'page': page,
'results_count': len(ordinances),
'ordinances': ordinances
}
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to search local ordinances'
}, ensure_ascii=False)
@mcp.tool()
async def get_legal_interpretation_details(
interp_id: str,
ctx: Optional[Context] = None
) -> str:
"""
Get full text of a specific legal interpretation (법령해석례 상세 조회).
Use this tool to retrieve the complete text of a legal interpretation including
the inquiry, response, and legal reasoning. You need the interpretation ID from
search_legal_interpretations results.
Args:
interp_id: Legal interpretation serial number (법령해석례일련번호) from search results
ctx: FastMCP context (optional)
Returns:
JSON string with full interpretation text, inquiry details, and response
"""
try:
client = _get_client(ctx)
result = await client.get_legal_interpretation_details(interp_id)
# Format the response
response = {
'interpretation_id': interp_id,
'data': result
}
return json.dumps(response, ensure_ascii=False, indent=2)
except Exception as e:
return json.dumps({
'error': str(e),
'message': 'Failed to get legal interpretation details'
}, ensure_ascii=False)
return mcp
if __name__ == "__main__":
# Run the MCP server (for local development)
server = create_server()
server.run()