search_okas_codes

Instructions

Implementation Reference

• ihale_mcp.py:205-223 (handler)

  MCP tool handler for 'search_okas_codes'. Includes registration via @mcp.tool decorator and input schema via Annotated type hints. Delegates to EKAPClient for execution.

  @mcp.tool
  async def search_okas_codes(
      search_term: Annotated[str, "Search term to find matching OKAS codes by description"] = "",
      kalem_turu: Annotated[Optional[Literal[1, 2, 3]], "Filter by item type: 1=Mal (Goods), 2=Hizmet (Service), 3=Yapım (Construction)"] = None,
      limit: Annotated[int, "Maximum number of results to return (1-500)"] = 50
  ) -> Dict[str, Any]:
      """
      Search OKAS procurement classification codes.
      
      Item types: 1=Goods, 2=Service, 3=Construction
      Search in Turkish descriptions for best results.
      """
      
      # Use the client to search OKAS codes
      return await ekap_client.search_okas_codes(
          search_term=search_term,
          kalem_turu=kalem_turu,
          limit=limit
      )

• ihale_mcp.py:207-210 (schema)

  Input schema defined by Annotated parameters in the tool handler function.

      search_term: Annotated[str, "Search term to find matching OKAS codes by description"] = "",
      kalem_turu: Annotated[Optional[Literal[1, 2, 3]], "Filter by item type: 1=Mal (Goods), 2=Hizmet (Service), 3=Yapım (Construction)"] = None,
      limit: Annotated[int, "Maximum number of results to return (1-500)"] = 50
  ) -> Dict[str, Any]:

• ihale_client.py:443-556 (helper)

  Core helper method in EKAPClient class that implements the OKAS code search logic: builds DevExtreme loadOptions payload, calls EKAP API endpoint /b_ihalearama/api/IhtiyacKalemleri/GetAll, applies client-side filtering for kalem_turu, formats hierarchical OKAS codes with descriptions in Turkish/English, and structures the response.

  async def search_okas_codes(
      self,
      search_term: str = "",
      kalem_turu: Optional[Literal[1, 2, 3]] = None,
      limit: int = 50
  ) -> Dict[str, Any]:
      """Search OKAS (public procurement classification) codes"""
      
      # Validate limit
      if limit > 500:
          limit = 500
      elif limit < 1:
          limit = 1
      
      # Build API request payload for OKAS search
      okas_params = {
          "loadOptions": {
              "filter": {
                  "sort": [],
                  "group": [],
                  "filter": [],
                  "totalSummary": [],
                  "groupSummary": [],
                  "select": [],
                  "preSelect": [],
                  "primaryKey": []
              }
          }
      }
      
      # Add search filters if provided
      filters = []
      
      if search_term:
          # Search in both Turkish and English descriptions
          filters.extend([
              ["kalemAdi", "contains", search_term],
              "or",
              ["kalemAdiEng", "contains", search_term]
          ])
      
      # Note: kalem_turu filtering causes 500 errors on the API
      # We'll filter client-side after getting results
      
      if filters:
          okas_params["loadOptions"]["filter"]["filter"] = filters
      
      # Set take limit for API
      okas_params["loadOptions"]["take"] = limit
      
      try:
          # Make API request to OKAS endpoint
          response_data = await self._make_request(self.okas_endpoint, okas_params)
          
          # Parse and format the response
          okas_items = response_data.get("loadResult", {}).get("data", [])
          
          # Format each OKAS code for better readability
          results = []
          for item in okas_items:
              kalem_turu_desc = {
                  1: "Mal (Goods)",
                  2: "Hizmet (Service)", 
                  3: "Yapım (Construction)"
              }.get(item.get("kalemTuru"), "Unknown")
              
              # Client-side filtering by kalem_turu since API filtering causes 500 errors
              if kalem_turu is not None and item.get("kalemTuru") != kalem_turu:
                  continue
              
              results.append({
                  "id": item.get("id"),
                  "code": item.get("kod"),
                  "description_tr": item.get("kalemAdi"),
                  "description_en": item.get("kalemAdiEng"),
                  "item_type": {
                      "code": item.get("kalemTuru"),
                      "description": kalem_turu_desc
                  },
                  "code_level": item.get("kodLevel"),
                  "parent_id": item.get("parentId"),
                  "has_items": item.get("hasItem", False),
                  "child_count": item.get("childCount", 0)
              })
          
          # Apply limit after client-side filtering
          if len(results) > limit:
              results = results[:limit]
          
          return {
              "okas_codes": results,
              "total_found": len(results),
              "search_params": {
                  "search_term": search_term,
                  "kalem_turu": kalem_turu,
                  "limit": limit
              },
              "item_type_legend": {
                  "1": "Mal (Goods)",
                  "2": "Hizmet (Service)",
                  "3": "Yapım (Construction)"
              }
          }
          
      except httpx.HTTPStatusError as e:
          return {
              "error": f"API request failed with status {e.response.status_code}",
              "message": str(e)
          }
      except Exception as e:
          return {
              "error": "Request failed",
              "message": str(e)
          }

• ihale_client.py:27-27 (helper)

  API endpoint constant used by search_okas_codes for fetching OKAS codes.

  self.okas_endpoint = "/b_ihalearama/api/IhtiyacKalemleri/GetAll"