LinktLinkt
Dashboard
How-To GuidesIngest

CSV Import

Upload and enrich company data from CSV files

Import your existing company or contact lists via CSV and let Linkt's AI agents enrich them with additional data points. This guide walks through the complete workflow from file upload to retrieving enriched entities.

Prerequisites

Before starting, ensure you have:

  • A Linkt account with API access
  • Your API key (see Authentication)
  • A CSV file with company or contact data

CSV Format Requirements

Your CSV file should follow these guidelines:

Required

  • Header row — First row must contain column names
  • Primary column — At least one column with entity names (company names or person names)
  • UTF-8 encoding — Ensure proper character encoding

For company data:

  • Company name (required as primary column)
  • Website/domain
  • Industry
  • Location
  • Employee count

For person data:

  • Full name (required as primary column)
  • Company name
  • Job title
  • Email
  • LinkedIn URL

Example CSV

company_name,website,industry,location
Acme Corporation,acme.com,Software,San Francisco
TechStartup Inc,techstartup.io,SaaS,New York
GlobalCorp,globalcorp.com,Manufacturing,Chicago

Step 1: Upload Your CSV

Upload your CSV file to get a file_id for the ingest task.

curl -X POST "https://api.linkt.ai/v1/files/upload" \
  -H "x-api-key: your-api-key" \
  -F "file=@companies.csv"

Response:

{
  "file_id": "507f1f77bcf86cd799439001",
  "name": "companies.csv",
  "content_type": "text/csv",
  "size_bytes": 2048,
  "csv_metadata": {
    "row_count": 150,
    "columns": ["company_name", "website", "industry", "location"],
    "preview_rows": [
      {"company_name": "Acme Corporation", "website": "acme.com", "industry": "Software", "location": "San Francisco"}
    ],
    "encoding": "utf-8"
  },
  "processing_status": "completed"
}

Step 2: Create an Enrichment ICP

Create an ICP that defines what data to research and add to your entities. For CSV import, the entity target description should focus on enrichment fields rather than search criteria.

curl -X POST "https://api.linkt.ai/v1/icp" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Company Enrichment",
    "description": "Enrich imported company data with additional fields",
    "mode": "discovery",
    "entity_targets": [
      {
        "entity_type": "company",
        "description": "## Enrichment Fields\n- primary_product: The main product or service offered\n- tech_stack: Key technologies and platforms used\n- recent_funding: Latest funding round and amount\n- employee_growth: Year-over-year headcount change",
        "root": true
      }
    ]
  }'

Response:

{
  "id": "507f1f77bcf86cd799439002",
  "name": "Company Enrichment",
  "mode": "discovery",
  "entity_targets": [...],
  "created_at": "2025-01-06T10:00:00Z"
}

Save the id — you'll need it when executing the task.

Step 3: Create a Sheet

Create a sheet to store the enriched entities. The entity_type should match the type of data in your CSV.

curl -X POST "https://api.linkt.ai/v1/sheet" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Q1 Imported Companies",
    "icp_id": "507f1f77bcf86cd799439002",
    "entity_type": "company"
  }'

Response:

{
  "id": "507f1f77bcf86cd799439003",
  "name": "Q1 Imported Companies",
  "icp_id": "507f1f77bcf86cd799439002",
  "entity_type": "company",
  "created_at": "2025-01-06T10:01:00Z"
}

Step 4: Configure the Ingest Task

Create an ingest task that references your uploaded file. The task configuration requires:

  • file_id — The ID from step 1
  • primary_column — Column name containing entity names to match
  • csv_entity_type — Entity type in the CSV (company or person)
curl -X POST "https://api.linkt.ai/v1/task" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Import Q1 Companies",
    "description": "Import and enrich 150 companies from Q1 leads list",
    "flow_name": "ingest",
    "deployment_name": "ingest/v1",
    "sheet_id": "507f1f77bcf86cd799439003",
    "task_config": {
      "version": "v1.0",
      "config_type": "ingest-task",
      "file_id": "507f1f77bcf86cd799439001",
      "primary_column": "company_name",
      "csv_entity_type": "company"
    }
  }'

Response:

{
  "id": "507f1f77bcf86cd799439004",
  "name": "Import Q1 Companies",
  "flow_name": "ingest",
  "task_config": {
    "version": "v1.0",
    "config_type": "ingest-task",
    "file_id": "507f1f77bcf86cd799439001",
    "primary_column": "company_name",
    "csv_entity_type": "company"
  },
  "created_at": "2025-01-06T10:02:00Z"
}

Task Configuration Fields

FieldRequiredDescription
file_idYesMongoDB ObjectId of the uploaded file
primary_columnYesColumn name containing entity names
csv_entity_typeYesEntity type: company or person
versionYesConfig version (use v1.0)
config_typeYesMust be ingest-task

Step 5: Execute and Monitor

Execute the task to start the import and enrichment process.

curl -X POST "https://api.linkt.ai/v1/task/507f1f77bcf86cd799439004/execute" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "icp_id": "507f1f77bcf86cd799439002"
  }'

Response:

{
  "run_id": "507f1f77bcf86cd799439005",
  "flow_run_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "SCHEDULED"
}

The run will progress through states: SCHEDULED → PENDING → RUNNING → COMPLETED. Poll the run endpoint until it reaches a terminal state.

Monitor Processing Queue

For large imports, you can monitor the processing queue:

curl -X GET "https://api.linkt.ai/v1/run/507f1f77bcf86cd799439005/queue" \
  -H "x-api-key: your-api-key"

Queue states:

  • queued — Waiting to be processed
  • processing — Currently being enriched
  • completed — Successfully imported
  • discarded — Skipped (entity not found or not qualified)

Step 6: Review Results

Once the run completes, retrieve the enriched entities from your sheet.

curl -X GET "https://api.linkt.ai/v1/sheet/507f1f77bcf86cd799439003/entities" \
  -H "x-api-key: your-api-key"

Response:

{
  "entities": [
    {
      "id": "507f1f77bcf86cd799439010",
      "sheet_id": "507f1f77bcf86cd799439003",
      "data": {
        "name": {
          "value": "Acme Corporation",
          "references": ["https://acme.com"]
        },
        "website": {
          "value": "https://acme.com",
          "references": []
        },
        "primary_product": {
          "value": "Enterprise CRM software for mid-market companies",
          "references": ["https://acme.com/products"]
        },
        "tech_stack": {
          "value": "React, Node.js, PostgreSQL, AWS",
          "references": ["https://stackshare.io/acme"]
        },
        "recent_funding": {
          "value": "Series B - $45M (January 2025)",
          "references": ["https://techcrunch.com/acme-series-b"]
        }
      },
      "created_at": "2025-01-06T10:15:00Z"
    }
  ],
  "total": 150,
  "page": 1,
  "page_size": 20
}

Complete Example

Here's a complete Python script for CSV import:

import requests
import time
 
API_KEY = "your-api-key"
BASE_URL = "https://api.linkt.ai/v1"
HEADERS = {"x-api-key": API_KEY}
 
def csv_import(file_path, primary_column, entity_type="company"):
    """Import and enrich a CSV file."""
 
    # Step 1: Upload CSV
    print("Uploading CSV...")
    with open(file_path, "rb") as f:
        response = requests.post(
            f"{BASE_URL}/files/upload",
            headers=HEADERS,
            files={"file": f}
        )
    file_data = response.json()
    file_id = file_data["file_id"]
    print(f"Uploaded: {file_data['csv_metadata']['row_count']} rows")
 
    # Step 2: Create ICP
    print("Creating ICP...")
    icp_response = requests.post(
        f"{BASE_URL}/icp",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={
            "name": f"Enrichment - {file_path}",
            "mode": "discovery",
            "entity_targets": [{
                "entity_type": entity_type,
                "description": "## Enrichment Fields\n- industry: Industry sector\n- description: Company description\n- employee_count: Number of employees",
                "root": True
            }]
        }
    )
    icp_id = icp_response.json()["id"]
 
    # Step 3: Create Sheet
    print("Creating sheet...")
    sheet_response = requests.post(
        f"{BASE_URL}/sheet",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={
            "name": f"Import - {file_path}",
            "icp_id": icp_id,
            "entity_type": entity_type
        }
    )
    sheet_id = sheet_response.json()["id"]
 
    # Step 4: Create Task
    print("Creating ingest task...")
    task_response = requests.post(
        f"{BASE_URL}/task",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={
            "name": f"Import {file_path}",
            "description": "CSV import and enrichment",
            "flow_name": "ingest",
            "deployment_name": "ingest/v1",
            "sheet_id": sheet_id,
            "task_config": {
                "version": "v1.0",
                "config_type": "ingest-task",
                "file_id": file_id,
                "primary_column": primary_column,
                "csv_entity_type": entity_type
            }
        }
    )
    task_id = task_response.json()["id"]
 
    # Step 5: Execute
    print("Executing...")
    run_response = requests.post(
        f"{BASE_URL}/task/{task_id}/execute",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"icp_id": icp_id}
    )
    run_id = run_response.json()["run_id"]
 
    # Poll for completion
    while True:
        run = requests.get(
            f"{BASE_URL}/run/{run_id}",
            headers=HEADERS
        ).json()
 
        print(f"  Status: {run['status']}")
 
        if run["status"] == "COMPLETED":
            break
        elif run["status"] in ["FAILED", "CANCELED", "CRASHED"]:
            raise Exception(f"Import failed: {run.get('error')}")
 
        time.sleep(10)  # Poll every 10 seconds
 
    # Step 6: Get results
    entities = requests.get(
        f"{BASE_URL}/sheet/{sheet_id}/entities",
        headers=HEADERS
    ).json()
 
    print(f"Success! Imported {entities['total']} entities")
    return entities
 
# Run import
result = csv_import("companies.csv", "company_name", "company")

Best Practices

Data Quality

  • Clean your data — Remove duplicates and fix obvious errors before upload
  • Consistent naming — Use consistent company name formats
  • Include domains — Website/domain helps with entity matching

Performance

  • Batch size — For large files (1000+ rows), consider splitting into smaller batches
  • Timeout — Allow sufficient time for enrichment (larger files take longer)
  • Monitor queue — Use the queue endpoint to track progress

Enrichment ICP

  • Focus on custom fields — Default fields (name, website, etc.) are populated automatically
  • Be specific — Clear field descriptions produce better results
  • Prioritize fields — List most important fields first

Error Handling

Common Issues

IssueCauseSolution
"Column not found"Primary column doesn't existCheck column names in csv_metadata
"Invalid CSV format"Malformed CSVEnsure proper UTF-8 encoding and formatting
"Entity not found"Company/person couldn't be matchedCheck for typos in entity names
Partial completionSome rows failed enrichmentReview discarded items in queue

Handling Partial Failures

Some entities may be discarded if they can't be matched or enriched. Check the run queue for details:

curl -X GET "https://api.linkt.ai/v1/run/{run_id}/queue?state=discarded" \
  -H "x-api-key: your-api-key"

Next Steps

Previous

Ingest Guides

Next

Bulk Enrichment

On this page