Mevzuat MCP

Instructions

Implementation Reference

• mevzuat_mcp_server.py:869-952 (handler)

  The primary handler and registration for the 'search_cbgenelge' MCP tool. This FastMCP-decorated async function defines the tool's input schema via Pydantic Fields, sets mevzuat_tur='CB Genelgesi' for Presidential Circulars search, and delegates execution to MevzuatApiClientNew.search_documents for the actual API interaction and result processing.
  async def search_cbgenelge( aranacak_ifade: Optional[str] = Field( None, description='Search query with optional Boolean operators: simple word (organize), AND (organize AND suç), OR (suç OR ceza), NOT (organize NOT terör), + for required (+term), grouping with (), exact phrase with quotes ("organize suç"). Leave empty to list all circulars.' ), tam_cumle: bool = Field( False, description="Exact phrase match (true) or any word match (false, default). Set to true when searching for exact phrases." ), baslangic_tarihi: Optional[str] = Field( None, description="Start year for filtering results (format: YYYY, e.g., '2018')" ), bitis_tarihi: Optional[str] = Field( None, description="End year for filtering results (format: YYYY, e.g., '2024')" ), page_number: int = Field( 1, ge=1, description="Page number (1-indexed)" ), page_size: int = Field( 25, ge=1, le=100, description="Number of results per page (max 100)" ) ) -> MevzuatSearchResultNew: """ Search for Turkish Presidential Circulars (Cumhurbaşkanlığı Genelgesi / CB Genelgesi) in titles. This tool searches in circular titles (search is title-only by default for this document type). Use 'get_cbgenelge_content' to retrieve the full PDF content of a specific circular. Boolean operators (must be uppercase): - AND: Both terms must be present (organize AND suç) - OR: At least one term must be present (suç OR ceza) - NOT: Exclude term (organize NOT terör) - Exact phrase: Use quotes ("organize suç") Returns: - List of matching circulars with metadata (number, title, publication date, Official Gazette info) - Total result count and pagination info - Use the 'mevzuat_no' and 'resmi_gazete_tarihi' from results for content retrieval Note: Presidential Circulars are available as PDF files only. Example usage: - List all: search_cbgenelge() - Search: search_cbgenelge(aranacak_ifade="koordinasyon") - Filter by year: search_cbgenelge(baslangic_tarihi="2024", bitis_tarihi="2025") """ logger.info(f"Tool 'search_cbgenelge' called with query: {aranacak_ifade}") try: search_req = MevzuatSearchRequestNew( mevzuat_tur="CB Genelgesi", aranacak_ifade=aranacak_ifade or "", aranacak_yer=1, tam_cumle=tam_cumle, mevzuat_no=None, baslangic_tarihi=baslangic_tarihi, bitis_tarihi=bitis_tarihi, page_number=page_number, page_size=page_size ) result = await mevzuat_client.search_documents(search_req) logger.info(f"Search completed: {result.total_results} total results") return result except Exception as e: logger.exception("Error in tool 'search_cbgenelge'") return MevzuatSearchResultNew( documents=[], total_results=0, current_page=page_number, page_size=page_size, total_pages=0, query_used={"aranacak_ifade": aranacak_ifade}, error_message=f"An unexpected error occurred: {str(e)}" )
• mevzuat_models.py:107-117 (schema)

  Output schema (MevzuatSearchResultNew) returned by the search_cbgenelge tool.
  class MevzuatSearchResultNew(BaseModel): """Model for search results from mevzuat.gov.tr""" documents: List[MevzuatDocumentNew] total_results: int current_page: int page_size: int total_pages: int query_used: Dict[str, Any] error_message: Optional[str] = None
• mevzuat_models.py:32-85 (schema)

  Input request schema (MevzuatSearchRequestNew) constructed and passed to the client by the search_cbgenelge handler.
  class MevzuatSearchRequestNew(BaseModel): """Request model for searching legislation on mevzuat.gov.tr""" mevzuat_tur: MevzuatTurLiteral = Field( "Kanun", description="Type of legislation. Currently only 'Kanun' (laws) are fully supported for content extraction." ) aranacak_ifade: Optional[str] = Field( None, description="Search term or phrase to look for in legislation" ) aranacak_yer: int = Field( 3, ge=1, le=3, description="Where to search: 1=Title only, 2=Article titles, 3=Full text (default)" ) tam_cumle: bool = Field( False, description="Exact phrase match (true) or any word match (false, default)" ) mevzuat_no: Optional[str] = Field( None, description="Specific legislation number to search for" ) baslangic_tarihi: Optional[str] = Field( None, description="Start date for filtering (format: DD.MM.YYYY)" ) bitis_tarihi: Optional[str] = Field( None, description="End date for filtering (format: DD.MM.YYYY)" ) page_number: int = Field( 1, ge=1, description="Page number of results" ) page_size: int = Field( 10, ge=1, le=100, description="Number of results per page" )
• mevzuat_client.py:451-459 (helper)

  Thin wrapper helper method in MevzuatApiClientNew called by the tool handler to perform the search.
  async def search_documents(self, request: MevzuatSearchRequestNew) -> MevzuatSearchResultNew: """Search for legislation documents using httpx with Playwright cookies.""" # Get session/cookies with Playwright first await self._ensure_session() # If session establishment failed, fallback to full Playwright method if not self._antiforgery_token and not self._cookies: logger.warning("Session establishment failed, using full Playwright search method as fallback") return await self.search_documents_with_playwright(request)
• mevzuat_client.py:315-450 (helper)

  Core helper implementation for document search using browser automation (Playwright). This is the actual search logic invoked indirectly by the tool.
  async def search_documents_with_playwright(self, request: MevzuatSearchRequestNew) -> MevzuatSearchResultNew: """Search using Playwright - get cookies then make fetch request from page context.""" from playwright.async_api import async_playwright try: # Ensure browsers are installed self._ensure_playwright_browsers() logger.info(f"Searching with Playwright fetch: {request.aranacak_ifade or request.mevzuat_no}") async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() page = await context.new_page() # Visit main page to get cookies/session await page.goto(f"{self.BASE_URL}/", wait_until="domcontentloaded", timeout=30000) # Get antiforgery token from cookies cookies = await context.cookies() antiforgery_token = None for cookie in cookies: if 'Antiforgery' in cookie['name']: antiforgery_token = cookie['value'] break logger.info(f"Got antiforgery token: {antiforgery_token[:20] if antiforgery_token else 'None'}...") # Build payload # Normalize mevzuat type for API mevzuat_tur_api = self._normalize_mevzuat_tur_for_api(request.mevzuat_tur) payload = { "draw": 1, "columns": [ {"data": None, "name": "", "searchable": True, "orderable": False, "search": {"value": "", "regex": False}}, {"data": None, "name": "", "searchable": True, "orderable": False, "search": {"value": "", "regex": False}}, {"data": None, "name": "", "searchable": True, "orderable": False, "search": {"value": "", "regex": False}} ], "order": [], "start": (request.page_number - 1) * request.page_size, "length": request.page_size, "search": {"value": "", "regex": False}, "parameters": { "MevzuatTur": mevzuat_tur_api, "YonetmelikMevzuatTur": "OsmanliKanunu", # Required for all searches "AranacakIfade": request.aranacak_ifade or "", "TamCumle": "true" if request.tam_cumle else "false", "AranacakYer": str(request.aranacak_yer), "MevzuatNo": request.mevzuat_no or "", "KurumId": "0", "AltKurumId": "0", "BaslangicTarihi": request.baslangic_tarihi or "", "BitisTarihi": request.bitis_tarihi or "", "antiforgerytoken": antiforgery_token or "" } } # Make fetch request from within page context (has cookies) result = await page.evaluate(""" async (payload) => { const response = await fetch('https://www.mevzuat.gov.tr/Anasayfa/MevzuatDatatable', { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-Requested-With': 'XMLHttpRequest' }, body: JSON.stringify(payload) }); const text = await response.text(); if (!response.ok) { return { error: true, status: response.status, text: text }; } try { return JSON.parse(text); } catch(e) { return { error: true, parseError: e.message, text: text.substring(0, 200) }; } } """, payload) await browser.close() # Check for errors if result.get("error"): logger.error(f"Search API error: {result}") return MevzuatSearchResultNew( documents=[], total_results=0, current_page=request.page_number, page_size=request.page_size, total_pages=0, query_used=request.model_dump(), error_message=f"API error: {result.get('text', 'Unknown error')[:100]}" ) # Parse response total_results = result.get("recordsTotal", 0) documents = [] for item in result.get("data", []): doc = MevzuatDocumentNew( mevzuat_no=item.get("mevzuatNo", ""), mev_adi=item.get("mevAdi", ""), kabul_tarih=item.get("kabulTarih", ""), resmi_gazete_tarihi=item.get("resmiGazeteTarihi", ""), resmi_gazete_sayisi=item.get("resmiGazeteSayisi", ""), mevzuat_tertip=item.get("mevzuatTertip", ""), mevzuat_tur=item.get("tur", 1), url=item.get("url", "") ) documents.append(doc) logger.info(f"Found {total_results} results via Playwright") return MevzuatSearchResultNew( documents=documents, total_results=total_results, current_page=request.page_number, page_size=request.page_size, total_pages=(total_results + request.page_size - 1) // request.page_size if request.page_size > 0 else 0, query_used=request.model_dump() ) except Exception as e: logger.error(f"Playwright search failed: {e}") return MevzuatSearchResultNew( documents=[], total_results=0, current_page=request.page_number, page_size=request.page_size, total_pages=0, query_used=request.model_dump(), error_message=f"Playwright search error: {str(e)}" )