LinktLinkt

Files

Upload and manage CSV/XLSX files for data workflows

Files are uploaded data sources used in ingest and signal workflows. Upload your company or contact lists as CSV or Excel files, then reference them in tasks to enrich the data or monitor for signals.

File Response Structure

When you upload a file, the API returns metadata about the file and its contents.

FileUploadResponse

{
  "file_id": "507f1f77bcf86cd799439001",
  "name": "companies.csv",
  "s3_uri": "s3://linkt-files/org123/507f1f77bcf86cd799439001.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",
  "original_file_type": null
}

Response Fields

FieldTypeDescription
file_idstringUnique identifier for the file
namestringOriginal filename
s3_uristringInternal storage location
content_typestringMIME type (e.g., text/csv)
size_bytesintegerFile size in bytes
csv_metadataobjectParsed CSV information
processing_statusstringCurrent processing state
original_file_typestringOriginal format if converted (e.g., xlsx)

CSV Metadata

The csv_metadata object contains information extracted from the CSV:

FieldTypeDescription
row_countintegerNumber of data rows (excluding header)
columnsarrayColumn names from the header row
preview_rowsarrayFirst few rows as key-value objects
encodingstringDetected character encoding

Use csv_metadata to verify your file was parsed correctly before creating tasks.

Processing Status

Files go through a processing pipeline after upload. The processing_status field indicates the current state.

Status Lifecycle

Loading diagram...
StatusDescription
pendingFile uploaded, waiting to be processed
processingFile is being parsed and validated
completedFile ready for use in workflows
failedProcessing error occurred

Checking Processing Status

For large files, poll the file endpoint until processing completes:

import requests
import time
 
def wait_for_file_ready(file_id, api_key, max_wait=60):
    """Wait for file processing to complete."""
    headers = {"x-api-key": api_key}
    start = time.time()
 
    while time.time() - start < max_wait:
        response = requests.get(
            f"https://api.linkt.ai/v1/files/{file_id}",
            headers=headers
        )
        file_data = response.json()
        status = file_data.get("processing_status")
 
        if status == "completed":
            return file_data
        elif status == "failed":
            raise Exception("File processing failed")
 
        time.sleep(2)  # Poll every 2 seconds
 
    raise Exception("Timeout waiting for file processing")

Supported File Types

FormatExtensionsNotes
CSV.csvUTF-8 encoding recommended
Excel.xlsx, .xlsFirst sheet only, converted to CSV

Using Files in Workflows

Files are referenced by file_id in task configurations. Two workflow types use files:

Ingest Tasks

Import and enrich entities from a CSV file:

{
  "type": "ingest",
  "file_id": "507f1f77bcf86cd799439001",
  "primary_column": "company_name",
  "csv_entity_type": "company"
}

See CSV Import for the complete workflow.

Signal CSV Tasks

Monitor entities from a CSV file for signals:

{
  "type": "signal-csv",
  "file_id": "507f1f77bcf86cd799439001",
  "primary_column": "company_name",
  "entity_type": "company",
  "signal_types": [
    {
      "type": "funding",
      "display": "Funding Rounds",
      "description": "New funding announcements"
    }
  ]
}

See Topic Monitoring for signal configuration details.

CSV Format Requirements

Required

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

For company data:

ColumnPurpose
Company nameRequired as primary identifier
Website/domainImproves entity matching accuracy
IndustryAdditional context for enrichment
LocationGeographic information

For person data:

ColumnPurpose
Full nameRequired as primary identifier
Company nameAssociates person with company
Job titleRole information
EmailContact information
LinkedIn URLProfile matching

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

Complete Example

Here's a workflow for uploading a file and using it in an ingest task:

import requests
import time
 
API_KEY = "your-api-key"
BASE_URL = "https://api.linkt.ai/v1"
HEADERS = {"x-api-key": API_KEY}
 
def upload_and_process(file_path):
    """Upload a file and wait for processing."""
 
    # Step 1: Upload file
    print(f"Uploading {file_path}...")
    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"  File ID: {file_id}")
    print(f"  Status: {file_data['processing_status']}")
 
    # Step 2: Wait for processing (if needed)
    if file_data["processing_status"] != "completed":
        print("Waiting for processing...")
        while True:
            response = requests.get(
                f"{BASE_URL}/files/{file_id}",
                headers=HEADERS
            )
            file_data = response.json()
            status = file_data["processing_status"]
 
            if status == "completed":
                break
            elif status == "failed":
                raise Exception("File processing failed")
 
            time.sleep(2)
 
    # Step 3: Verify CSV metadata
    metadata = file_data["csv_metadata"]
    print(f"  Rows: {metadata['row_count']}")
    print(f"  Columns: {metadata['columns']}")
 
    return file_data
 
def list_my_files():
    """List all uploaded files."""
    response = requests.get(
        f"{BASE_URL}/files",
        headers=HEADERS,
        params={"page": 1, "page_size": 50}
    )
 
    data = response.json()
    print(f"Total files: {data['total']}")
 
    for file in data["files"]:
        print(f"  {file['name']} ({file['file_id']})")
 
    return data["files"]
 
# Upload a file
file_data = upload_and_process("companies.csv")
 
# List all files
files = list_my_files()

Next Steps

On this page