diff --git a/emsal_mcp_module/client.py b/emsal_mcp_module/client.py index 1efd49d..bfc1de2 100644 --- a/emsal_mcp_module/client.py +++ b/emsal_mcp_module/client.py @@ -6,13 +6,15 @@ import httpx from typing import Dict, Any, List, Optional import logging import html +import os import re import io +import time from markitdown import MarkItDown from .models import ( EmsalSearchRequest, - EmsalDetailedSearchRequestData, + EmsalDetailedSearchRequestData, EmsalApiResponse, EmsalDocumentMarkdown ) @@ -21,12 +23,87 @@ logger = logging.getLogger(__name__) if not logger.hasHandlers(): logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') + +class EmsalRateLimited(Exception): + """Raised when the local rate-limit bucket would block longer than allowed. + + Carries the suggested retry-after (seconds) so callers can surface a + structured 429-style response instead of silently blocking the + event-loop slot for the full bucket-pause window. + """ + + def __init__(self, retry_after: float) -> None: + self.retry_after = retry_after + super().__init__(f"local bucket would block {retry_after:.1f}s") + + +class _TokenBucket: + """Asyncio token bucket with explicit back-pressure. + + The UYAP Emsal endpoint (emsal.uyap.gov.tr) rate-limits per source IP and + returns HTTP 429 (an HTML error page, no Retry-After header) after a small + burst of rapid requests. On the shared-egress-IP production deployment this + is hit constantly, making unrelated searches appear to "return 0 results" + depending only on request order. This bucket spaces requests to a safe rate + and freezes on an actual 429 via ``penalize_until``. + """ + + def __init__(self, capacity: int, refill_per_s: float) -> None: + self.capacity = float(capacity) + self.refill_per_s = float(refill_per_s) + self._tokens = float(capacity) + self._last = time.monotonic() + self._not_before = 0.0 + self._lock = asyncio.Lock() + + async def acquire(self, max_wait: Optional[float] = None) -> None: + """Acquire one token. If ``max_wait`` is set and the next wait would + exceed it, raise :class:`EmsalRateLimited` immediately instead of + sleeping — keeps a single rate-limited request from holding the + worker-slot for the full bucket-pause window.""" + deadline = (time.monotonic() + max_wait) if max_wait is not None else None + while True: + async with self._lock: + now = time.monotonic() + if now < self._not_before: + wait_s = self._not_before - now + else: + self._tokens = min( + self.capacity, + self._tokens + (now - self._last) * self.refill_per_s, + ) + self._last = now + if self._tokens >= 1.0: + self._tokens -= 1.0 + return + wait_s = (1.0 - self._tokens) / self.refill_per_s + if deadline is not None: + remaining = deadline - time.monotonic() + if wait_s > remaining: + raise EmsalRateLimited(retry_after=wait_s) + await asyncio.sleep(wait_s) + + def penalize_until(self, monotonic_deadline: float) -> None: + """Pause the bucket until ``monotonic_deadline`` (drains tokens).""" + self._not_before = max(self._not_before, monotonic_deadline) + self._tokens = 0.0 + self._last = time.monotonic() + class EmsalApiClient: """API Client for Emsal (UYAP Precedent Decision) search system.""" BASE_URL = "https://emsal.uyap.gov.tr" - DETAILED_SEARCH_ENDPOINT = "/aramadetaylist" + DETAILED_SEARCH_ENDPOINT = "/aramadetaylist" DOCUMENT_ENDPOINT = "/getDokuman" + # UYAP Emsal rate-limits per source IP. Defaults mirror the sibling + # Bedesten client (conservative: no burst, ~3.5s spacing). Override via env: + # EMSAL_RATE_CAPACITY (default 1) + # EMSAL_RATE_REFILL_S (default 3.5; seconds per token) + # EMSAL_RATE_MAX_WAIT_S (default 8.0; max local wait before a structured 429) + _DEFAULT_CAPACITY = int(os.getenv("EMSAL_RATE_CAPACITY", "1")) + _DEFAULT_REFILL_S = float(os.getenv("EMSAL_RATE_REFILL_S", "3.5")) + _DEFAULT_MAX_WAIT_S = float(os.getenv("EMSAL_RATE_MAX_WAIT_S", "8.0")) + def __init__(self, request_timeout: float = 30.0): self.http_client = httpx.AsyncClient( base_url=self.BASE_URL, @@ -38,6 +115,27 @@ class EmsalApiClient: timeout=request_timeout, verify=False # As per user's original FastAPI code ) + self._bucket = _TokenBucket( + capacity=self._DEFAULT_CAPACITY, + refill_per_s=1.0 / self._DEFAULT_REFILL_S, + ) + + def _handle_429(self, response: httpx.Response, op: str) -> None: + """Apply back-pressure to the shared bucket based on Retry-After. + + Emsal returns 429 as an HTML error page with no Retry-After header, so + the 30s fallback almost always applies.""" + retry_after_raw = response.headers.get("Retry-After", "") + try: + retry_after = float(retry_after_raw) + except (TypeError, ValueError): + retry_after = 30.0 + # Cap penalty so a hostile/buggy server can't freeze us indefinitely. + retry_after = max(1.0, min(retry_after, 60.0)) + self._bucket.penalize_until(time.monotonic() + retry_after + 0.5) + logger.warning( + f"EmsalApiClient: 429 on {op}; bucket paused {retry_after + 0.5:.1f}s" + ) async def search_detailed_decisions( self, @@ -76,7 +174,10 @@ class EmsalApiClient: async def _execute_api_search(self, endpoint: str, payload: Dict) -> EmsalApiResponse: """Helper method to execute search POST request and process response for Emsal.""" try: + await self._bucket.acquire(max_wait=self._DEFAULT_MAX_WAIT_S) response = await self.http_client.post(endpoint, json=payload) + if response.status_code == 429: + self._handle_429(response, "search") response.raise_for_status() response_json_data = response.json() logger.debug(f"EmsalApiClient: Raw API response from {endpoint}: {response_json_data}") @@ -143,9 +244,12 @@ class EmsalApiClient: logger.info(f"EmsalApiClient: Fetching Emsal document for Markdown (ID: {id}) from {source_url}") try: + await self._bucket.acquire(max_wait=self._DEFAULT_MAX_WAIT_S) response = await self.http_client.get(document_api_url) + if response.status_code == 429: + self._handle_429(response, f"document {id}") response.raise_for_status() - + # Emsal /getDokuman returns JSON with HTML in 'data' field (confirmed by user example) response_json = response.json() html_content_from_api = response_json.get("data")