classify_entity

Instructions

Implementation Reference

• server.py:130-130 (registration)

  Tool registration via @mcp.tool() decorator on classify_entity function.

  @mcp.tool()

• server.py:131-222 (handler)

  Main handler for classify_entity tool: checks access, rate limits, matches entity description against Annex I/II sectors, applies size threshold logic (Article 2), and returns JSON with classification (ESSENTIAL/IMPORTANT/OUT_OF_SCOPE), penalties, and next steps.

  def classify_entity(entity_description: str, employees: int = 0, turnover_million_eur: float = 0.0, api_key: str = "") -> str:
      """Classify an entity's NIS2 scope (essential / important / out-of-scope) + sector.
      Size-cap rules (Article 2): essential if in Annex I AND medium-size (>50 FTE or >€10M) — generally large (>250 FTE or >€50M). Important if Annex II + medium-size.
  
      Behavior:
          This tool is read-only and stateless — it produces analysis output
          without modifying any external systems, databases, or files.
          Safe to call repeatedly with identical inputs (idempotent).
          Free tier: 10/day rate limit. Pro tier: unlimited.
          No authentication required for basic usage.
  
      When to use:
          Use this tool when you need to assess, audit, or verify compliance
          requirements. Ideal for gap analysis, readiness checks, and generating
          compliance documentation.
  
      When NOT to use:
          Do not use as a substitute for qualified legal counsel. This tool
          provides technical compliance guidance, not legal advice.
  
      Args:
          entity_description (str): The entity description to analyze or process.
          employees (int): The employees to analyze or process.
          turnover_million_eur (float): The turnover million eur to analyze or process.
          api_key (str): The api key to analyze or process.
  
      Behavioral Transparency:
          - Side Effects: This tool is read-only and produces no side effects. It does not modify
            any external state, databases, or files. All output is computed in-memory and returned
            directly to the caller.
          - Authentication: No authentication required for basic usage. Pro/Enterprise tiers
            require a valid MEOK API key passed via the MEOK_API_KEY environment variable.
          - Rate Limits: Free tier: 10 calls/day. Pro tier: unlimited. Rate limit headers are
            included in responses (X-RateLimit-Remaining, X-RateLimit-Reset).
          - Error Handling: Returns structured error objects with 'error' key on failure.
            Never raises unhandled exceptions. Invalid inputs return descriptive validation errors.
          - Idempotency: Fully idempotent — calling with the same inputs always produces the
            same output. Safe to retry on timeout or transient failure.
          - Data Privacy: No input data is stored, logged, or transmitted to external services.
            All processing happens locally within the MCP server process.
      """
      allowed, msg, tier = check_access(api_key)
      if not allowed:
          return json.dumps({"error": msg, "upgrade_url": UPGRADE_STRIPE_49})
      if err := _check_rate_limit(tier=tier):
          return json.dumps({"error": err, "upgrade_url": UPGRADE_STRIPE_49})
  
      d = entity_description.lower()
      matched_annex_i = [k for k, v in ANNEX_I_ESSENTIAL_SECTORS.items() if any(t in d for t in k.split("_")) or any(t in d for t in v.lower().split(","))]
      matched_annex_ii = [k for k, v in ANNEX_II_IMPORTANT_SECTORS.items() if any(t in d for t in k.split("_")) or any(t in d for t in v.lower().split(","))]
  
      # Size thresholds (Article 2.1 / Article 2.2 + Commission Recommendation 2003/361/EC)
      # Medium = 50+ FTE OR €10M+ turnover. Large = 250+ FTE OR €50M+.
      is_large = employees >= 250 or turnover_million_eur >= 50
      is_medium = employees >= 50 or turnover_million_eur >= 10 or is_large
      is_micro_small = not is_medium and (employees > 0 or turnover_million_eur > 0)
  
      # Classification
      status = "OUT_OF_SCOPE"
      classification = None
      if matched_annex_i and is_medium:
          status = "IN_SCOPE"
          classification = "ESSENTIAL"
      elif matched_annex_ii and is_medium:
          status = "IN_SCOPE"
          classification = "IMPORTANT"
      elif (matched_annex_i or matched_annex_ii) and is_micro_small:
          status = "OUT_OF_SCOPE_BY_SIZE"
          classification = "Below size thresholds (Annex I needs medium+; Annex II needs medium+)"
      elif matched_annex_i or matched_annex_ii:
          status = "LIKELY_IN_SCOPE (provide employees/turnover to confirm)"
  
      days_since = (datetime.now(timezone.utc) - ENFORCEMENT_DATE).days
      penalty_headline = (
          "Up to €10M or 2% of global annual turnover" if classification == "ESSENTIAL"
          else "Up to €7M or 1.4% of global annual turnover" if classification == "IMPORTANT"
          else "Not applicable (out of scope)"
      )
  
      return json.dumps({
          "status": status,
          "classification": classification,
          "matched_annex_i_sectors": matched_annex_i,
          "matched_annex_ii_sectors": matched_annex_ii,
          "size": "large" if is_large else "medium" if is_medium else "small/micro" if is_micro_small else "unknown",
          "penalties_headline": penalty_headline,
          "days_since_eu_enforcement": days_since,
          "eu_enforcement_date": ENFORCEMENT_DATE.isoformat(),
          "national_transposition_note": "Each Member State transposes NIS2 into national law; exact deadlines and competent authorities vary. Check your national CSIRT and the ENISA NIS360 tracker.",
          "registration_required": status.startswith("IN_SCOPE") or status == "LIKELY_IN_SCOPE (provide employees/turnover to confirm)",
          "next_step": "Run audit_article_21 to score your 10 risk-management measures.",
      }, indent=2)

• server.py:79-101 (helper)

  Annex I and Annex II sector definitions used by classify_entity to match entity descriptions.

  ANNEX_I_ESSENTIAL_SECTORS = {
      "energy": "Energy — electricity, district heating/cooling, oil, gas, hydrogen",
      "transport": "Transport — air, rail, water, road",
      "banking": "Banking — credit institutions (overlaps DORA)",
      "financial_market_infra": "Financial market infrastructure — trading venues, CCPs",
      "health": "Health — healthcare providers, EU reference labs, R&D of medicinal products, manufacturing of medical devices critical during public health emergency",
      "drinking_water": "Drinking water — suppliers and distributors",
      "waste_water": "Waste water — collection, disposal, treatment",
      "digital_infrastructure": "Digital infrastructure — IXPs, DNS service providers, TLD name registries, cloud computing services, data centre services, content delivery networks, trust service providers, public electronic comms networks, public electronic comms services",
      "ict_service_management": "ICT service management (B2B) — MSPs, MSSPs",
      "public_administration": "Public administration — central government, regional authorities (as designated by Member States)",
      "space": "Space — operators of ground-based infrastructure supporting space services",
  }
  
  ANNEX_II_IMPORTANT_SECTORS = {
      "postal": "Postal and courier services",
      "waste_management": "Waste management",
      "chemicals": "Manufacture, production and distribution of chemicals",
      "food": "Production, processing and distribution of food",
      "manufacturing": "Manufacturing — medical devices & IVDs, computers/electronics/optical, electrical equipment, machinery, motor vehicles, other transport",
      "digital_providers": "Digital providers — online marketplaces, online search engines, social networking platforms",
      "research": "Research organisations",
  }

• server.py:49-50 (helper)

  check_access helper called by classify_entity to validate API key.

  def check_access(api_key: str = ""):
      return _shared_check_access(api_key)

• server.py:61-73 (helper)

  _check_rate_limit helper called by classify_entity to enforce free tier daily limit.

  def _check_rate_limit(caller: str = "anonymous", tier: str = "free") -> Optional[str]:
      if tier in ("pro", "professional", "enterprise"):
          return None
      now = datetime.now(timezone.utc)
      cutoff = now - timedelta(days=1)
      _usage[caller] = [t for t in _usage[caller] if t > cutoff]
      if len(_usage[caller]) >= FREE_DAILY_LIMIT:
          return (
              f"Free tier limit reached ({FREE_DAILY_LIMIT}/day). Unlock unlimited + "
              f"audit_all_obligations + signed certificates for £49/mo: {UPGRADE_STRIPE_49}"
          )
      _usage[caller].append(now)
      return None