classify_incident

Instructions

Implementation Reference

• server.py:354-453 (handler)

  The classify_incident function is the handler for the MCP tool. It classifies cyber incidents against NIS2 Article 23 thresholds, checking criteria like affected users, duration, cross-border impact, data breach, and financial loss to determine if an incident is 'significant', triggering 24h/72h/1-month reporting obligations. Registered via @mcp.tool() decorator on line 354.

  @mcp.tool()
  def classify_incident(
      incident_description: str,
      users_affected: int = 0,
      duration_hours: float = 0,
      cross_border: bool = False,
      data_breach: bool = False,
      financial_loss_eur: float = 0,
      api_key: str = "",
  ) -> str:
      """Classify a cyber incident against NIS2 Article 23 thresholds.
      Returns whether 'significant' — triggering 24h early warning, 72h incident notification, 1-month final report.
  
      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:
          incident_description (str): The incident description to analyze or process.
          users_affected (int): The users affected to analyze or process.
          duration_hours (float): The duration hours to analyze or process.
          cross_border (bool): The cross border to analyze or process.
          data_breach (bool): The data breach to analyze or process.
          financial_loss_eur (float): The financial loss 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})
  
      # Significant incident criteria (Article 23.3 + Commission Implementing Regulation for specific sectors)
      triggers = []
      if users_affected >= 100000:
          triggers.append(f"{users_affected:,} users/customers affected (≥100,000)")
      if duration_hours >= 1 and "critical" in incident_description.lower():
          triggers.append(f"Critical service unavailable {duration_hours}h (≥1h)")
      if duration_hours >= 8:
          triggers.append(f"Service disruption {duration_hours}h (≥8h)")
      if cross_border:
          triggers.append("Cross-border impact — notify neighbouring Member States via EU-CyCLONe")
      if data_breach:
          triggers.append("Data confidentiality/integrity breach — GDPR Article 33/34 obligations also trigger")
      if financial_loss_eur >= 500000:
          triggers.append(f"Direct financial impact €{financial_loss_eur:,.0f} (≥€500k)")
  
      is_significant = len(triggers) >= 1 and (data_breach or users_affected >= 100000 or cross_border or financial_loss_eur >= 500000 or duration_hours >= 8)
      now = datetime.now(timezone.utc)
  
      return json.dumps({
          "directive": "NIS2 Article 23 — Reporting obligations",
          "classification": "SIGNIFICANT_INCIDENT" if is_significant else "NON_SIGNIFICANT",
          "reporting_required": is_significant,
          "timeline": {
              "early_warning": "Within 24 hours of becoming aware — notify national CSIRT/competent authority" if is_significant else "Not required",
              "incident_notification": "Within 72 hours — update with initial assessment of severity, impact, indicators of compromise" if is_significant else "Not required",
              "final_report": "Within 1 month — detailed description, severity, root cause, mitigation, cross-border impact" if is_significant else "Not required",
              "deadlines_utc": {
                  "early_warning": (now + timedelta(hours=24)).isoformat() if is_significant else None,
                  "incident_notification": (now + timedelta(hours=72)).isoformat() if is_significant else None,
                  "final_report": (now + timedelta(days=30)).isoformat() if is_significant else None,
              } if is_significant else None,
          },
          "criteria_met": triggers,
          "parallel_obligations": [
              "GDPR Article 33 — 72h data breach notification to DPA (if personal data involved)",
              "GDPR Article 34 — without undue delay to data subjects (if high risk)",
              "DORA Article 19 — if financial entity, parallel DORA reporting to ESA",
              "Sector-specific — e.g. telecom authority under European Electronic Communications Code",
          ],
          "recipient_national_csirt_link": "https://www.enisa.europa.eu/topics/incident-response/csirts-in-europe",
      }, indent=2)

• server.py:354-354 (registration)

  The @mcp.tool() decorator on line 354 registers classify_incident as an MCP tool with the FastMCP server instance.

  @mcp.tool()

• server.py:61-73 (helper)

  The _check_rate_limit helper function enforces a free tier daily rate limit (10 calls/day) used by classify_incident before processing.

  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

• server.py:49-50 (helper)

  The check_access helper function validates API keys and determines the tier (free/pro) used by classify_incident for access control.

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