AI-Directed Contact List Loader - Design Document¶

Date: December 3, 2025 Location: cbtenant (Tenant Manager) Status: Design Phase

Overview¶

The List Loader enables campaign staff to upload CSV/Excel contact lists via the Tenant Manager and have AI automatically suggest field mappings to the cbapp Person schema.

Architecture¶

┌─────────────────────────────────────────────────────────────────┐
│                    Tenant Manager (cbtenant)                     │
│                                                                  │
│  ┌─────────────────┐    ┌──────────────────┐                   │
│  │  List Loader    │    │  Claude Agent    │                   │
│  │  Frontend Page  │───▶│  (Field Mapper)  │                   │
│  │  /tenants/{id}  │    │                  │                   │
│  │  /list-loader   │◀───│  AI Suggestions  │                   │
│  └────────┬────────┘    └──────────────────┘                   │
│           │                                                      │
│           │ Mapped Data                                          │
│           ▼                                                      │
│  ┌─────────────────┐                                            │
│  │  Import Service │                                            │
│  │  - Validation   │                                            │
│  │  - Batch API    │                                            │
│  │  - Progress     │                                            │
│  └────────┬────────┘                                            │
│           │                                                      │
└───────────┼──────────────────────────────────────────────────────┘
            │ HTTP API Calls
            ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Tenant App (cbapp)                           │
│                                                                  │
│  ┌─────────────────┐    ┌──────────────────┐                   │
│  │  POST /persons  │───▶│  DuckDB          │                   │
│  │  (per record)   │    │  person table    │                   │
│  └─────────────────┘    └──────────────────┘                   │
│                                                                  │
│  NEW: POST /persons/batch (to be added)                         │
└─────────────────────────────────────────────────────────────────┘

cbapp Person Schema (Target Fields)¶

Core Fields¶

Field	Type	Required	Description
`first_name`	varchar	YES	First name
`last_name`	varchar	YES	Last name
`address1`	varchar	no	Street address
`address2`	varchar	no	Apt/Suite
`city`	varchar	no	City
`state`	varchar	no	State code
`zip`	varchar	no	ZIP code
`county`	varchar	no	County name
`congressional_district`	varchar	no	e.g., "KY-04"

Contact Fields¶

Field	Type	Required	Description
`home_phone`	varchar	no	Home phone
`cell_phone`	varchar	no	Cell phone
`email`	varchar	no	Email address

Campaign Fields¶

Field	Type	Required	Description
`precinct`	varchar	no	Precinct name
`precinct_number`	varchar	no	Precinct number
`precinct_status`	varchar	no	Voter status
`precinct_role`	varchar	no	Captain/Member
`notes`	varchar	no	Free-form notes
`whip_status`	varchar	no	Support level

Special Fields¶

Field	Type	Description
`tags`	list[str]	Auto-created if needed
`custom_fields`	dict	Key-value pairs

Component Design¶

1. Frontend Page (cbtenant)¶

Location: frontend/templates/tenant_list_loader.html

URL: /tenants/{tenant_id}/list-loader

Steps: 1. Upload - Drag/drop CSV/Excel file 2. Preview - Show first 5 rows + headers 3. AI Mapping - Claude suggests field mappings 4. Adjust - User can override AI suggestions 5. Validate - Show warnings (missing required, bad formats) 6. Import - Execute with progress bar 7. Summary - Show results (success/failures/duplicates)

2. List Loader Agent (cbtenant)¶

Location: agents/list_loader.py

class ListLoaderAgent:
    """AI-powered field mapping for contact imports."""

    async def analyze_file(self, file_path: str) -> AnalysisResult:
        """
        Analyze uploaded file structure.
        Returns: headers, sample_rows, detected_encoding
        """

    async def suggest_mappings(
        self,
        headers: list[str],
        sample_rows: list[dict]
    ) -> list[FieldMapping]:
        """
        Use Claude to suggest field mappings.
        Returns: list of {source_column, target_field, confidence}
        """

    def validate_mappings(
        self,
        data: list[dict],
        mappings: list[FieldMapping]
    ) -> ValidationResult:
        """
        Validate data against Person schema.
        Returns: errors, warnings, valid_count
        """

3. Import Service (cbtenant)¶

Location: api/services/import_service.py

class ImportService:
    """Execute imports to tenant cbapp instances."""

    async def execute_import(
        self,
        tenant_id: str,
        data: list[dict],
        mappings: list[FieldMapping],
        options: ImportOptions
    ) -> ImportResult:
        """
        Import data to tenant's cbapp via API.
        - Uses tenant's API credentials
        - Calls POST /persons for each record (or batch)
        - Tracks progress
        - Handles errors/retries
        """

4. API Routes (cbtenant)¶

Location: api/routes/list_loader.py

Method	Endpoint	Description
POST	`/api/tenants/{id}/list-loader/upload`	Upload file, return job_id
GET	`/api/tenants/{id}/list-loader/{job_id}/preview`	Get headers + sample
POST	`/api/tenants/{id}/list-loader/{job_id}/analyze`	Trigger AI mapping
GET	`/api/tenants/{id}/list-loader/{job_id}/mappings`	Get suggested mappings
PUT	`/api/tenants/{id}/list-loader/{job_id}/mappings`	Update mappings
POST	`/api/tenants/{id}/list-loader/{job_id}/validate`	Validate data
POST	`/api/tenants/{id}/list-loader/{job_id}/execute`	Run import
GET	`/api/tenants/{id}/list-loader/{job_id}/status`	Get import progress
GET	`/api/tenants/{id}/list-loader/history`	List past imports

AI Field Mapping¶

Claude Prompt¶

You are a data mapping assistant for a political campaign CRM.

Given these CSV headers and sample data, suggest mappings to the Person schema.

**CSV Headers:** {headers}

**Sample Rows:**
{sample_rows}

**Target Schema Fields:**
- first_name (required)
- last_name (required)
- address1, address2, city, state, zip
- county, congressional_district
- home_phone, cell_phone, email
- precinct, precinct_number, precinct_status
- notes
- tags (list)
- custom_fields (any other data)

**Instructions:**
1. Map each CSV column to the most appropriate schema field
2. Use "custom_fields.{name}" for unmapped but useful columns
3. Use "skip" for columns to ignore
4. Provide confidence score (0.0-1.0)

**Response Format (JSON):**
[
  {"source": "First Name", "target": "first_name", "confidence": 1.0},
  {"source": "Party", "target": "custom_fields.party", "confidence": 0.9},
  {"source": "ID", "target": "skip", "confidence": 0.8}
]

Common Field Aliases¶

The AI should recognize these common variations:

Target Field	Common Source Names
first_name	First, First Name, FirstName, FName, Given Name
last_name	Last, Last Name, LastName, LName, Surname, Family Name
address1	Address, Street, Street Address, Addr, Address 1, Mailing Address
city	City, Town, Municipality
state	State, ST, Province
zip	Zip, ZIP, Zip Code, Postal Code, ZIP Code
cell_phone	Cell, Mobile, Cell Phone, Mobile Phone, Phone
home_phone	Home, Home Phone, Landline, Phone 2
email	Email, E-mail, Email Address
county	County, Cty

Database Schema (cbtenant)¶

import_jobs table¶

CREATE TABLE import_jobs (
    id VARCHAR PRIMARY KEY,
    tenant_id VARCHAR NOT NULL,
    filename VARCHAR NOT NULL,
    file_path VARCHAR NOT NULL,
    file_size INTEGER,
    row_count INTEGER,
    status VARCHAR DEFAULT 'pending',  -- pending, analyzing, ready, importing, completed, failed
    mappings JSON,
    validation_result JSON,
    import_result JSON,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP,
    created_by VARCHAR
);

CREATE INDEX idx_import_jobs_tenant ON import_jobs(tenant_id);
CREATE INDEX idx_import_jobs_status ON import_jobs(status);

import_templates table¶

CREATE TABLE import_templates (
    id VARCHAR PRIMARY KEY,
    name VARCHAR NOT NULL,
    description VARCHAR,
    mappings JSON NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_by VARCHAR
);

API Flow Example¶

1. Upload File¶

POST /api/tenants/{tenant_id}/list-loader/upload
Content-Type: multipart/form-data

file: volunteers.csv

Response:

{
  "job_id": "imp_abc123",
  "filename": "volunteers.csv",
  "status": "pending",
  "row_count": 150
}

2. Get Preview¶

GET /api/tenants/{tenant_id}/list-loader/imp_abc123/preview

Response:

{
  "headers": ["First Name", "Last Name", "Email", "Phone", "Party"],
  "sample_rows": [
    {"First Name": "John", "Last Name": "Doe", "Email": "john@example.com", "Phone": "555-1234", "Party": "R"},
    {"First Name": "Jane", "Last Name": "Smith", "Email": "jane@example.com", "Phone": "555-5678", "Party": "D"}
  ],
  "total_rows": 150
}

3. Trigger AI Analysis¶

POST /api/tenants/{tenant_id}/list-loader/imp_abc123/analyze

Response:

{
  "status": "analyzing",
  "message": "AI is analyzing your file..."
}

4. Get Mappings¶

GET /api/tenants/{tenant_id}/list-loader/imp_abc123/mappings

Response:

{
  "status": "ready",
  "mappings": [
    {"source": "First Name", "target": "first_name", "confidence": 1.0},
    {"source": "Last Name", "target": "last_name", "confidence": 1.0},
    {"source": "Email", "target": "email", "confidence": 1.0},
    {"source": "Phone", "target": "cell_phone", "confidence": 0.9},
    {"source": "Party", "target": "custom_fields.party", "confidence": 0.85}
  ],
  "unmapped": [],
  "warnings": []
}

5. Execute Import¶

POST /api/tenants/{tenant_id}/list-loader/imp_abc123/execute
Content-Type: application/json

{
  "skip_duplicates": true,
  "update_existing": false,
  "tags": ["imported-2025-12-03"]
}

Response:

{
  "status": "importing",
  "progress": {
    "total": 150,
    "processed": 0,
    "success": 0,
    "failed": 0,
    "duplicates": 0
  }
}

cbapp API Enhancement (Required)¶

New Batch Import Endpoint¶

Add to cbapp/src/api/routes/persons.py:

@router.post("/batch", response_model=BatchImportResult)
async def batch_import_persons(
    persons: List[PersonCreate],
    skip_duplicates: bool = True,
    update_existing: bool = False,
    import_tag: Optional[str] = None,
    current_user: User = Depends(get_current_active_user)
):
    """
    Bulk import persons.
    - Much faster than individual POSTs
    - Transaction-safe (all or nothing, or best-effort)
    - Returns detailed results
    """

Alternative: Use existing single-record POST with concurrency.

Questions¶

Batch API Priority: Should we add a batch import endpoint to cbapp first, or use existing single-record API with concurrency?
Authentication: How does cbtenant authenticate to cbapp APIs?
Service-to-service token?
Admin user credentials per tenant?
API key in tenant config?
Duplicate Detection: What defines a duplicate?
Email only?
Name + Address?
Phone number?
Configurable?
Tag Assignment: Should imports auto-create a tag like import-2025-12-03 for easy filtering?
File Storage: Where to store uploaded files?
Temporary directory?
Tenant-specific folder?
Delete after import?
Error Handling: On partial failure, should we:
Rollback all?
Keep successful, report failures?
Retry failed records?

Implementation Phases¶

Phase 1: Foundation (1 day)¶

Create database tables (import_jobs, import_templates)
Create basic API routes
File upload + parsing (CSV/Excel)
Preview endpoint

Phase 2: AI Mapping (1 day)¶

Create ListLoaderAgent
Design and test Claude prompt
Implement mapping suggestion
Handle confidence scores

Phase 3: Import Execution (1 day)¶

Add batch endpoint to cbapp (or use single-record with concurrency)
Create ImportService
Progress tracking
Error handling

Phase 4: Frontend (1 day)¶

Phase 5: Polish (0.5 day)¶

Save/load mapping templates
Import history view
Documentation

Files to Create¶

cbtenant/
├── agents/
│   └── list_loader.py           # AI field mapping agent
├── api/
│   ├── routes/
│   │   └── list_loader.py       # API endpoints
│   └── services/
│       └── import_service.py    # Import execution
├── frontend/
│   └── templates/
│       └── tenant_list_loader.html
└── db/
    └── schema.sql               # Add import_jobs table

Next Steps¶

Answer questions above
Add import_jobs table to cbtenant schema
Create list_loader.py agent
Create API routes
Create frontend page
Test with real data

Ready to start building when you give the go!