❗ Exceptions¶

def __init__(
    self, message: str, error_code: Optional[str] = None, **kwargs: Any
) -> None:
    super().__init__(message)
    self.message = message
    self.error_code = error_code
    self.details = kwargs

def __init__(self, message: str, **kwargs: Any) -> None:
    super().__init__(message, error_code="VALIDATION_ERROR", **kwargs)

def __init__(self, agent_id: str, **kwargs: Any) -> None:
    message = f"Agent not gefatthe: {agent_id}"
    super().__init__(
        message, error_code="AGENT_NOT_FOUND", agent_id=agent_id, **kwargs
    )
    self.agent_id = agent_id

def __init__(
    self, message: str, status_code: Optional[int] = None, **kwargs: Any
) -> None:
    super().__init__(message, error_code="COMMUNICATION_ERROR", **kwargs)
    self.status_code = status_code

def __init__(self, message: str, **kwargs: Any) -> None:
    super().__init__(message, error_code="DISCOVERY_ERROR", **kwargs)

def __init__(
    self, attempts: int, last_error: Optional[Exception] = None, **kwargs: Any
) -> None:
    message = f"retry-Versuche erschöpft after {attempts} Versuchen"
    super().__init__(
        message, error_code="RETRY_EXHAUSTED", attempts=attempts, **kwargs
    )
    self.attempts = attempts
    self.last_error = last_error

def __init__(self, service_name: str, **kwargs: Any) -> None:
    message = f"circuit breaker is geopens for Service: {service_name}"
    super().__init__(
        message,
        error_code="CIRCUIT_BREAKER_OPEN",
        service_name=service_name,
        **kwargs,
    )
    self.service_name = service_name

def __init__(
    self, message: str, capability: Optional[str] = None, **kwargs: Any
) -> None:
    super().__init__(
        message, error_code="CAPABILITY_ERROR", capability=capability, **kwargs
    )
    self.capability = capability

def __init__(self, message: str, **kwargs: Any) -> None:
    super().__init__(message, error_code="TRACING_ERROR", **kwargs)

def __init__(
    self, message: str, config_key: Optional[str] = None, **kwargs: Any
) -> None:
    super().__init__(
        message, error_code="CONFIGURATION_ERROR", config_key=config_key, **kwargs
    )
    self.config_key = config_key

def __init__(self, message: str = "authentication failed", **kwargs: Any) -> None:
    super().__init__(message, error_code="AUTHENTICATION_ERROR", **kwargs)

def __init__(
    self, timeout_seconds: float, operation: Optional[str] = None, **kwargs: Any
) -> None:
    message = f"Timeout after {timeout_seconds}s"
    if operation:
        message += f" for operation: {operation}"
    super().__init__(
        message,
        error_code="TIMEOUT_ERROR",
        timeout_seconds=timeout_seconds,
        **kwargs,
    )
    self.timeout_seconds = timeout_seconds
    self.operation = operation

def __init__(
    self, message: str, protocol: Optional[str] = None, **kwargs: Any
) -> None:
    super().__init__(
        message, error_code="PROTOCOL_ERROR", protocol=protocol, **kwargs
    )
    self.protocol = protocol

def __init__(
    self, message: str, security_context: Optional[str] = None, **kwargs: Any
) -> None:
    super().__init__(
        message,
        error_code="SECURITY_ERROR",
        security_context=security_context,
        **kwargs,
    )
    self.security_context = security_context