search_direct_procurement_authorities

Instructions

Implementation Reference

• ihale_client.py:165-207 (handler)

  Core implementation of the search_direct_procurement_authorities tool in EKAPClient class. Makes HTTP GET request to legacy EKAP endpoint (metot=idareAra), parses response, extracts token and name for each authority, handles errors.

  async def search_direct_procurement_authorities(
      self,
      search_term: str,
      cookies: Optional[Any] = None,
  ) -> Dict[str, Any]:
      """Search authorities for Direct Procurement filter (legacy idareAra).
  
      Returns list of { token, name } where token is the encrypted idareId (A) and name is D.
      """
      params = {
          "metot": "idareAra",
          "aranan": search_term or "",
          "ES": "",
          "ihaleidListesi": "",
      }
      try:
          data = await self._make_get_request_full_url(
              self.direct_procurement_url,
              params=params,
              cookies=cookies,
          )
          items = data.get("idareAramaResultList", [])
          results = []
          for it in items:
              results.append({
                  "token": it.get("A"),
                  "name": it.get("D"),
              })
          return {
              "authorities": results,
              "returned_count": len(results),
              "search_term": search_term,
          }
      except httpx.HTTPStatusError as e:
          return {
              "error": f"Authority search failed with status {e.response.status_code}",
              "message": str(e)
          }
      except Exception as e:
          return {
              "error": "Authority search failed",
              "message": str(e)
          }

• ihale_mcp.py:423-434 (registration)

  MCP tool registration via @mcp.tool decorator. Defines input schema via Annotated parameters and thin wrapper delegating to EKAPClient instance.

  @mcp.tool
  async def search_direct_procurement_authorities(
      search_term: Annotated[str, "Authority search term (İdare arama), e.g., 'antalya' or institution name"] = "",
      cookies: Annotated[Optional[str], "Cookie header (Çerez) for EKAP session (optional)"] = None,
  ) -> Dict[str, Any]:
      """
      Search authorities (İdare) for Direct Procurement (idareAra). Use returned 'token' as idareId.
      """
      return await ekap_client.search_direct_procurement_authorities(
          search_term=search_term,
          cookies=cookies,
      )

• ihale_client.py:87-132 (helper)

  Helper method used by the handler to make authenticated GET requests to legacy EKAP endpoints, handles cookies, SSL, session warmup for redirects, and error handling.

  async def _make_get_request_full_url(
      self,
      url: str,
      params: dict,
      headers: Optional[Dict[str, str]] = None,
      cookies: Optional[Any] = None,
  ) -> dict:
      """Make a GET request to a full URL (used for legacy EKAP endpoints).
  
      cookies: can be a cookie header string or a dict suitable for httpx.
      """
      ssl_context = self._create_ssl_context()
      req_headers = {
          'Accept': 'application/json, text/plain, */*',
          'Accept-Encoding': 'identity',
          'Connection': 'keep-alive',
          'Referer': 'https://ekap.kik.gov.tr/EKAP/YeniIhaleArama.aspx',
          'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
          'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
          'sec-ch-ua-mobile': '?0',
          'sec-ch-ua-platform': '"macOS"'
      }
      if headers:
          req_headers.update(headers)
      async with httpx.AsyncClient(
          timeout=30.0,
          verify=ssl_context,
          http2=False,
          limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
      ) as client:
          # If cookies is a string, set Cookie header; if dict, pass to client
          request_headers = dict(req_headers)
          httpx_cookies = None
          if isinstance(cookies, str) and cookies.strip():
              request_headers['Cookie'] = cookies
          elif isinstance(cookies, dict):
              httpx_cookies = cookies
          # First attempt
          response = await client.get(url, params=params, headers=request_headers, cookies=httpx_cookies, follow_redirects=False)
          # If redirected to error page, try warming up to obtain cookies and retry once
          if response.status_code == 302 and '/EKAP/error_page.html' in response.headers.get('location', '') and not cookies:
              await self._warmup_legacy_ekap(client)
              response = await client.get(url, params=params, headers=req_headers, follow_redirects=False)
          response.raise_for_status()
          return response.json()

• ihale_mcp.py:32-34 (helper)

  Initialization of the EKAPClient instance used by all EKAP-related tools including search_direct_procurement_authorities.

  ekap_client = EKAPClient()
  ilan_client = IlanClient()