Python SDK¶

Complete guide for using the SPOT Contracts Python SDK.

Installation¶

pip install spot-sdk \
  --index-url https://pypi.org/simple/

Requirements¶

Python 3.11+
pydantic >= 2.0.0
httpx >= 0.25.0
typing-extensions >= 4.8.0

Quick Start¶

from spot_sdk.api_gateway import Email, EmailHeader
from spot_sdk.api_gateway import AnalysisResult, AnalysisIndicator, AnalysisMetadata

# Create email object
email = Email(
    id="test-123",
    headers=EmailHeader(
        subject="Test email",
        sender="test@example.com",
        recipients=["user@company.com"]
    ),
    body_text="Email content here"
)

# Create analysis result
result = AnalysisResult(
    is_phishing=True,
    threat_level="high",
    confidence=0.85,
    explanation="Found suspicious indicators",
    indicators=[
        AnalysisIndicator(
            type="suspicious_links",
            description="Suspicious domain detected",
            severity="high",
            confidence=0.9
        )
    ],
    metadata=AnalysisMetadata(
        analyzer_id="my-analyzer",
        analyzer_version="1.0.0",
        analysis_duration_ms=250
    )
)

# Serialize to JSON (Pydantic v2)
json_data = result.model_dump_json()
print(json_data)

Available Models¶

Email Models¶

Module: spot_sdk.email

Email

from spot_sdk.api_gateway import Email

email = Email(
    id="unique-id",
    headers=EmailHeader(...),
    body_text="Plain text body",
    body_html="<html>HTML body</html>",
    attachments=[Attachment(...)]
)

EmailHeader

from spot_sdk.api_gateway import EmailHeader

headers = EmailHeader(
    subject="Email subject",
    sender="sender@example.com",
    recipients=["recipient@example.com"],
    cc=["cc@example.com"],
    bcc=["bcc@example.com"],
    date="2024-01-15T10:30:00Z",
    message_id="<unique-message-id>",
    in_reply_to="<parent-message-id>",
    references=["<ref1>", "<ref2>"]
)

Attachment

from spot_sdk.api_gateway import Attachment

attachment = Attachment(
    filename="document.pdf",
    content_type="application/pdf",
    size=1024,
    content="base64-encoded-content",
    content_id="unique-content-id"
)

Analysis Models¶

Module: spot_sdk.results

AnalysisResult

from spot_sdk.api_gateway import AnalysisResult, AnalysisMetadata

result = AnalysisResult(
    is_phishing=True,           # bool: Is this email phishing?
    threat_level="high",         # str: safe|low|medium|high|critical
    confidence=0.85,             # float: 0.0-1.0
    explanation="Detailed explanation",
    indicators=[...],            # List[AnalysisIndicator]
    metadata=AnalysisMetadata(...),
    raw_output={"custom": "data"}  # Optional: any additional data
)

AnalysisIndicator

from spot_sdk.api_gateway import AnalysisIndicator, IndicatorType

indicator = AnalysisIndicator(
    type=IndicatorType.SUSPICIOUS_LINKS,  # or use string
    description="Found shortened URL: bit.ly/xyz",
    severity="high",              # low|medium|high
    confidence=0.9,               # 0.0-1.0
    evidence="bit.ly/xyz123",     # Optional: what triggered this
    location="body"               # Optional: where in email
)

IndicatorType Enum

from spot_sdk.api_gateway import IndicatorType

# Available indicator types:
IndicatorType.SUSPICIOUS_LINKS           # Malicious/shortened URLs
IndicatorType.DOMAIN_SPOOFING            # Sender domain mismatch
IndicatorType.URGENT_LANGUAGE            # Pressure tactics
IndicatorType.SPELLING_GRAMMAR_ERRORS    # Poor grammar/typos
IndicatorType.SOCIAL_ENGINEERING         # Social engineering tactics
IndicatorType.SUSPICIOUS_ATTACHMENTS     # Dangerous file types
IndicatorType.CONTEXT_MISMATCH           # Content doesn't match sender
IndicatorType.HEADER_ANOMALIES           # Email header issues

AnalysisMetadata

from spot_sdk.api_gateway import AnalysisMetadata

metadata = AnalysisMetadata(
    analyzer_id="my-analyzer",
    analyzer_version="1.0.0",
    analysis_duration_ms=245,
    model_version="optional-model-version",
    timestamp="2024-01-15T10:30:00Z"
)

Workflow Models¶

Module: spot_sdk.workflow

Workflow

from spot_sdk.api_gateway import Workflow, WorkflowStage

workflow = Workflow(
    id="workflow-id",
    name="Default Workflow",
    description="Standard phishing detection workflow",
    stages=[WorkflowStage(...)],
    enabled=True
)

Orchestrator Models¶

Module: spot_sdk.orchestrator

OrchestrationResult

from spot_sdk.api_gateway import OrchestrationResult, AnalyzerResult

result = OrchestrationResult(
    job_id="unique-job-id",
    email_id="email-id",
    workflow_id="workflow-id",
    overall_verdict="phishing",
    overall_confidence=0.85,
    analyzer_results=[AnalyzerResult(...)],
    timestamp="2024-01-15T10:30:00Z"
)

API Client Example¶

import httpx
from spot_sdk.api_gateway import Email, EmailHeader
from spot_sdk.api_gateway import AnalysisResult

class SpotApiClient:
    def __init__(
        self,
        base_url: str = "http://localhost:8001",
        token: str = None
    ):
        headers = {"Content-Type": "application/json"}
        if token:
            headers["Authorization"] = f"Bearer {token}"

        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers=headers,
            timeout=30.0
        )

    async def analyze_email(self, email: Email) -> AnalysisResult:
        """Analyze an email for spear-phishing indicators."""
        response = await self.client.post(
            "/api/v1/analyze",
            json=email.model_dump()
        )
        response.raise_for_status()
        return AnalysisResult(**response.json())

    async def get_status(self, job_id: str) -> dict:
        """Check analysis job status."""
        response = await self.client.get(f"/api/v1/analyze/{job_id}")
        response.raise_for_status()
        return response.json()

    async def close(self):
        """Close the HTTP client."""
        await self.client.aclose()

# Usage example
async def main():
    async with SpotApiClient(token="your-token") as client:
        # Create email
        email = Email(
            id="test-123",
            headers=EmailHeader(
                subject="Urgent: Verify your account",
                sender="noreply@suspicious-domain.com",
                recipients=["user@company.com"]
            ),
            body_text="Click here to verify: http://bit.ly/xyz123"
        )

        # Analyze email
        result = await client.analyze_email(email)

        print(f"Phishing: {result.is_phishing}")
        print(f"Threat Level: {result.threat_level}")
        print(f"Confidence: {result.confidence}")
        print(f"Indicators: {len(result.indicators)}")

        for indicator in result.indicators:
            print(f"  - {indicator.type}: {indicator.description}")

# Run
import asyncio
asyncio.run(main())

Validation and Serialization¶

Pydantic v2 provides powerful validation and serialization:

from spot_sdk.api_gateway import AnalysisResult

# Create with validation
result = AnalysisResult(
    is_phishing=True,
    threat_level="high",
    confidence=0.85,
    explanation="Test",
    indicators=[],
    metadata=AnalysisMetadata(...)
)

# Serialize to dict
data = result.model_dump()

# Serialize to JSON
json_str = result.model_dump_json(indent=2)

# Parse from dict
result2 = AnalysisResult(**data)

# Parse from JSON
result3 = AnalysisResult.model_validate_json(json_str)

# Exclude fields
data = result.model_dump(exclude={"raw_output"})

# Include only specific fields
data = result.model_dump(include={"is_phishing", "threat_level"})

Error Handling¶

from pydantic import ValidationError
from spot_sdk.api_gateway import AnalysisResult

try:
    result = AnalysisResult(
        is_phishing="invalid",  # Should be bool
        threat_level="unknown",  # Invalid threat level
        confidence=1.5,          # Out of range
        explanation="test"
    )
except ValidationError as e:
    print("Validation errors:")
    for error in e.errors():
        print(f"  Field: {error['loc']}")
        print(f"  Error: {error['msg']}")
        print(f"  Type: {error['type']}")

Testing with SDK¶

import pytest
from spot_sdk.api_gateway import Email, EmailHeader
from spot_sdk.api_gateway import AnalysisResult, AnalysisIndicator

def test_email_creation():
    """Test creating an email object."""
    email = Email(
        id="test-1",
        headers=EmailHeader(
            subject="Test",
            sender="test@example.com",
            recipients=["user@example.com"]
        ),
        body_text="Test content"
    )
    assert email.id == "test-1"
    assert email.headers.subject == "Test"

def test_analysis_result_validation():
    """Test analysis result validation."""
    with pytest.raises(ValueError):
        AnalysisResult(
            is_phishing=True,
            threat_level="invalid",  # Should fail
            confidence=0.5,
            explanation="test"
        )

@pytest.mark.asyncio
async def test_api_client(mock_api_response):
    """Test API client integration."""
    client = SpotApiClient()
    email = Email(...)
    result = await client.analyze_email(email)
    assert isinstance(result, AnalysisResult)
    await client.close()

Configuration¶

Using with Poetry¶

[tool.poetry.dependencies]
python = "^3.11"
spot-sdk = {version = "^2.0.0", source = "gitlab"}

[[tool.poetry.source]]
name = "gitlab"
url = "https://pypi.org/simple/"
priority = "supplemental"

Using with pip¶

# requirements.txt
--index-url https://pypi.org/simple/
spot-sdk>=2.0.0

Troubleshooting¶

Cannot find package¶

Ensure pip is configured to use the GitLab registry:

pip install spot-sdk \
  --index-url https://pypi.org/simple/

Import errors¶

Check that you're using the correct module paths:

# Correct
from spot_sdk.api_gateway import Email
from spot_sdk.api_gateway import AnalysisResult

# Incorrect
from spot_sdk.api_gateway Email  # Module not found

Validation errors¶

Check the error details to see which field failed validation:

try:
    result = AnalysisResult(...)
except ValidationError as e:
    print(e.json(indent=2))

Support¶

Issues: GitLab Issues
Documentation: Main Documentation
SPOT Platform: Platform Docs