Add bounceban provider

llmstack/client/src/components/apps/ProviderIcon.jsx

@@ -31,6 +31,8 @@ import pineconeIcon_light from "../../assets/images/pinecone-icon-light.png";
 import pineconeIcon_dark from "../../assets/images/pinecone-icon-dark.png";
 import qdrantIcon_light from "../../assets/images/qdrant-icon-light.png";
 import qdrantIcon_dark from "../../assets/images/qdrant-icon-dark.png";
+import bouncbanIcon_light from "../../assets/images/bounceban-icon-light.png";
+import bouncbanIcon_dark from "../../assets/images/bounceban-icon-light.png";
 import weaviateIcon_light from "../../assets/images/weaviate-icon-light.png";
 import weaviateIcon_dark from "../../assets/images/weaviate-icon-dark.png";
 import singlestoreIcon_light from "../../assets/images/singlestore-icon-light.png";
@@ -74,6 +76,8 @@ export const getProviderIconImage = (icon, isActive) => {
       return isActive ? weaviateIcon_dark : weaviateIcon_light;
     case "singlestore":
       return isActive ? singlestoreIcon_dark : singlestoreIcon_light;
+    case "bounceban":
+      return isActive ? bouncbanIcon_dark : bouncbanIcon_light;
     default:
       return promptlyIcon_light;
   }

llmstack/processors/providers/bounceban/__init__.py

@@ -0,0 +1,13 @@
+from pydantic import Field
+
+from llmstack.processors.providers.config import ProviderConfig
+
+
+class BouncebanProviderConfig(ProviderConfig):
+    provider_slug: str = "bounceban"
+    api_key: str = Field(
+        title="API Key",
+        default="",
+        description="API Key for the BounceBan API",
+        json_schema_extra={"widget": "password", "advanced_parameter": False},
+    )

llmstack/processors/providers/bounceban/check.py

@@ -0,0 +1,97 @@
+import logging
+from typing import Any, Dict, Optional
+
+from asgiref.sync import async_to_sync
+from pydantic import Field
+
+from llmstack.apps.schemas import OutputTemplate
+from llmstack.common.utils.prequests import get
+from llmstack.processors.providers.api_processor_interface import (
+    ApiProcessorInterface,
+    ApiProcessorSchema,
+)
+from llmstack.processors.providers.metrics import MetricType
+
+logger = logging.getLogger(__name__)
+
+
+class CheckProcessorInput(ApiProcessorSchema):
+    email: Optional[str] = Field(description="The email to verify", default=None)
+    domain: Optional[str] = Field(description="The domain to verify", default=None)
+
+
+class CheckProcessorOutput(ApiProcessorSchema):
+    response: str = Field(description="The response from the API call as a string", default="")
+    response_json: Optional[Dict[str, Any]] = Field(
+        description="The response from the API call as a JSON object", default={}
+    )
+    response_objref: Optional[str] = Field(description="The reference to the response object", default=None)
+    headers: Optional[Dict[str, str]] = Field(description="The headers from the API call", default={})
+    code: int = Field(description="The status code from the API call", default=200)
+    size: int = Field(description="The size of the response from the API call", default=0)
+    time: float = Field(description="The time it took to get the response from the API call", default=0.0)
+
+
+class CheckProcessorConfiguration(ApiProcessorSchema):
+    pass
+
+
+class EchoProcessor(
+    ApiProcessorInterface[CheckProcessorInput, CheckProcessorOutput, CheckProcessorConfiguration],
+):
+    """
+    Check basic information for an email or domain.
+    """
+
+    @staticmethod
+    def name() -> str:
+        return "Check domain or email"
+
+    @staticmethod
+    def slug() -> str:
+        return "check"
+
+    @staticmethod
+    def description() -> str:
+        return "Check basic information for an email or domain."
+
+    @staticmethod
+    def provider_slug() -> str:
+        return "bounceban"
+
+    @classmethod
+    def get_output_template(cls) -> OutputTemplate | None:
+        return OutputTemplate(
+            markdown="{{response}}",
+            jsonpath="$.response",
+        )
+
+    def process(self) -> dict:
+        provider_config = self.get_provider_config(provider_slug=self.provider_slug(), processor_slug="*")
+        deployment_config = self.get_provider_config(provider_slug=self.provider_slug(), processor_slug="*")
+        api_key = deployment_config.api_key
+        response = get(
+            url="https://api.bounceban.com/v1/check",
+            headers={"Authorization": f"{api_key}"},
+            params=self._input.model_dump(),
+        )
+        self._usage_data.append(
+            (
+                f"{self.provider_slug()}/*/*/*",
+                MetricType.API_INVOCATION,
+                (provider_config.provider_config_source, 1),
+            )
+        )
+        async_to_sync(self._output_stream.write)(
+            CheckProcessorOutput(
+                response=response.text,
+                response_json=response.json(),
+                headers=response.headers,
+                code=response.status_code,
+                size=len(response.content),
+                time=response.elapsed.total_seconds(),
+            )
+        )
+
+        output = self._output_stream.finalize()
+        return output

llmstack/processors/providers/bounceban/verify.py

@@ -0,0 +1,101 @@
+import logging
+from typing import Any, Dict, Optional
+
+from asgiref.sync import async_to_sync
+from pydantic import Field
+
+from llmstack.apps.schemas import OutputTemplate
+from llmstack.common.utils.prequests import get
+from llmstack.processors.providers.api_processor_interface import (
+    ApiProcessorInterface,
+    ApiProcessorSchema,
+)
+from llmstack.processors.providers.metrics import MetricType
+
+logger = logging.getLogger(__name__)
+
+
+class VerifyProcessorInput(ApiProcessorSchema):
+    email: str = Field(description="The email to verify")
+    mode: Optional[str] = Field(description="The mode to use for verification", default="regular")
+    url: Optional[str] = Field(
+        description="A webhook target URL specified to receive verification result event in real-time through an HTTP POST request.",
+        default=None,
+    )
+
+
+class VerifyProcessorOutput(ApiProcessorSchema):
+    response: str = Field(description="The response from the API call as a string", default="")
+    response_json: Optional[Dict[str, Any]] = Field(
+        description="The response from the API call as a JSON object", default={}
+    )
+    response_objref: Optional[str] = Field(description="The reference to the response object", default=None)
+    headers: Optional[Dict[str, str]] = Field(description="The headers from the API call", default={})
+    code: int = Field(description="The status code from the API call", default=200)
+    size: int = Field(description="The size of the response from the API call", default=0)
+    time: float = Field(description="The time it took to get the response from the API call", default=0.0)
+
+
+class VerifyProcessorConfiguration(ApiProcessorSchema):
+    pass
+
+
+class EchoProcessor(
+    ApiProcessorInterface[VerifyProcessorInput, VerifyProcessorOutput, VerifyProcessorConfiguration],
+):
+    """
+    Single email verification processor
+    """
+
+    @staticmethod
+    def name() -> str:
+        return "Email Verification"
+
+    @staticmethod
+    def slug() -> str:
+        return "verify"
+
+    @staticmethod
+    def description() -> str:
+        return "Submit email for verification."
+
+    @staticmethod
+    def provider_slug() -> str:
+        return "bounceban"
+
+    @classmethod
+    def get_output_template(cls) -> OutputTemplate | None:
+        return OutputTemplate(
+            markdown="{{response}}",
+            jsonpath="$.response",
+        )
+
+    def process(self) -> dict:
+        provider_config = self.get_provider_config(provider_slug=self.provider_slug(), processor_slug="*")
+        deployment_config = self.get_provider_config(provider_slug=self.provider_slug(), processor_slug="*")
+        api_key = deployment_config.api_key
+        response = get(
+            url="https://api.bounceban.com/v1/verify/single",
+            headers={"Authorization": f"{api_key}"},
+            params=self._input.model_dump(),
+        )
+        self._usage_data.append(
+            (
+                f"{self.provider_slug()}/*/*/*",
+                MetricType.API_INVOCATION,
+                (provider_config.provider_config_source, 1),
+            )
+        )
+        async_to_sync(self._output_stream.write)(
+            VerifyProcessorOutput(
+                response=response.text,
+                response_json=response.json(),
+                headers=response.headers,
+                code=response.status_code,
+                size=len(response.content),
+                time=response.elapsed.total_seconds(),
+            )
+        )
+
+        output = self._output_stream.finalize()
+        return output

llmstack/processors/providers/bounceban/verify_status.py

@@ -0,0 +1,96 @@
+import logging
+from typing import Any, Dict, Optional
+
+from asgiref.sync import async_to_sync
+from pydantic import Field
+
+from llmstack.apps.schemas import OutputTemplate
+from llmstack.common.utils.prequests import get
+from llmstack.processors.providers.api_processor_interface import (
+    ApiProcessorInterface,
+    ApiProcessorSchema,
+)
+from llmstack.processors.providers.metrics import MetricType
+
+logger = logging.getLogger(__name__)
+
+
+class VerifyStatusProcessorInput(ApiProcessorSchema):
+    id: str = Field(description="The unique ID for a single email verification task.")
+
+
+class VerifyStatusProcessorOutput(ApiProcessorSchema):
+    response: str = Field(description="The response from the API call as a string", default="")
+    response_json: Optional[Dict[str, Any]] = Field(
+        description="The response from the API call as a JSON object", default={}
+    )
+    response_objref: Optional[str] = Field(description="The reference to the response object", default=None)
+    headers: Optional[Dict[str, str]] = Field(description="The headers from the API call", default={})
+    code: int = Field(description="The status code from the API call", default=200)
+    size: int = Field(description="The size of the response from the API call", default=0)
+    time: float = Field(description="The time it took to get the response from the API call", default=0.0)
+
+
+class VerifyStatusProcessorConfiguration(ApiProcessorSchema):
+    pass
+
+
+class EchoProcessor(
+    ApiProcessorInterface[VerifyStatusProcessorInput, VerifyStatusProcessorOutput, VerifyStatusProcessorConfiguration],
+):
+    """
+    Single email verification result processor
+    """
+
+    @staticmethod
+    def name() -> str:
+        return "Email Verification Status"
+
+    @staticmethod
+    def slug() -> str:
+        return "verify_status"
+
+    @staticmethod
+    def description() -> str:
+        return "Get the verification result for single email verification."
+
+    @staticmethod
+    def provider_slug() -> str:
+        return "bounceban"
+
+    @classmethod
+    def get_output_template(cls) -> OutputTemplate | None:
+        return OutputTemplate(
+            markdown="{{response}}",
+            jsonpath="$.response",
+        )
+
+    def process(self) -> dict:
+        provider_config = self.get_provider_config(provider_slug=self.provider_slug(), processor_slug="*")
+        deployment_config = self.get_provider_config(provider_slug=self.provider_slug(), processor_slug="*")
+        api_key = deployment_config.api_key
+        response = get(
+            url="https://api.bounceban.com/v1/verify/single/status",
+            headers={"Authorization": f"{api_key}"},
+            params=self._input.model_dump(),
+        )
+        self._usage_data.append(
+            (
+                f"{self.provider_slug()}/*/*/*",
+                MetricType.API_INVOCATION,
+                (provider_config.provider_config_source, 1),
+            )
+        )
+        async_to_sync(self._output_stream.write)(
+            VerifyStatusProcessorOutput(
+                response=response.text,
+                response_json=response.json(),
+                headers=response.headers,
+                code=response.status_code,
+                size=len(response.content),
+                time=response.elapsed.total_seconds(),
+            )
+        )
+
+        output = self._output_stream.finalize()
+        return output