unify anayasa tools

This commit is contained in:
saidsurucu
2025-07-17 20:09:50 +03:00
parent 856ecdf13d
commit f96c1a2e44
6 changed files with 271 additions and 152 deletions
+1
View File
@@ -212,3 +212,4 @@ analyze_playwright_mcp.py
measure_mcp_directly.py measure_mcp_directly.py
playwright_mcp_overhead.json playwright_mcp_overhead.json
simple_test.py simple_test.py
analyze_anayasa_html.py
+12 -12
View File
@@ -99,11 +99,11 @@ class AnayasaBireyselBasvuruApiClient:
for decision_div in decision_divs: for decision_div in decision_divs:
title_tag = decision_div.find("h4") title_tag = decision_div.find("h4")
title_text = title_tag.get_text(strip=True) if title_tag and title_tag.strong else (title_tag.get_text(strip=True) if title_tag else None) title_text = title_tag.get_text(strip=True) if title_tag and title_tag.strong else (title_tag.get_text(strip=True) if title_tag else "")
alti_cizili_div = decision_div.find("div", class_="AltiCizili") alti_cizili_div = decision_div.find("div", class_="AltiCizili")
ref_no, dec_type, body, app_date, dec_date, url_path = None, None, None, None, None, None ref_no, dec_type, body, app_date, dec_date, url_path = "", "", "", "", "", ""
if alti_cizili_div: if alti_cizili_div:
link_tag = alti_cizili_div.find("a", href=True) link_tag = alti_cizili_div.find("a", href=True)
if link_tag: if link_tag:
@@ -124,14 +124,14 @@ class AnayasaBireyselBasvuruApiClient:
ref_no = parts[current_idx] ref_no = parts[current_idx]
current_idx += 1 current_idx += 1
dec_type = parts[current_idx] if len(parts) > current_idx else None dec_type = parts[current_idx] if len(parts) > current_idx else ""
current_idx += 1 current_idx += 1
body = parts[current_idx] if len(parts) > current_idx else None body = parts[current_idx] if len(parts) > current_idx else ""
current_idx += 1 current_idx += 1
app_date_raw = parts[current_idx] if len(parts) > current_idx else None app_date_raw = parts[current_idx] if len(parts) > current_idx else ""
current_idx += 1 current_idx += 1
dec_date_raw = parts[current_idx] if len(parts) > current_idx else None dec_date_raw = parts[current_idx] if len(parts) > current_idx else ""
if app_date_raw and "Başvuru Tarihi :" in app_date_raw: if app_date_raw and "Başvuru Tarihi :" in app_date_raw:
app_date = app_date_raw.replace("Başvuru Tarihi :", "").strip() app_date = app_date_raw.replace("Başvuru Tarihi :", "").strip()
@@ -148,7 +148,7 @@ class AnayasaBireyselBasvuruApiClient:
subject_div = decision_div.find(lambda tag: tag.name == 'div' and not tag.has_attr('class') and tag.get_text(strip=True).startswith("BAŞVURU KONUSU :")) subject_div = decision_div.find(lambda tag: tag.name == 'div' and not tag.has_attr('class') and tag.get_text(strip=True).startswith("BAŞVURU KONUSU :"))
subject_text = subject_div.get_text(strip=True).replace("BAŞVURU KONUSU :", "").strip() if subject_div else None subject_text = subject_div.get_text(strip=True).replace("BAŞVURU KONUSU :", "").strip() if subject_div else ""
details_list: List[AnayasaBireyselReportDecisionDetail] = [] details_list: List[AnayasaBireyselReportDecisionDetail] = []
karar_detaylari_div = decision_div.find_next_sibling("div", id="KararDetaylari") # Corrected: was KararDetaylari karar_detaylari_div = decision_div.find_next_sibling("div", id="KararDetaylari") # Corrected: was KararDetaylari
@@ -159,13 +159,13 @@ class AnayasaBireyselBasvuruApiClient:
cells = row.find_all("td") cells = row.find_all("td")
if len(cells) == 4: # Hak, Müdahale İddiası, Sonuç, Giderim if len(cells) == 4: # Hak, Müdahale İddiası, Sonuç, Giderim
details_list.append(AnayasaBireyselReportDecisionDetail( details_list.append(AnayasaBireyselReportDecisionDetail(
hak=cells[0].get_text(strip=True) or None, hak=cells[0].get_text(strip=True) or "",
mudahale_iddiasi=cells[1].get_text(strip=True) or None, mudahale_iddiasi=cells[1].get_text(strip=True) or "",
sonuc=cells[2].get_text(strip=True) or None, sonuc=cells[2].get_text(strip=True) or "",
giderim=cells[3].get_text(strip=True) or None, giderim=cells[3].get_text(strip=True) or "",
)) ))
full_decision_page_url = urljoin(self.BASE_URL, url_path) if url_path else None full_decision_page_url = urljoin(self.BASE_URL, url_path) if url_path else ""
processed_decisions.append(AnayasaBireyselReportDecisionSummary( processed_decisions.append(AnayasaBireyselReportDecisionSummary(
title=title_text, title=title_text,
+5
View File
@@ -274,6 +274,11 @@ class AnayasaMahkemesiApiClient:
if not karar_metni_div: # Fallback if not in KararMetni if not karar_metni_div: # Fallback if not in KararMetni
karar_metni_div = soup.find("div", class_="WordSection1") karar_metni_div = soup.find("div", class_="WordSection1")
# Initialize with empty string defaults
decision_ek_no_from_page = ""
decision_date_from_page = ""
official_gazette_from_page = ""
if karar_metni_div: if karar_metni_div:
# Attempt to find E.K. No (Esas No, Karar No) # Attempt to find E.K. No (Esas No, Karar No)
# Norm Denetimi pages often have this in bold <p> tags directly or in the WordSection1 # Norm Denetimi pages often have this in bold <p> tags directly or in the WordSection1
+45
View File
@@ -4,6 +4,11 @@ from pydantic import BaseModel, Field, HttpUrl
from typing import List, Optional, Dict, Any from typing import List, Optional, Dict, Any
from enum import Enum from enum import Enum
# --- New Unified Enums ---
class AnayasaDecisionTypeEnum(str, Enum):
NORM_DENETIMI = "norm_denetimi"
BIREYSEL_BASVURU = "bireysel_basvuru"
# --- Enums (AnayasaDonemEnum, AnayasaBasvuruTuruEnum, etc. - same as before) --- # --- Enums (AnayasaDonemEnum, AnayasaBasvuruTuruEnum, etc. - same as before) ---
class AnayasaDonemEnum(str, Enum): class AnayasaDonemEnum(str, Enum):
TUMU = "ALL" TUMU = "ALL"
@@ -210,3 +215,43 @@ class AnayasaBireyselBasvuruDocumentMarkdown(BaseModel):
is_paginated: bool = Field(description="True if the full markdown content is split into multiple pages.") is_paginated: bool = Field(description="True if the full markdown content is split into multiple pages.")
# --- End Models for Bireysel Başvuru --- # --- End Models for Bireysel Başvuru ---
# --- Unified Models ---
class AnayasaUnifiedSearchRequest(BaseModel):
"""Unified search request for both Norm Denetimi and Bireysel Başvuru."""
decision_type: AnayasaDecisionTypeEnum = Field(..., description="Decision type: norm_denetimi or bireysel_basvuru")
# Common parameters
keywords: List[str] = Field(default_factory=list, description="Keywords to search for")
page_to_fetch: int = Field(1, ge=1, le=100, description="Page number to fetch (1-100)")
results_per_page: int = Field(10, ge=1, le=100, description="Results per page (1-100)")
# Norm Denetimi specific parameters (ignored for bireysel_basvuru)
keywords_all: List[str] = Field(default_factory=list, description="All keywords must be present (norm_denetimi only)")
keywords_any: List[str] = Field(default_factory=list, description="Any of these keywords (norm_denetimi only)")
decision_type_norm: Optional[AnayasaBasvuruTuruEnum] = Field(None, description="Decision type for norm denetimi")
application_date_start: str = Field("", description="Application start date (norm_denetimi only)")
application_date_end: str = Field("", description="Application end date (norm_denetimi only)")
# Bireysel Başvuru specific parameters (ignored for norm_denetimi)
decision_start_date: str = Field("", description="Decision start date (bireysel_basvuru only)")
decision_end_date: str = Field("", description="Decision end date (bireysel_basvuru only)")
norm_type: Optional[AnayasaNormTuruEnum] = Field(None, description="Norm type (bireysel_basvuru only)")
subject_category: str = Field("", description="Subject category (bireysel_basvuru only)")
class AnayasaUnifiedSearchResult(BaseModel):
"""Unified search result containing decisions from either system."""
decision_type: AnayasaDecisionTypeEnum = Field(..., description="Type of decisions returned")
decisions: List[Dict[str, Any]] = Field(default_factory=list, description="Decision list (structure varies by type)")
total_records_found: int = Field(0, description="Total number of records found")
retrieved_page_number: int = Field(1, description="Page number that was retrieved")
class AnayasaUnifiedDocumentMarkdown(BaseModel):
"""Unified document model for both Norm Denetimi and Bireysel Başvuru."""
decision_type: AnayasaDecisionTypeEnum = Field(..., description="Type of document")
source_url: HttpUrl = Field(..., description="Source URL of the document")
document_data: Dict[str, Any] = Field(default_factory=dict, description="Document content and metadata")
markdown_chunk: Optional[str] = Field(None, description="Markdown content chunk")
current_page: int = Field(1, description="Current page number")
total_pages: int = Field(1, description="Total number of pages")
is_paginated: bool = Field(False, description="Whether document is paginated")
+122
View File
@@ -0,0 +1,122 @@
# anayasa_mcp_module/unified_client.py
# Unified client for both Norm Denetimi and Bireysel Başvuru
import logging
from typing import Optional
from urllib.parse import urlparse
from .models import (
AnayasaUnifiedSearchRequest,
AnayasaUnifiedSearchResult,
AnayasaUnifiedDocumentMarkdown,
AnayasaDecisionTypeEnum,
AnayasaNormDenetimiSearchRequest,
AnayasaBireyselReportSearchRequest
)
from .client import AnayasaMahkemesiApiClient
from .bireysel_client import AnayasaBireyselBasvuruApiClient
logger = logging.getLogger(__name__)
class AnayasaUnifiedClient:
"""Unified client that handles both Norm Denetimi and Bireysel Başvuru searches."""
def __init__(self, request_timeout: float = 60.0):
self.norm_client = AnayasaMahkemesiApiClient(request_timeout)
self.bireysel_client = AnayasaBireyselBasvuruApiClient(request_timeout)
async def search_unified(self, params: AnayasaUnifiedSearchRequest) -> AnayasaUnifiedSearchResult:
"""Unified search that routes to appropriate client based on decision_type."""
if params.decision_type == AnayasaDecisionTypeEnum.NORM_DENETIMI:
# Convert to norm denetimi request
norm_params = AnayasaNormDenetimiSearchRequest(
keywords_all=params.keywords_all or params.keywords,
keywords_any=params.keywords_any,
application_type=params.decision_type_norm,
page_to_fetch=params.page_to_fetch,
results_per_page=params.results_per_page
)
result = await self.norm_client.search_norm_denetimi_decisions(norm_params)
# Convert to unified format
decisions_list = [decision.model_dump() for decision in result.decisions]
return AnayasaUnifiedSearchResult(
decision_type=AnayasaDecisionTypeEnum.NORM_DENETIMI,
decisions=decisions_list,
total_records_found=result.total_records_found,
retrieved_page_number=result.retrieved_page_number
)
elif params.decision_type == AnayasaDecisionTypeEnum.BIREYSEL_BASVURU:
# Convert to bireysel başvuru request
bireysel_params = AnayasaBireyselReportSearchRequest(
keywords=params.keywords,
decision_start_date=params.decision_start_date,
decision_end_date=params.decision_end_date,
norm_type=params.norm_type,
subject_category=params.subject_category,
page_to_fetch=params.page_to_fetch,
results_per_page=params.results_per_page
)
result = await self.bireysel_client.search_bireysel_basvuru_report(bireysel_params)
# Convert to unified format
decisions_list = [decision.model_dump() for decision in result.decisions]
return AnayasaUnifiedSearchResult(
decision_type=AnayasaDecisionTypeEnum.BIREYSEL_BASVURU,
decisions=decisions_list,
total_records_found=result.total_records_found,
retrieved_page_number=result.retrieved_page_number
)
else:
raise ValueError(f"Unsupported decision type: {params.decision_type}")
async def get_document_unified(self, document_url: str, page_number: int = 1) -> AnayasaUnifiedDocumentMarkdown:
"""Unified document retrieval that auto-detects the appropriate client."""
# Auto-detect decision type based on URL
parsed_url = urlparse(document_url)
if "normkararlarbilgibankasi" in parsed_url.netloc or "/ND/" in document_url:
# Norm Denetimi document
result = await self.norm_client.get_decision_document_as_markdown(document_url, page_number)
return AnayasaUnifiedDocumentMarkdown(
decision_type=AnayasaDecisionTypeEnum.NORM_DENETIMI,
source_url=result.source_url,
document_data=result.model_dump(),
markdown_chunk=result.markdown_chunk,
current_page=result.current_page,
total_pages=result.total_pages,
is_paginated=result.is_paginated
)
elif "kararlarbilgibankasi" in parsed_url.netloc or "/BB/" in document_url:
# Bireysel Başvuru document
result = await self.bireysel_client.get_decision_document_as_markdown(document_url, page_number)
return AnayasaUnifiedDocumentMarkdown(
decision_type=AnayasaDecisionTypeEnum.BIREYSEL_BASVURU,
source_url=result.source_url,
document_data=result.model_dump(),
markdown_chunk=result.markdown_chunk,
current_page=result.current_page,
total_pages=result.total_pages,
is_paginated=result.is_paginated
)
else:
raise ValueError(f"Cannot determine document type from URL: {document_url}")
async def close_client_session(self):
"""Close both client sessions."""
if hasattr(self.norm_client, 'close_client_session'):
await self.norm_client.close_client_session()
if hasattr(self.bireysel_client, 'close_client_session'):
await self.bireysel_client.close_client_session()
+82 -136
View File
@@ -262,6 +262,7 @@ from uyusmazlik_mcp_module.models import (
) )
from anayasa_mcp_module.client import AnayasaMahkemesiApiClient from anayasa_mcp_module.client import AnayasaMahkemesiApiClient
from anayasa_mcp_module.bireysel_client import AnayasaBireyselBasvuruApiClient from anayasa_mcp_module.bireysel_client import AnayasaBireyselBasvuruApiClient
from anayasa_mcp_module.unified_client import AnayasaUnifiedClient
from anayasa_mcp_module.models import ( from anayasa_mcp_module.models import (
AnayasaNormDenetimiSearchRequest, AnayasaNormDenetimiSearchRequest,
AnayasaSearchResult, AnayasaSearchResult,
@@ -269,6 +270,10 @@ from anayasa_mcp_module.models import (
AnayasaBireyselReportSearchRequest, AnayasaBireyselReportSearchRequest,
AnayasaBireyselReportSearchResult, AnayasaBireyselReportSearchResult,
AnayasaBireyselBasvuruDocumentMarkdown, AnayasaBireyselBasvuruDocumentMarkdown,
AnayasaUnifiedSearchRequest,
AnayasaUnifiedSearchResult,
AnayasaUnifiedDocumentMarkdown,
AnayasaDecisionTypeEnum,
AnayasaDonemEnum, AnayasaBasvuruTuruEnum, AnayasaVarYokEnum, AnayasaDonemEnum, AnayasaBasvuruTuruEnum, AnayasaVarYokEnum,
AnayasaNormTuruEnum, AnayasaIncelemeSonucuEnum, AnayasaSonucGerekcesiEnum AnayasaNormTuruEnum, AnayasaIncelemeSonucuEnum, AnayasaSonucGerekcesiEnum
) )
@@ -1045,6 +1050,7 @@ emsal_client_instance = EmsalApiClient()
uyusmazlik_client_instance = UyusmazlikApiClient() uyusmazlik_client_instance = UyusmazlikApiClient()
anayasa_norm_client_instance = AnayasaMahkemesiApiClient() anayasa_norm_client_instance = AnayasaMahkemesiApiClient()
anayasa_bireysel_client_instance = AnayasaBireyselBasvuruApiClient() anayasa_bireysel_client_instance = AnayasaBireyselBasvuruApiClient()
anayasa_unified_client_instance = AnayasaUnifiedClient()
kik_client_instance = KikApiClient() kik_client_instance = KikApiClient()
rekabet_client_instance = RekabetKurumuApiClient() rekabet_client_instance = RekabetKurumuApiClient()
bedesten_client_instance = BedestenApiClient() bedesten_client_instance = BedestenApiClient()
@@ -1449,7 +1455,10 @@ async def get_uyusmazlik_document_markdown_from_url(
logger.exception(f"Error in tool 'get_uyusmazlik_document_markdown_from_url'.") logger.exception(f"Error in tool 'get_uyusmazlik_document_markdown_from_url'.")
raise raise
# --- MCP Tools for Anayasa Mahkemesi (Norm Denetimi) --- # --- DEACTIVATED: MCP Tools for Anayasa Mahkemesi (Individual Tools) ---
# Use search_anayasa_unified and get_anayasa_document_unified instead
"""
@app.tool( @app.tool(
description="Search Constitutional Court norm control decisions with comprehensive filtering", description="Search Constitutional Court norm control decisions with comprehensive filtering",
annotations={ annotations={
@@ -1458,168 +1467,104 @@ async def get_uyusmazlik_document_markdown_from_url(
"idempotentHint": True "idempotentHint": True
} }
) )
async def search_anayasa_norm_denetimi_decisions( # DEACTIVATED TOOL - Use search_anayasa_unified instead
keywords_all: List[str] = Field(default_factory=list, description="AND"), # @app.tool(
keywords_any: List[str] = Field(default_factory=list, description="OR"), # description="DEACTIVATED - Use search_anayasa_unified instead",
keywords_exclude: List[str] = Field(default_factory=list, description="NOT"), # annotations={"readOnlyHint": True, "openWorldHint": False, "idempotentHint": True}
period: Literal["ALL", "1", "2"] = Field("ALL", description="Period"), # )
case_number_esas: str = Field("", description="Case number"), # async def search_anayasa_norm_denetimi_decisions(...) -> AnayasaSearchResult:
decision_number_karar: str = Field("", description="Decision number"), # raise ValueError("This tool is deactivated. Use search_anayasa_unified instead.")
first_review_date_start: str = Field("", description="Start"),
first_review_date_end: str = Field("", description="End"),
decision_date_start: str = Field("", description="Start"),
decision_date_end: str = Field("", description="End"),
application_type: Literal["ALL", "1", "2", "3"] = Field("ALL", description="App type"),
applicant_general_name: str = Field("", description="Applicant"),
applicant_specific_name: str = Field("", description="Name"),
official_gazette_date_start: str = Field("", description="Start"),
official_gazette_date_end: str = Field("", description="End"),
official_gazette_number_start: str = Field("", description="Start"),
official_gazette_number_end: str = Field("", description="End"),
has_press_release: Literal["ALL", "0", "1"] = Field("ALL", description="Press"),
has_dissenting_opinion: Literal["ALL", "0", "1"] = Field("ALL", description="Dissenting"),
has_different_reasoning: Literal["ALL", "0", "1"] = Field("ALL", description="Reason"),
attending_members_names: List[str] = Field(default_factory=list, description="Members"),
rapporteur_name: str = Field("", description="Rapporteur"),
norm_type: Literal["ALL", "1", "2", "14", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "0", "13"] = Field("ALL", description="Norm type"),
norm_id_or_name: str = Field("", description="Norm name"),
norm_article: str = Field("", description="Article"),
review_outcomes: List[Literal["ALL", "1", "2", "3", "4", "5", "6", "7", "8", "12"]] = Field(default_factory=list, description="Outcomes"),
reason_for_final_outcome: Literal["ALL", "29", "1", "2", "30", "3", "4", "27", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26"] = Field("ALL", description="Reason"),
basis_constitution_article_numbers: List[str] = Field(default_factory=list, description="Arts"),
results_per_page: int = Field(10, description="Count"),
page_to_fetch: int = Field(1, ge=1, description="Page"),
sort_by_criteria: str = Field("KararTarihi", description="Sort")
) -> AnayasaSearchResult:
"""Search Constitutional Court norm control decisions with comprehensive filtering."""
# Convert string literals to enums # DEACTIVATED TOOL - Use get_anayasa_document_unified instead
period_enum = AnayasaDonemEnum(period) # @app.tool(...)
application_type_enum = AnayasaBasvuruTuruEnum(application_type) # async def get_anayasa_norm_denetimi_document_markdown(...) -> AnayasaDocumentMarkdown:
has_press_release_enum = AnayasaVarYokEnum(has_press_release) # raise ValueError("This tool is deactivated. Use get_anayasa_document_unified instead.")
has_dissenting_opinion_enum = AnayasaVarYokEnum(has_dissenting_opinion)
has_different_reasoning_enum = AnayasaVarYokEnum(has_different_reasoning)
norm_type_enum = AnayasaNormTuruEnum(norm_type)
review_outcomes_enums = [AnayasaIncelemeSonucuEnum(ro) for ro in review_outcomes]
reason_for_final_outcome_enum = AnayasaSonucGerekcesiEnum(reason_for_final_outcome)
search_query = AnayasaNormDenetimiSearchRequest( # DEACTIVATED TOOL - Use search_anayasa_unified instead
keywords_all=keywords_all, # @app.tool(...)
keywords_any=keywords_any, # async def search_anayasa_bireysel_basvuru_report(...) -> AnayasaBireyselReportSearchResult:
keywords_exclude=keywords_exclude, # raise ValueError("This tool is deactivated. Use search_anayasa_unified instead.")
period=period_enum,
case_number_esas=case_number_esas,
decision_number_karar=decision_number_karar,
first_review_date_start=first_review_date_start,
first_review_date_end=first_review_date_end,
decision_date_start=decision_date_start,
decision_date_end=decision_date_end,
application_type=application_type_enum,
applicant_general_name=applicant_general_name,
applicant_specific_name=applicant_specific_name,
official_gazette_date_start=official_gazette_date_start,
official_gazette_date_end=official_gazette_date_end,
official_gazette_number_start=official_gazette_number_start,
official_gazette_number_end=official_gazette_number_end,
has_press_release=has_press_release_enum,
has_dissenting_opinion=has_dissenting_opinion_enum,
has_different_reasoning=has_different_reasoning_enum,
attending_members_names=attending_members_names,
rapporteur_name=rapporteur_name,
norm_type=norm_type_enum,
norm_id_or_name=norm_id_or_name,
norm_article=norm_article,
review_outcomes=review_outcomes_enums,
reason_for_final_outcome=reason_for_final_outcome_enum,
basis_constitution_article_numbers=basis_constitution_article_numbers,
results_per_page=results_per_page,
page_to_fetch=page_to_fetch,
sort_by_criteria=sort_by_criteria
)
logger.info(f"Tool 'search_anayasa_norm_denetimi_decisions' called.") # DEACTIVATED TOOL - Use get_anayasa_document_unified instead
try: # @app.tool(...)
return await anayasa_norm_client_instance.search_norm_denetimi_decisions(search_query) # async def get_anayasa_bireysel_basvuru_document_markdown(...) -> AnayasaBireyselBasvuruDocumentMarkdown:
except Exception as e: # raise ValueError("This tool is deactivated. Use get_anayasa_document_unified instead.")
logger.exception(f"Error in tool 'search_anayasa_norm_denetimi_decisions'.") """
raise
# --- Unified MCP Tools for Anayasa Mahkemesi ---
@app.tool( @app.tool(
description="Get Constitutional Court norm control decision text in paginated Markdown format", description="Unified search for Constitutional Court decisions: both norm control (normkararlarbilgibankasi) and individual applications (kararlarbilgibankasi) in one tool",
annotations={
"readOnlyHint": True,
"idempotentHint": True
}
)
async def get_anayasa_norm_denetimi_document_markdown(
document_url: str = Field(..., description="The URL path (e.g., /ND/YYYY/NN) or full https URL of the AYM Norm Denetimi decision from normkararlarbilgibankasi.anayasa.gov.tr."),
page_number: Optional[int] = Field(1, ge=1, description="Page number for paginated Markdown content (1-indexed, accepts int). Default is 1 (first 5,000 characters).")
) -> AnayasaDocumentMarkdown:
"""Get Constitutional Court norm control decision as paginated Markdown."""
logger.info(f"Tool 'get_anayasa_norm_denetimi_document_markdown' called for URL: {document_url}, Page: {page_number}")
if not document_url or not document_url.strip():
raise ValueError("Document URL is required for Anayasa Norm Denetimi document retrieval.")
current_page_to_fetch = page_number if page_number is not None and page_number >= 1 else 1
try:
return await anayasa_norm_client_instance.get_decision_document_as_markdown(document_url, page_number=current_page_to_fetch)
except Exception as e:
logger.exception(f"Error in tool 'get_anayasa_norm_denetimi_document_markdown'.")
raise
# --- MCP Tools for Anayasa Mahkemesi (Bireysel Başvuru Karar Raporu & Belgeler) ---
@app.tool(
description="Search Constitutional Court individual application decisions for human rights violations",
annotations={ annotations={
"readOnlyHint": True, "readOnlyHint": True,
"openWorldHint": True, "openWorldHint": True,
"idempotentHint": True "idempotentHint": True
} }
) )
async def search_anayasa_bireysel_basvuru_report( async def search_anayasa_unified(
keywords: List[str] = Field(default_factory=list, description="Keywords for AND logic."), decision_type: AnayasaDecisionTypeEnum = Field(..., description="Decision type: norm_denetimi (norm control) or bireysel_basvuru (individual applications)"),
page_to_fetch: int = Field(1, ge=1, description="Page number to fetch for the report. Default is 1.") keywords: List[str] = Field(default_factory=list, description="Keywords to search for (common parameter)"),
) -> AnayasaBireyselReportSearchResult: page_to_fetch: int = Field(1, ge=1, le=100, description="Page number to fetch (1-100)"),
"""Search Constitutional Court individual application decisions.""" results_per_page: int = Field(10, ge=1, le=100, description="Results per page (1-100)"),
search_query = AnayasaBireyselReportSearchRequest( # Norm Denetimi specific parameters (ignored for bireysel_basvuru)
keywords_all: List[str] = Field(default_factory=list, description="All keywords must be present (norm_denetimi only)"),
keywords_any: List[str] = Field(default_factory=list, description="Any of these keywords (norm_denetimi only)"),
decision_type_norm: Optional[AnayasaBasvuruTuruEnum] = Field(None, description="Decision type for norm denetimi"),
application_date_start: str = Field("", description="Application start date (norm_denetimi only)"),
application_date_end: str = Field("", description="Application end date (norm_denetimi only)"),
# Bireysel Başvuru specific parameters (ignored for norm_denetimi)
decision_start_date: str = Field("", description="Decision start date (bireysel_basvuru only)"),
decision_end_date: str = Field("", description="Decision end date (bireysel_basvuru only)"),
norm_type: Optional[AnayasaNormTuruEnum] = Field(None, description="Norm type (bireysel_basvuru only)"),
subject_category: str = Field("", description="Subject category (bireysel_basvuru only)")
) -> str:
logger.info(f"Tool 'search_anayasa_unified' called for decision_type: {decision_type}")
try:
request = AnayasaUnifiedSearchRequest(
decision_type=decision_type,
keywords=keywords, keywords=keywords,
page_to_fetch=page_to_fetch page_to_fetch=page_to_fetch,
results_per_page=results_per_page,
keywords_all=keywords_all,
keywords_any=keywords_any,
decision_type_norm=decision_type_norm,
application_date_start=application_date_start,
application_date_end=application_date_end,
decision_start_date=decision_start_date,
decision_end_date=decision_end_date,
norm_type=norm_type,
subject_category=subject_category
) )
logger.info(f"Tool 'search_anayasa_bireysel_basvuru_report' called.") result = await anayasa_unified_client_instance.search_unified(request)
try: return json.dumps(result.model_dump(), ensure_ascii=False, indent=2)
return await anayasa_bireysel_client_instance.search_bireysel_basvuru_report(search_query)
except Exception as e: except Exception as e:
logger.exception(f"Error in tool 'search_anayasa_bireysel_basvuru_report'.") logger.exception(f"Error in tool 'search_anayasa_unified'.")
raise raise
@app.tool( @app.tool(
description="Get Constitutional Court individual application decision text in paginated Markdown format", description="Unified document retrieval for Constitutional Court decisions: auto-detects norm control vs individual applications based on URL",
annotations={ annotations={
"readOnlyHint": True, "readOnlyHint": True,
"openWorldHint": False,
"idempotentHint": True "idempotentHint": True
} }
) )
async def get_anayasa_bireysel_basvuru_document_markdown( async def get_anayasa_document_unified(
document_url_path: str = Field(..., description="The URL path (e.g., /BB/YYYY/NNNN) of the AYM Bireysel Başvuru decision from kararlarbilgibankasi.anayasa.gov.tr."), document_url: str = Field(..., description="Document URL from search results"),
page_number: Union[int, str] = Field(1, description="Page number for paginated Markdown content (1-indexed, accepts int). Default is 1 (first 5,000 characters).") page_number: int = Field(1, ge=1, description="Page number for paginated content (1-indexed)")
) -> AnayasaBireyselBasvuruDocumentMarkdown: ) -> str:
"""Get Constitutional Court individual application decision as paginated Markdown.""" logger.info(f"Tool 'get_anayasa_document_unified' called for URL: {document_url}, Page: {page_number}")
logger.info(f"Tool 'get_anayasa_bireysel_basvuru_document_markdown' called for URL path: {document_url_path}, Page: {page_number}")
if not document_url_path or not document_url_path.strip() or not document_url_path.startswith("/BB/"):
raise ValueError("Document URL path (e.g., /BB/YYYY/NNNN) is required for Anayasa Bireysel Başvuru document retrieval.")
# Handle both int and string page_number inputs
try: try:
current_page_to_fetch = int(page_number) if page_number is not None else 1 result = await anayasa_unified_client_instance.get_document_unified(document_url, page_number)
if current_page_to_fetch < 1: return json.dumps(result.model_dump(mode='json'), ensure_ascii=False, indent=2)
current_page_to_fetch = 1
except (ValueError, TypeError):
current_page_to_fetch = 1
try:
return await anayasa_bireysel_client_instance.get_decision_document_as_markdown(document_url_path, page_number=current_page_to_fetch)
except Exception as e: except Exception as e:
logger.exception(f"Error in tool 'get_anayasa_bireysel_basvuru_document_markdown'.") logger.exception(f"Error in tool 'get_anayasa_document_unified'.")
raise raise
# --- MCP Tools for KIK (Kamu İhale Kurulu) --- # --- MCP Tools for KIK (Kamu İhale Kurulu) ---
@@ -2106,6 +2051,7 @@ def perform_cleanup():
globals().get('uyusmazlik_client_instance'), globals().get('uyusmazlik_client_instance'),
globals().get('anayasa_norm_client_instance'), globals().get('anayasa_norm_client_instance'),
globals().get('anayasa_bireysel_client_instance'), globals().get('anayasa_bireysel_client_instance'),
globals().get('anayasa_unified_client_instance'),
globals().get('kik_client_instance'), globals().get('kik_client_instance'),
globals().get('rekabet_client_instance'), globals().get('rekabet_client_instance'),
globals().get('bedesten_client_instance'), globals().get('bedesten_client_instance'),