FastAPI

Integrate the ScamVerify™ API with your FastAPI application using async endpoints, httpx, and Pydantic models.

This guide shows you how to call the ScamVerify™ API from a FastAPI application using async HTTP calls with httpx and Pydantic models for request/response validation.

Environment Setup

Install dependencies:

pip install fastapi uvicorn httpx python-dotenv pydantic-settings

Create a .env file:

SCAMVERIFY_API_KEY=sv_live_your_key_here

Configuration

Use Pydantic Settings for type-safe configuration.

# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    scamverify_api_key: str
    scamverify_base_url: str = "https://scamverify.ai/api/v1"

    class Config:
        env_file = ".env"


settings = Settings()

Pydantic Models

Define models for request validation and response typing.

# models.py
from pydantic import BaseModel, Field
from typing import Optional


# Request models
class PhoneLookupRequest(BaseModel):
    phone_number: str = Field(..., description="US phone number in E.164 or standard format")


class UrlLookupRequest(BaseModel):
    url: str = Field(..., description="URL to verify")
    force_refresh: bool = Field(False, description="Bypass cache and force a fresh lookup")


class TextAnalyzeRequest(BaseModel):
    text: str = Field(..., description="Text message content to analyze")
    sender_number: Optional[str] = Field(None, description="Sender phone number if available")


class EmailAnalyzeRequest(BaseModel):
    email_body: str = Field(..., description="Email body content")
    raw_headers: Optional[str] = Field(None, description="Raw email headers for SPF/DKIM/DMARC analysis")


# Response models
class SignalsResponse(BaseModel):
    ftc_complaints: Optional[int] = None
    fcc_complaints: Optional[int] = None
    carrier: Optional[str] = None
    line_type: Optional[str] = None
    robocall_detected: Optional[bool] = None
    community_reports: Optional[int] = None
    urlhaus_match: Optional[bool] = None
    threatfox_match: Optional[bool] = None
    extracted_urls: Optional[list] = None
    extracted_phones: Optional[list] = None
    header_analysis: Optional[dict] = None
    brand_impersonation: Optional[str] = None

    class Config:
        extra = "allow"


class LookupResponse(BaseModel):
    risk_score: int = Field(..., ge=0, le=100)
    verdict: str
    explanation: str
    signals: SignalsResponse
    cached: bool = False

Async ScamVerify™ Client

Use httpx.AsyncClient for non-blocking API calls.

# scamverify_client.py
import httpx
from config import settings


class ScamVerifyError(Exception):
    def __init__(self, message: str, status_code: int = 0):
        super().__init__(message)
        self.status_code = status_code


class ScamVerifyClient:
    def __init__(self):
        self.base_url = settings.scamverify_base_url
        self.headers = {
            "Authorization": f"Bearer {settings.scamverify_api_key}",
            "Content-Type": "application/json",
        }

    async def _request(self, endpoint: str, payload: dict) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}{endpoint}",
                json=payload,
                headers=self.headers,
                timeout=30.0,
            )

        if response.status_code == 401:
            raise ScamVerifyError("Invalid or revoked API key", 401)
        if response.status_code == 402:
            raise ScamVerifyError("Quota exhausted", 402)
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After", "60")
            raise ScamVerifyError(f"Rate limited. Retry after {retry_after}s", 429)
        if not response.is_success:
            raise ScamVerifyError(f"API error: {response.status_code}", response.status_code)

        return response.json()

    async def lookup_phone(self, phone_number: str) -> dict:
        return await self._request("/phone/lookup", {"phone_number": phone_number})

    async def lookup_url(self, url: str, force_refresh: bool = False) -> dict:
        payload = {"url": url}
        if force_refresh:
            payload["force_refresh"] = True
        return await self._request("/url/lookup", payload)

    async def analyze_text(self, text: str, sender_number: str = None) -> dict:
        payload = {"text": text}
        if sender_number:
            payload["sender_number"] = sender_number
        return await self._request("/text/analyze", payload)

    async def analyze_email(self, email_body: str, raw_headers: str = None) -> dict:
        payload = {"email_body": email_body}
        if raw_headers:
            payload["raw_headers"] = raw_headers
        return await self._request("/email/analyze", payload)

    async def batch_phone(self, phone_numbers: list[str]) -> dict:
        return await self._request(
            "/batch/phone",
            {"items": [{"phone_number": n} for n in phone_numbers]},
        )

    async def batch_url(self, urls: list[str]) -> dict:
        return await self._request(
            "/batch/url",
            {"items": [{"url": u} for u in urls]},
        )

FastAPI Application

# main.py
from fastapi import FastAPI, HTTPException
from models import (
    PhoneLookupRequest,
    UrlLookupRequest,
    TextAnalyzeRequest,
    EmailAnalyzeRequest,
    LookupResponse,
)
from scamverify_client import ScamVerifyClient, ScamVerifyError

app = FastAPI(
    title="Verification API",
    description="Phone, URL, text, and email verification powered by ScamVerify™",
)

client = ScamVerifyClient()


@app.post("/verify/phone", response_model=LookupResponse)
async def verify_phone(request: PhoneLookupRequest):
    """Look up a phone number for scam indicators."""
    try:
        result = await client.lookup_phone(request.phone_number)
        return result
    except ScamVerifyError as e:
        raise HTTPException(status_code=e.status_code or 500, detail=str(e))


@app.post("/verify/url", response_model=LookupResponse)
async def verify_url(request: UrlLookupRequest):
    """Check a URL against threat intelligence databases."""
    try:
        result = await client.lookup_url(request.url, request.force_refresh)
        return result
    except ScamVerifyError as e:
        raise HTTPException(status_code=e.status_code or 500, detail=str(e))


@app.post("/verify/text", response_model=LookupResponse)
async def analyze_text(request: TextAnalyzeRequest):
    """Analyze a text message for scam indicators."""
    try:
        result = await client.analyze_text(request.text, request.sender_number)
        return result
    except ScamVerifyError as e:
        raise HTTPException(status_code=e.status_code or 500, detail=str(e))


@app.post("/verify/email", response_model=LookupResponse)
async def analyze_email(request: EmailAnalyzeRequest):
    """Analyze an email for phishing and spoofing indicators."""
    try:
        result = await client.analyze_email(request.email_body, request.raw_headers)
        return result
    except ScamVerifyError as e:
        raise HTTPException(status_code=e.status_code or 500, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}

Running the Server

uvicorn main:app --reload --port 3000

FastAPI automatically generates interactive API docs at http://localhost:3000/docs (Swagger UI) and http://localhost:3000/redoc (ReDoc).

Testing

# Phone lookup
curl -X POST http://localhost:3000/verify/phone \
  -H "Content-Type: application/json" \
  -d '{"phone_number": "+12025551234"}'

# URL verification
curl -X POST http://localhost:3000/verify/url \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com"}'

# Email analysis with headers
curl -X POST http://localhost:3000/verify/email \
  -H "Content-Type: application/json" \
  -d '{"email_body": "Click here to verify your account", "raw_headers": "From: noreply@example.com\nReceived-SPF: pass"}'

Dependency Injection Pattern

For more advanced setups, use FastAPI's dependency injection to manage the client lifecycle.

from fastapi import Depends

async def get_scamverify_client() -> ScamVerifyClient:
    return ScamVerifyClient()

@app.post("/verify/phone")
async def verify_phone(
    request: PhoneLookupRequest,
    client: ScamVerifyClient = Depends(get_scamverify_client),
):
    try:
        result = await client.lookup_phone(request.phone_number)
        return result
    except ScamVerifyError as e:
        raise HTTPException(status_code=e.status_code or 500, detail=str(e))

Connection pooling: For high-throughput applications, consider creating the httpx.AsyncClient once at startup and reusing it across requests. Use FastAPI's lifespan context manager to manage the client lifecycle.

Django Integration if you prefer synchronous Python
Add URL Scanning to Your App for a complete Python tutorial
Common Issues for troubleshooting

On this page