Webhooks

Receive instant HTTP POST notifications when your workflows complete. Webhooks enable real-time integrations with CRMs, Slack, data pipelines, and other external systems without polling the API.

Overview

When a workflow run completes, Linkt can send an HTTP POST request to a URL you specify. The webhook payload contains details about the run including timing, credits used, and IDs of created resources.

Supported Workflows

Webhooks are available for all workflow types:

Workflow	Task Type	Use Case
Search	`search`	Discover new companies and contacts matching your ICP
Ingest	`ingest`	Import and enrich entities from CSV files
Signal	`signal-topic`, `signal-csv`, `signal-sheet`	Monitor for business signals and events

Use Cases

CRM Integration — Automatically sync discovered entities or detected signals to your CRM
Slack/Teams Alerts — Notify your team when workflows complete or signals are detected
Data Pipelines — Trigger downstream processing when new entities are enriched
Custom Dashboards — Push real-time updates to monitoring systems

Configuring Webhooks

Add the webhook_url field to your task configuration. The URL must use HTTPS.

Search Task with Webhook

Receive a notification when a search workflow completes:

curl -X POST "https://api.linkt.ai/v1/task" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Enterprise SaaS Discovery",
    "flow_name": "search",
    "deployment_name": "main",
    "icp_id": "507f1f77bcf86cd799439001",
    "task_config": {
      "type": "search",
      "desired_contact_count": 3,
      "user_feedback": "",
      "webhook_url": "https://your-server.com/webhooks/linkt"
    }
  }'

Ingest Task with Webhook

Receive a notification when CSV enrichment completes:

curl -X POST "https://api.linkt.ai/v1/task" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Q1 Leads Enrichment",
    "flow_name": "ingest",
    "deployment_name": "main",
    "icp_id": "507f1f77bcf86cd799439001",
    "task_config": {
      "type": "ingest",
      "file_id": "507f1f77bcf86cd799439030",
      "primary_column": "company_name",
      "csv_entity_type": "company",
      "webhook_url": "https://your-server.com/webhooks/linkt"
    }
  }'

Signal Task with Webhook

Receive a notification when signal monitoring completes:

curl -X POST "https://api.linkt.ai/v1/task" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "AI SaaS Signal Monitor",
    "flow_name": "signal",
    "deployment_name": "main",
    "icp_id": "507f1f77bcf86cd799439001",
    "task_config": {
      "type": "signal-topic",
      "entity_type": "company",
      "topic_criteria": "AI/ML-focused SaaS companies with Series A+ funding",
      "signal_types": [
        {
          "type": "funding",
          "display": "Funding Rounds",
          "description": "New funding announcements"
        }
      ],
      "monitoring_frequency": "daily",
      "webhook_url": "https://your-server.com/webhooks/linkt"
    }
  }'

URL Requirements

Requirement	Description
HTTPS Required	Webhook URLs must use `https://`
No Localhost	Cannot use `localhost`, `127.0.0.1`, or `::1`
Public Endpoint	URL must be publicly accessible
30s Timeout	Endpoint must respond within 30 seconds

HTTPS Only

Webhook URLs must use HTTPS for security. HTTP URLs will be rejected when creating the task.

Event Types

Each workflow type has its own set of event types:

Search Events

Event Type	Description
`run.search.completed`	Search workflow completed successfully
`run.search.failed`	Search workflow failed with an error
`run.search.crashed`	Search workflow crashed unexpectedly
`run.search.cancelled`	Search workflow was cancelled

Ingest Events

Event Type	Description
`run.ingest.completed`	Ingest workflow completed successfully
`run.ingest.failed`	Ingest workflow failed with an error
`run.ingest.crashed`	Ingest workflow crashed unexpectedly
`run.ingest.cancelled`	Ingest workflow was cancelled

Signal Events

Event Type	Description
`run.signal.completed`	Signal workflow completed successfully
`run.signal.failed`	Signal workflow failed with an error
`run.signal.crashed`	Signal workflow crashed unexpectedly
`run.signal.cancelled`	Signal workflow was cancelled

Webhook Payload

When a workflow completes, Linkt sends a POST request with a JSON payload. The payload structure is consistent across all workflow types, with some flow-specific fields.

Common Payload Structure

{
  "event_type": "run.search.completed",
  "timestamp": "2025-01-07T14:30:00.000000+00:00",
  "data": {
    "run_id": "507f1f77bcf86cd799439003",
    "run_name": "search Run",
    "icp_name": "Enterprise SaaS Companies",
    "icp_id": "507f1f77bcf86cd799439001",
    "user_email": "user@example.com",
    "user_first_name": "Jane",
    "started_at": "2025-01-07T14:00:00.000000+00:00",
    "ended_at": "2025-01-07T14:30:00.000000+00:00",
    "duration_seconds": 1800.0,
    "duration_formatted": "30 minutes",
    "credits_used": 75.0,
    "error_message": null,
    "resources": {
      "entities_created": [
        "507f1f77bcf86cd799439020",
        "507f1f77bcf86cd799439021"
      ],
      "entities_updated": [],
      "signals_created": []
    }
  }
}

Common Fields

Envelope Fields

Field	Type	Description
`event_type`	string	Event type (e.g., `run.search.completed`)
`timestamp`	string	ISO 8601 timestamp when event was generated
`data`	object	Event payload data

Data Fields

Field	Type	Description
`run_id`	string	Unique run identifier
`run_name`	string	Run display name
`icp_name`	string	Associated ICP name
`icp_id`	string	Associated ICP ID
`user_email`	string	User who created the task
`user_first_name`	string	User's first name
`started_at`	string	Run start timestamp (ISO 8601)
`ended_at`	string	Run end timestamp (ISO 8601)
`duration_seconds`	float	Total duration in seconds
`duration_formatted`	string	Human-readable duration (e.g., "9 minutes")
`credits_used`	float	Credits consumed by the run
`error_message`	string	Error details (for failed runs, otherwise `null`)
`resources`	object	Created/updated resource IDs

Resources Field

Field	Type	Description
`entities_created`	array	IDs of newly created entities
`entities_updated`	array	IDs of updated entities
`signals_created`	array	IDs of newly created signals (signal workflows only)

Search Workflow Payload

Search workflows create and update entities:

{
  "event_type": "run.search.completed",
  "timestamp": "2025-01-07T14:30:00.000000+00:00",
  "data": {
    "run_id": "507f1f77bcf86cd799439003",
    "run_name": "search Run",
    "icp_name": "Enterprise SaaS Companies",
    "icp_id": "507f1f77bcf86cd799439001",
    "user_email": "user@example.com",
    "user_first_name": "Jane",
    "started_at": "2025-01-07T14:00:00.000000+00:00",
    "ended_at": "2025-01-07T14:30:00.000000+00:00",
    "duration_seconds": 1800.0,
    "duration_formatted": "30 minutes",
    "credits_used": 75.0,
    "error_message": null,
    "resources": {
      "entities_created": [
        "507f1f77bcf86cd799439020",
        "507f1f77bcf86cd799439021",
        "507f1f77bcf86cd799439022"
      ],
      "entities_updated": [],
      "signals_created": []
    }
  }
}

Use the entity IDs in resources.entities_created to fetch full entity details via the API.

Ingest Workflow Payload

Ingest workflows primarily update existing entities:

{
  "event_type": "run.ingest.completed",
  "timestamp": "2025-01-07T14:30:00.000000+00:00",
  "data": {
    "run_id": "507f1f77bcf86cd799439003",
    "run_name": "ingest Run",
    "icp_name": "Q1 Leads Import",
    "icp_id": "507f1f77bcf86cd799439001",
    "user_email": "user@example.com",
    "user_first_name": "Jane",
    "started_at": "2025-01-07T14:00:00.000000+00:00",
    "ended_at": "2025-01-07T14:15:00.000000+00:00",
    "duration_seconds": 900.0,
    "duration_formatted": "15 minutes",
    "credits_used": 50.0,
    "error_message": null,
    "resources": {
      "entities_created": [
        "507f1f77bcf86cd799439030"
      ],
      "entities_updated": [
        "507f1f77bcf86cd799439031",
        "507f1f77bcf86cd799439032"
      ],
      "signals_created": []
    }
  }
}

Signal Workflow Payload

Signal workflows include additional signal-specific fields:

{
  "event_type": "run.signal.completed",
  "timestamp": "2025-01-07T14:30:00.000000+00:00",
  "data": {
    "run_id": "507f1f77bcf86cd799439003",
    "run_name": "signal Run",
    "icp_name": "AI SaaS Companies",
    "icp_id": "507f1f77bcf86cd799439001",
    "user_email": "user@example.com",
    "user_first_name": "Jane",
    "started_at": "2025-01-07T14:00:00.000000+00:00",
    "ended_at": "2025-01-07T14:30:00.000000+00:00",
    "duration_seconds": 1800.0,
    "duration_formatted": "30 minutes",
    "credits_used": 50.0,
    "error_message": null,
    "resources": {
      "entities_created": [],
      "entities_updated": [],
      "signals_created": [
        "507f1f77bcf86cd799439010",
        "507f1f77bcf86cd799439011",
        "507f1f77bcf86cd799439012"
      ]
    },
    "total_signals": 12,
    "signal_breakdown": {
      "funding": 5,
      "product_launch": 4,
      "leadership_change": 3
    }
  }
}

Signal-Specific Fields

Field	Type	Description
`total_signals`	integer	Total signals detected
`signal_breakdown`	object	Signal counts by type

Building Webhook Endpoints

Your webhook endpoint should accept POST requests and respond quickly.

Requirements

Accept POST requests with Content-Type: application/json
Respond with 2xx status within 30 seconds
Handle retries idempotently (same webhook may be delivered multiple times)

Python Example (Flask)

from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
@app.route('/webhooks/linkt', methods=['POST'])
def handle_webhook():
    """Handle Linkt webhook notifications."""
    payload = request.get_json()
    event_type = payload.get('event_type')
    data = payload.get('data', {})
 
    # Log the event
    print(f"Received {event_type}")
    print(f"  Run: {data.get('run_id')}")
    print(f"  Duration: {data.get('duration_formatted')}")
 
    # Route by workflow type
    if event_type.startswith('run.search.'):
        handle_search_event(event_type, data)
    elif event_type.startswith('run.ingest.'):
        handle_ingest_event(event_type, data)
    elif event_type.startswith('run.signal.'):
        handle_signal_event(event_type, data)
 
    # Respond quickly - do heavy processing async
    return jsonify({'status': 'received'}), 200
 
def handle_search_event(event_type, data):
    """Process search workflow events."""
    if event_type == 'run.search.completed':
        entity_ids = data.get('resources', {}).get('entities_created', [])
        print(f"Search discovered {len(entity_ids)} entities")
        # Sync to CRM, update dashboard, etc.
 
def handle_ingest_event(event_type, data):
    """Process ingest workflow events."""
    if event_type == 'run.ingest.completed':
        created = len(data.get('resources', {}).get('entities_created', []))
        updated = len(data.get('resources', {}).get('entities_updated', []))
        print(f"Ingest: {created} created, {updated} updated")
 
def handle_signal_event(event_type, data):
    """Process signal workflow events."""
    if event_type == 'run.signal.completed':
        total = data.get('total_signals', 0)
        breakdown = data.get('signal_breakdown', {})
        print(f"Detected {total} signals: {breakdown}")
        # Send Slack notification, update CRM, etc.
 
if __name__ == '__main__':
    app.run(port=5000)

Node.js Example (Express)

const express = require('express');
const app = express();
 
app.use(express.json());
 
app.post('/webhooks/linkt', (req, res) => {
  const { event_type, data, timestamp } = req.body;
 
  console.log(`Received ${event_type} at ${timestamp}`);
  console.log(`  Run: ${data.run_id}`);
  console.log(`  Duration: ${data.duration_formatted}`);
 
  // Route by workflow type
  if (event_type.startsWith('run.search.')) {
    handleSearchEvent(event_type, data);
  } else if (event_type.startsWith('run.ingest.')) {
    handleIngestEvent(event_type, data);
  } else if (event_type.startsWith('run.signal.')) {
    handleSignalEvent(event_type, data);
  }
 
  // Respond immediately
  res.status(200).json({ status: 'received' });
});
 
function handleSearchEvent(eventType, data) {
  if (eventType === 'run.search.completed') {
    const entityIds = data.resources?.entities_created || [];
    console.log(`Search discovered ${entityIds.length} entities`);
  }
}
 
function handleIngestEvent(eventType, data) {
  if (eventType === 'run.ingest.completed') {
    const created = (data.resources?.entities_created || []).length;
    const updated = (data.resources?.entities_updated || []).length;
    console.log(`Ingest: ${created} created, ${updated} updated`);
  }
}
 
function handleSignalEvent(eventType, data) {
  if (eventType === 'run.signal.completed') {
    console.log(`Detected ${data.total_signals} signals`);
    console.log(`  Breakdown: ${JSON.stringify(data.signal_breakdown)}`);
  }
}
 
app.listen(5000);

Retry Behavior

If your webhook endpoint is unavailable or returns an error, Linkt automatically retries delivery.

Retry Policy

Attempt	Delay	Cumulative Time
1 (initial)	—	0s
2	2s	2s
3	4s	6s
4	8s	14s
5	16s	30s
6	32s	62s

Retriable Errors

Error Type	Retried?	Description
5xx Server Errors	Yes	Server errors (500, 502, 503, etc.)
Timeout	Yes	No response within 30 seconds
Connection Error	Yes	Failed to connect to endpoint
4xx Client Errors	No	Bad request, unauthorized, not found

Webhook Failures Don't Block Workflows

Webhook delivery failures are logged but do not affect workflow completion. Your entities and signals are still created and available via the API even if the webhook fails.

Handling Duplicate Deliveries

Due to retries, your endpoint may receive the same webhook multiple times. Implement idempotent handling:

# Track processed webhooks by run_id + event_type
processed_events = set()
 
def handle_webhook(payload):
    event_key = f"{payload['data']['run_id']}:{payload['event_type']}"
 
    if event_key in processed_events:
        print(f"Skipping duplicate: {event_key}")
        return
 
    processed_events.add(event_key)
    # Process the webhook...

Best Practices

Respond Quickly

Return a 2xx response immediately and process webhooks asynchronously:

from celery import Celery
 
celery = Celery('tasks')
 
@app.route('/webhooks/linkt', methods=['POST'])
def webhook():
    payload = request.get_json()
 
    # Queue for async processing
    process_webhook.delay(payload)
 
    # Respond immediately
    return '', 200
 
@celery.task
def process_webhook(payload):
    # Heavy processing here
    pass

Log Everything

Log webhook payloads for debugging:

import logging
import json
 
logger = logging.getLogger(__name__)
 
@app.route('/webhooks/linkt', methods=['POST'])
def webhook():
    payload = request.get_json()
 
    logger.info(f"Webhook received: {json.dumps(payload)}")
 
    # Process...

Validate Payloads

Verify required fields exist before processing:

def validate_payload(payload):
    required = ['event_type', 'timestamp', 'data']
    for field in required:
        if field not in payload:
            raise ValueError(f"Missing required field: {field}")
 
    data = payload['data']
    if 'run_id' not in data:
        raise ValueError("Missing run_id in data")

Troubleshooting

Webhook Not Received

Check URL is HTTPS — HTTP URLs are rejected
Verify endpoint is public — Linkt cannot reach private networks
Check firewall rules — Allow incoming POST requests
Review server logs — Look for incoming requests

Receiving Duplicate Webhooks

Duplicates occur during retries. Implement idempotent handling using run_id as a deduplication key.

Timeouts

If your endpoint takes longer than 30 seconds:

Respond immediately with 200
Process webhook data asynchronously
Use a task queue (Celery, RQ, Bull, etc.)

Next Steps

Tasks — Task configuration reference
Runs — Understanding run lifecycle
Execution — Monitoring workflow execution
Search Guides — Discovery workflow tutorials
Ingest Guides — CSV import tutorials
Signal Guides — Signal monitoring tutorials

On this page