> ## Documentation Index
> Fetch the complete documentation index at: https://docs.edges.run/llms.txt
> Use this file to discover all available pages before exploring further.
# Build a LinkedIn Outreach Sequence
> Complete guide to building automated LinkedIn outreach sequences with de-duplication, error handling, and production-ready patterns.
This guide walks you through building a production-ready LinkedIn outreach sequence using the Edges API. You'll learn how to:
* Set up identities and connect LinkedIn accounts
* Source and manage leads
* Send connection requests and messages at scale
* Handle replies, de-duplication, and errors
* Respect rate limits and best practices
**Recommended setup:** Use [Engagement Identities](/v1/identities/engagement) for outreach actions. They're credit-free for connection requests, messages, and profile visits.
***
## Section 1: Prerequisites & Key Concepts
Before building your sequence, familiarize yourself with these core concepts:
| Concept | What it is | Documentation |
| ----------------------- | --------------------------------------------------------------------- | --------------------------------------------------------------- |
| **Workspace** | Isolated environment with its own API key and identities | [Core Concepts](/v1/core-concepts) |
| **Identity** | Represents a user whose LinkedIn account will perform actions | [Add & Manage Identities](/v1/identities/add-manage-identities) |
| **Engagement Identity** | Special identity type with credit-free outreach actions | [Engagement Identities](/v1/identities/engagement) |
| **Identity Modes** | `direct`, `auto`, `managed` - controls which account performs actions | [Identity Modes](/v1/identities/modes) |
| **Smart Limits** | Per-action, per-identity daily limits to protect accounts | [LinkedIn Smart Limits](/v1/linkedin/limits) |
| **Execution Modes** | `live` (sync), `async` (background), `schedule` (recurring) | [Execute Actions](/v1/runs/overview) |
| **Callbacks** | Webhook delivery for async/scheduled action results | [Callbacks](/v1/runs/callbacks) |
### Identity Setup Checklist
Use the [Create Identity API](/v1/api/identities/create) with `is_engagement: true`
Use [Connect Integration](/v1/api/integrations/connect) with cookies, email/password, or login links
Monitor integration status changes like `AUTH_EXPIRED`. See [LinkedIn Integration Webhooks](/v1/identities/linkedin#monitor-linkedin-integration-status-with-webhooks)
***
## Section 2: Lead Sourcing
Leads can come from Edges search actions or external sources like CSV uploads.
### Option A: Edges Search
```javascript JavaScript theme={null}
// Search for leads using LinkedIn search
// 1. Build your search URL on LinkedIn first (e.g., filter by title, location, company)
// 2. Copy the search URL from your browser
const searchUrl = 'https://www.linkedin.com/search/results/people/?keywords=CTO&geoUrn=%5B103644278%5D&origin=FACETED_SEARCH';
const searchResults = await fetch('https://api.edges.run/v1/actions/linkedin-search-people/run/live', {
method: 'POST',
headers: {
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
body: JSON.stringify({
identity_mode: 'auto',
input: {
linkedin_people_search_url: searchUrl
}
})
});
const leads = await searchResults.json();
// Each result contains: linkedin_profile_url, full_name, headline, etc.
```
```python Python theme={null}
import requests
# Search for leads using LinkedIn search
# 1. Build your search URL on LinkedIn first (e.g., filter by title, location, company)
# 2. Copy the search URL from your browser
search_url = 'https://www.linkedin.com/search/results/people/?keywords=CTO&geoUrn=%5B103644278%5D&origin=FACETED_SEARCH'
response = requests.post(
'https://api.edges.run/v1/actions/linkedin-search-people/run/live',
headers={
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
json={
'identity_mode': 'auto',
'input': {
'linkedin_people_search_url': search_url
}
}
)
leads = response.json()
# Each result contains: linkedin_profile_url, full_name, headline, etc.
```
**How it works:** Build your search filters on LinkedIn (title, location, company, etc.), then copy the search URL and pass it to the API. See [Search LinkedIn People](/v1/api/actions/linkedin-search-people) for details.
**Have Sales Navigator?** Use [salesnavigator-search-people](/v1/api/actions/salesnavigator-search-people) instead for access to advanced filters (seniority, function, company headcount growth, etc.). Works the same way — build your search on LinkedIn Sales Navigator and pass the URL.
### Option B: External Sources
Import leads from your CRM, CSV, or other sources. Required fields:
* `linkedin_profile_url` (required) - e.g., `https://www.linkedin.com/in/johndoe`
* `full_name` (recommended for personalization)
* `company_name` (recommended for personalization)
**Best practice:** Store the immutable `linkedin_profile_id` once you have it. Profile URLs can change if users update their vanity URL.
***
## Section 3: Get User's LinkedIn Profile ID
The user's `linkedin_profile_id` is needed to detect replies (comparing who sent the last message).
**Good news:** The `linkedin_profile_id` is already available in the integration's `meta` field when you retrieve the integration details. No extra API call needed!
```javascript JavaScript theme={null}
// Get linkedin_profile_id from integration meta (recommended)
async function getIntegrationProfileId(identityId, integration = 'linkedin') {
const response = await fetch(
`https://api.edges.run/v1/identities/${identityId}/integrations/${integration}`,
{
headers: { 'X-API-Key': EDGES_API_KEY }
}
);
const data = await response.json();
return data.meta?.linkedin_profile_id; // Already available!
}
// Cache this when identity is created/connected
const userProfileId = await getIntegrationProfileId('identity_abc123');
```
```python Python theme={null}
import requests
def get_integration_profile_id(identity_id: str, integration: str = 'linkedin') -> int:
"""Get linkedin_profile_id from integration meta (recommended)."""
response = requests.get(
f'https://api.edges.run/v1/identities/{identity_id}/integrations/{integration}',
headers={'X-API-Key': EDGES_API_KEY}
)
data = response.json()
return data.get('meta', {}).get('linkedin_profile_id') # Already available!
# Cache this when identity is created/connected
user_profile_id = get_integration_profile_id('identity_abc123')
```
See [Get an Identity's Integration](/v1/api/integrations/get) for full response schema.
**Alternative:** If you need additional profile details (headline, company, etc.), you can use [linkedin-me](/v1/api/actions/linkedin-me) which returns full profile information.
***
## Section 4: Database Schema (Pseudocode)
Track leads, conversations, and outreach history:
```sql theme={null}
-- Leads table
CREATE TABLE leads (
id UUID PRIMARY KEY,
identity_id VARCHAR NOT NULL, -- Which identity owns this lead
linkedin_profile_url VARCHAR NOT NULL,
linkedin_profile_id BIGINT, -- Immutable ID (populate when known)
full_name VARCHAR,
company_name VARCHAR,
location VARCHAR, -- For timezone/business hours
-- Sequence state
sequence_status VARCHAR DEFAULT 'NEW', -- NEW, IN_SEQUENCE, PAUSED, REPLIED, ARCHIVED
current_step INT DEFAULT 0,
next_action_at TIMESTAMP,
-- Connection state
connection_status VARCHAR DEFAULT 'NOT_CONNECTED', -- NOT_CONNECTED, PENDING, CONNECTED
connection_sent_at TIMESTAMP,
-- Tracking
last_contacted_at TIMESTAMP,
replied_at TIMESTAMP,
paused_reason VARCHAR,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Conversations cache (sync from extract-conversations)
CREATE TABLE conversations (
id UUID PRIMARY KEY,
identity_id VARCHAR NOT NULL,
linkedin_thread_id VARCHAR NOT NULL UNIQUE,
participant_profile_id BIGINT,
last_message_sender_id BIGINT,
last_message_at TIMESTAMP,
synced_at TIMESTAMP DEFAULT NOW()
);
-- Outreach log (audit trail)
CREATE TABLE outreach_log (
id UUID PRIMARY KEY,
lead_id UUID REFERENCES leads(id),
action_type VARCHAR NOT NULL, -- 'VISIT', 'CONNECT', 'MESSAGE', etc.
step_number INT,
status VARCHAR, -- 'SUCCESS', 'FAILED', 'SKIPPED'
error_label VARCHAR, -- e.g., 'LIMIT_REACHED', 'NOT_CONNECTED'
edges_run_id VARCHAR,
created_at TIMESTAMP DEFAULT NOW()
);
-- Indexes
CREATE INDEX idx_leads_next_action ON leads(identity_id, sequence_status, next_action_at);
CREATE INDEX idx_conversations_participant ON conversations(identity_id, participant_profile_id);
```
***
## Section 5: Lead Lifecycle State Machine
Visualize how leads progress through your sequence:
### Part A: Main Outreach Flow
The happy path from new lead to reply:
```mermaid theme={null}
stateDiagram-v2
[*] --> New: Lead added
New --> InSequence: Start sequence
InSequence --> PendingConnection: Connect sent
PendingConnection --> Connected: Accepted
Connected --> ReadyToMessage: Wait 3 days
ReadyToMessage --> WaitingReply: Message sent
WaitingReply --> FollowUp1: 7 days, no reply
FollowUp1 --> WaitingReply: Follow-up sent
WaitingReply --> FollowUp2: 7 more days
FollowUp2 --> WaitingReply: Follow-up 2 sent
WaitingReply --> Replied: Lead responds
WaitingReply --> Archived: No response
Replied --> [*]: Human takes over
Archived --> [*]: Sequence ended
```
### Part B: Edge Cases (Already Connected / Existing Conversation)
Handle leads who are already in your network or have messaged first:
```mermaid theme={null}
stateDiagram-v2
[*] --> New: Lead added
New --> CheckConnection: Check if connected
CheckConnection --> AlreadyConnected: Is connected
CheckConnection --> InSequence: Not connected
New --> LeadMessagedFirst: Has existing conversation
LeadMessagedFirst --> Replied: Human takes over
AlreadyConnected --> ReadyToMessage: No recent messages
AlreadyConnected --> WaitCooldown: Messaged in last 3 days
WaitCooldown --> ReadyToMessage: Cooldown passed
ReadyToMessage --> WaitingReply: Message sent
```
### Part C: Identity Failure Recovery
When LinkedIn auth expires, pause affected leads and resume when fixed:
```mermaid theme={null}
stateDiagram-v2
WaitingReply --> Paused: AUTH_EXPIRED
PendingConnection --> Paused: AUTH_EXPIRED
ReadyToMessage --> Paused: AUTH_EXPIRED
Paused --> WaitingReply: Identity restored
Paused --> ReadyToMessage: Identity restored
Paused --> PendingConnection: Identity restored
```
### Edge Cases Summary
| Scenario | Detection | Action |
| ------------------------------ | -------------------------------------------- | ---------------------------------- |
| **Lead already connected** | `extract-connections` contains lead | Skip connect, go to ReadyToMessage |
| **Recent conversation exists** | `last_message` within 3 days | Wait for cooldown before messaging |
| **Lead messaged first** | `extract-conversations` shows lead initiated | Mark as REPLIED, human takes over |
| **Identity auth expired** | Webhook `AUTH_EXPIRED` event | PAUSE all leads for that identity |
***
## Section 6: Action Limits Quick Reference
Each action in the sequence has specific daily limits. Link to the action doc for full response schema.
| Step | Action | Limit (Classic / Sales Nav) | Action Doc |
| --------------------- | -------------------------------- | --------------------------- | -------------------------------------------------------- |
| Visit | `linkedin-visit-profile` | 80 / 500 per day | [View →](/v1/api/actions/linkedin-visit-profile) |
| Connect | `linkedin-connect-profile` | 25 / 30 per day | [View →](/v1/api/actions/linkedin-connect-profile) |
| Message | `linkedin-message-profile` | 50 / 250 per day | [View →](/v1/api/actions/linkedin-message-profile) |
| InMail | `linkedin-inmail-profile` | Based on subscription | [View →](/v1/api/actions/linkedin-inmail-profile) |
| Extract Conversations | `linkedin-extract-conversations` | Free with Engagement | [View →](/v1/api/actions/linkedin-extract-conversations) |
| Extract Connections | `linkedin-extract-connections` | 30,000 per day | [View →](/v1/api/actions/linkedin-extract-connections) |
**Connection notes limit:** LinkedIn Classic accounts can only send **5 connection requests WITH a personalized note per month**. After that, notes are silently dropped. Consider: connect without note, then message after they accept.
**Full limits reference:** [LinkedIn Smart Limits →](/v1/linkedin/limits)
***
## Section 7: Error Labels for Outreach Actions
Each action returns specific error labels. Handle these in your code.
| Error Label | Actions Affected | Meaning | How to Handle |
| ------------------------ | ----------------- | ---------------------------------------- | ---------------------------------------------------------------- |
| `LIMIT_REACHED` | All | Daily Smart Limit hit | Schedule retry for tomorrow, use `postponed_until` from response |
| `NOT_CONNECTED` | `message-profile` | Can't message non-connection | Send connection request first |
| `ALREADY_CONNECTED` | `connect-profile` | Already 1st degree connection | Skip to message step |
| `INVITATION_PENDING` | `connect-profile` | Connection request already sent | Wait for accept or withdraw |
| `PROFILE_NOT_ACCESSIBLE` | All | Profile deleted, blocked, or URL changed | Remove from sequence or re-enrich |
| `AUTH_EXPIRED` | All | Identity LinkedIn session expired | Pause sequence, trigger re-auth webhook |
| `INVALID_INPUT` | All | Bad parameters (wrong URL format, etc.) | Fix input, check against action docs |
| `STATUS_429` | All | LinkedIn raw rate limit (not guarded) | Exponential backoff, wait and retry |
**How to handle in code:**
```javascript JavaScript theme={null}
async function executeWithErrorHandling(identityId, lead, actionFn) {
try {
const result = await actionFn();
return { success: true, result };
} catch (error) {
const errorLabel = error.error_label || 'UNKNOWN';
switch (errorLabel) {
case 'LIMIT_REACHED':
// Retry tomorrow - use postponed_until if available
const retryAt = error.postponed_until || tomorrow();
await updateLead(lead.id, { next_action_at: retryAt });
break;
case 'ALREADY_CONNECTED':
// Skip connect step, go to message
await updateLead(lead.id, {
connection_status: 'CONNECTED',
current_step: 3 // Skip to message step
});
break;
case 'INVITATION_PENDING':
// Already sent, just wait
await updateLead(lead.id, { connection_status: 'PENDING' });
break;
case 'NOT_CONNECTED':
// Can't message - need to connect first
await updateLead(lead.id, { current_step: 1 }); // Go to connect step
break;
case 'PROFILE_NOT_ACCESSIBLE':
// Remove from sequence
await updateLead(lead.id, {
sequence_status: 'ARCHIVED',
paused_reason: 'PROFILE_INACCESSIBLE'
});
break;
case 'AUTH_EXPIRED':
// Pause all leads for this identity
await pauseIdentityLeads(identityId, 'AUTH_EXPIRED');
break;
default:
// Log and investigate
console.error('Unexpected error:', error);
await logOutreachError(lead.id, error);
}
return { success: false, errorLabel };
}
}
function tomorrow() {
const date = new Date();
date.setDate(date.getDate() + 1);
date.setHours(9, 0, 0, 0); // 9am tomorrow
return date.toISOString();
}
```
```python Python theme={null}
from datetime import datetime, timedelta
async def execute_with_error_handling(identity_id: str, lead, action_fn):
try:
result = await action_fn()
return {'success': True, 'result': result}
except EdgesError as error:
error_label = getattr(error, 'error_label', 'UNKNOWN')
if error_label == 'LIMIT_REACHED':
# Retry tomorrow - use postponed_until if available
retry_at = getattr(error, 'postponed_until', None) or tomorrow()
await update_lead(lead.id, next_action_at=retry_at)
elif error_label == 'ALREADY_CONNECTED':
# Skip connect step, go to message
await update_lead(lead.id, connection_status='CONNECTED', current_step=3)
elif error_label == 'INVITATION_PENDING':
# Already sent, just wait
await update_lead(lead.id, connection_status='PENDING')
elif error_label == 'NOT_CONNECTED':
# Can't message - need to connect first
await update_lead(lead.id, current_step=1)
elif error_label == 'PROFILE_NOT_ACCESSIBLE':
# Remove from sequence
await update_lead(lead.id,
sequence_status='ARCHIVED',
paused_reason='PROFILE_INACCESSIBLE'
)
elif error_label == 'AUTH_EXPIRED':
# Pause all leads for this identity
await pause_identity_leads(identity_id, 'AUTH_EXPIRED')
else:
# Log and investigate
print(f'Unexpected error: {error}')
await log_outreach_error(lead.id, error)
return {'success': False, 'error_label': error_label}
def tomorrow():
date = datetime.now() + timedelta(days=1)
return date.replace(hour=9, minute=0, second=0, microsecond=0).isoformat()
```
**Full error reference:** [Error Reference →](/v1/error-reference)
***
## Section 8: Optimized Sync Flow (last\_message optimization)
The `last_message` field in `extract-conversations` contains the sender's `linkedin_profile_id`. This lets you detect replies without calling `extract-messages`.
```javascript JavaScript theme={null}
// Sync conversations and detect replies efficiently
async function syncConversations(identityId) {
const userProfileId = await getUserProfileId(identityId);
const response = await fetch('https://api.edges.run/v1/actions/linkedin-extract-conversations/run/live', {
method: 'POST',
headers: {
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
body: JSON.stringify({
identity_ids: [identityId]
})
});
const data = await response.json();
for (const conv of data) {
const lastSenderId = conv.last_message?.linkedin_profile_id;
// If last message sender != user, the lead replied!
if (lastSenderId && lastSenderId !== userProfileId) {
const leadProfileId = conv.participants[0]?.linkedin_profile_id;
await markLeadAsReplied(identityId, leadProfileId);
// NO NEED to call extract-messages for reply detection!
}
// Cache conversation for de-dup
await upsertConversation({
identity_id: identityId,
linkedin_thread_id: conv.linkedin_thread_id,
participant_profile_id: conv.participants[0]?.linkedin_profile_id,
last_message_sender_id: lastSenderId,
last_message_at: conv.last_message?.delivered_at
});
}
}
```
```python Python theme={null}
async def sync_conversations(identity_id: str):
"""Sync conversations and detect replies efficiently."""
user_profile_id = await get_user_profile_id(identity_id)
response = requests.post(
'https://api.edges.run/v1/actions/linkedin-extract-conversations/run/live',
headers={
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
json={
'identity_ids': [identity_id]
}
)
data = response.json()
for conv in data:
last_sender_id = conv.get('last_message', {}).get('linkedin_profile_id')
# If last message sender != user, the lead replied!
if last_sender_id and last_sender_id != user_profile_id:
lead_profile_id = conv['participants'][0].get('linkedin_profile_id')
await mark_lead_as_replied(identity_id, lead_profile_id)
# NO NEED to call extract-messages for reply detection!
# Cache conversation for de-dup
await upsert_conversation(
identity_id=identity_id,
linkedin_thread_id=conv['linkedin_thread_id'],
participant_profile_id=conv['participants'][0].get('linkedin_profile_id'),
last_message_sender_id=last_sender_id,
last_message_at=conv.get('last_message', {}).get('delivered_at')
)
```
**When you DO need `extract-messages`:**
* **Building a Lemlist-like inbox UI** - Display full conversation threads to users in your app
* Full conversation history for CRM sync
* Message analytics/sentiment analysis
* Compliance/audit requirements
***
## Section 9: Sending Outreach Actions (Complete Examples)
### Send a Connection Request
```javascript JavaScript theme={null}
async function sendConnectionRequest(identityId, lead, message = null) {
const response = await fetch('https://api.edges.run/v1/actions/linkedin-connect-profile/run/live', {
method: 'POST',
headers: {
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
body: JSON.stringify({
identity_ids: [identityId],
parameters: {
message: message // Optional, 300 char limit. Only 5/month with note!
},
input: {
linkedin_profile_url: lead.linkedin_profile_url
}
})
});
if (!response.ok) {
const error = await response.json();
throw error;
}
const data = await response.json();
// Update lead status
await updateLead(lead.id, {
connection_status: 'PENDING',
connection_sent_at: new Date().toISOString(),
last_contacted_at: new Date().toISOString()
});
// Log the action
await logOutreach(lead.id, 'CONNECT', data.run_id);
return data;
}
```
```python Python theme={null}
async def send_connection_request(identity_id: str, lead, message: str = None):
response = requests.post(
'https://api.edges.run/v1/actions/linkedin-connect-profile/run/live',
headers={
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
json={
'identity_ids': [identity_id],
'parameters': {
'message': message # Optional, 300 char limit. Only 5/month with note!
},
'input': {
'linkedin_profile_url': lead.linkedin_profile_url
}
}
)
if not response.ok:
raise EdgesError(response.json())
data = response.json()
# Update lead status
await update_lead(lead.id,
connection_status='PENDING',
connection_sent_at=datetime.now().isoformat(),
last_contacted_at=datetime.now().isoformat()
)
# Log the action
await log_outreach(lead.id, 'CONNECT', data['run_id'])
return data
```
### Send a Message
```javascript JavaScript theme={null}
async function sendMessage(identityId, lead, messageText) {
const response = await fetch('https://api.edges.run/v1/actions/linkedin-message-profile/run/live', {
method: 'POST',
headers: {
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
body: JSON.stringify({
identity_ids: [identityId],
parameters: {
message: messageText
},
input: {
linkedin_profile_url: lead.linkedin_profile_url
}
})
});
if (!response.ok) {
const error = await response.json();
throw error;
}
const data = await response.json();
// Update lead status
await updateLead(lead.id, {
sequence_status: 'WAITING_REPLY',
last_contacted_at: new Date().toISOString()
});
// Log the action
await logOutreach(lead.id, 'MESSAGE', data.run_id);
return data;
}
```
```python Python theme={null}
async def send_message(identity_id: str, lead, message_text: str):
response = requests.post(
'https://api.edges.run/v1/actions/linkedin-message-profile/run/live',
headers={
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
json={
'identity_ids': [identity_id],
'parameters': {
'message': message_text
},
'input': {
'linkedin_profile_url': lead.linkedin_profile_url
}
}
)
if not response.ok:
raise EdgesError(response.json())
data = response.json()
# Update lead status
await update_lead(lead.id,
sequence_status='WAITING_REPLY',
last_contacted_at=datetime.now().isoformat()
)
# Log the action
await log_outreach(lead.id, 'MESSAGE', data['run_id'])
return data
```
***
## Section 10: De-duplication Logic
Before sending any outreach, check for existing conversations and pending requests.
```javascript JavaScript theme={null}
async function canSendOutreach(identityId, lead) {
// Check 1: Already replied?
if (lead.sequence_status === 'REPLIED') {
return { canSend: false, reason: 'ALREADY_REPLIED' };
}
// Check 2: Pending connection request?
if (lead.connection_status === 'PENDING') {
const daysSinceSent = daysBetween(lead.connection_sent_at, new Date());
if (daysSinceSent < 21) {
return { canSend: false, reason: 'CONNECTION_PENDING' };
}
}
// Check 3: Existing conversation? (from synced data)
const existingConv = await db.conversations.findFirst({
where: {
identity_id: identityId,
participant_profile_id: lead.linkedin_profile_id
}
});
if (existingConv) {
const userProfileId = await getUserProfileId(identityId);
// Lead replied - last message is from them
if (existingConv.last_message_sender_id !== userProfileId) {
await updateLead(lead.id, { sequence_status: 'REPLIED' });
return { canSend: false, reason: 'LEAD_REPLIED' };
}
// We already messaged, check cooldown
const daysSinceMessage = daysBetween(existingConv.last_message_at, new Date());
if (daysSinceMessage < 7) {
return { canSend: false, reason: 'COOLDOWN_ACTIVE' };
}
}
return { canSend: true };
}
function daysBetween(date1, date2) {
const diffMs = new Date(date2) - new Date(date1);
return Math.floor(diffMs / (1000 * 60 * 60 * 24));
}
```
```python Python theme={null}
from datetime import datetime
async def can_send_outreach(identity_id: str, lead):
# Check 1: Already replied?
if lead.sequence_status == 'REPLIED':
return {'can_send': False, 'reason': 'ALREADY_REPLIED'}
# Check 2: Pending connection request?
if lead.connection_status == 'PENDING':
days_since_sent = days_between(lead.connection_sent_at, datetime.now())
if days_since_sent < 21:
return {'can_send': False, 'reason': 'CONNECTION_PENDING'}
# Check 3: Existing conversation? (from synced data)
existing_conv = await db.conversations.find_first(
identity_id=identity_id,
participant_profile_id=lead.linkedin_profile_id
)
if existing_conv:
user_profile_id = await get_user_profile_id(identity_id)
# Lead replied - last message is from them
if existing_conv.last_message_sender_id != user_profile_id:
await update_lead(lead.id, sequence_status='REPLIED')
return {'can_send': False, 'reason': 'LEAD_REPLIED'}
# We already messaged, check cooldown
days_since_message = days_between(existing_conv.last_message_at, datetime.now())
if days_since_message < 7:
return {'can_send': False, 'reason': 'COOLDOWN_ACTIVE'}
return {'can_send': True}
def days_between(date1, date2):
diff = date2 - date1
return diff.days
```
***
## Section 11: Business Hours & Timezone Sending
Send messages during the lead's business hours for better response rates.
```javascript JavaScript theme={null}
function getSendTime(lead) {
// Default business hours: 9am - 6pm in lead's timezone
const targetHour = 9 + Math.floor(Math.random() * 9); // Random hour 9-17
const targetMinute = Math.floor(Math.random() * 60);
// Get lead's timezone offset (you'd lookup from location)
const leadTimezone = getTimezoneFromLocation(lead.location) || 'America/New_York';
const now = new Date();
const sendTime = new Date(now);
// Set to target time in lead's timezone
sendTime.setHours(targetHour, targetMinute, 0, 0);
// If it's past business hours today, schedule for tomorrow
const nowInLeadTz = new Date(now.toLocaleString('en-US', { timeZone: leadTimezone }));
if (nowInLeadTz.getHours() >= 18) {
sendTime.setDate(sendTime.getDate() + 1);
}
// Skip weekends
while (sendTime.getDay() === 0 || sendTime.getDay() === 6) {
sendTime.setDate(sendTime.getDate() + 1);
}
return sendTime;
}
// Helper: Map location to timezone (implement based on your data)
function getTimezoneFromLocation(location) {
const timezoneMap = {
'San Francisco': 'America/Los_Angeles',
'New York': 'America/New_York',
'London': 'Europe/London',
'Paris': 'Europe/Paris',
// Add more mappings
};
for (const [city, tz] of Object.entries(timezoneMap)) {
if (location?.includes(city)) return tz;
}
return null;
}
```
```python Python theme={null}
from datetime import datetime, timedelta
import random
def get_send_time(lead):
"""Calculate optimal send time in lead's business hours."""
# Default business hours: 9am - 6pm in lead's timezone
target_hour = 9 + random.randint(0, 8) # Random hour 9-17
target_minute = random.randint(0, 59)
# Get lead's timezone (you'd lookup from location)
lead_timezone = get_timezone_from_location(lead.location) or 'America/New_York'
now = datetime.now()
send_time = now.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
# If it's past business hours today, schedule for tomorrow
if now.hour >= 18:
send_time += timedelta(days=1)
# Skip weekends (0=Monday, 6=Sunday in Python)
while send_time.weekday() >= 5: # Saturday=5, Sunday=6
send_time += timedelta(days=1)
return send_time
def get_timezone_from_location(location: str) -> str:
"""Map location to timezone. Implement based on your data."""
timezone_map = {
'San Francisco': 'America/Los_Angeles',
'New York': 'America/New_York',
'London': 'Europe/London',
'Paris': 'Europe/Paris',
# Add more mappings
}
if not location:
return None
for city, tz in timezone_map.items():
if city in location:
return tz
return None
```
***
## Section 12: Callback Correlation with custom\_data
When using async execution, pass `custom_data` to correlate callbacks with your leads.
```javascript JavaScript theme={null}
// Send async action with custom_data
async function sendAsyncMessage(identityId, lead) {
const response = await fetch('https://api.edges.run/v1/actions/linkedin-message-profile/run/async', {
method: 'POST',
headers: {
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
body: JSON.stringify({
identity_ids: [identityId],
parameters: {
message: 'Hi! I wanted to connect...'
},
inputs: [{
linkedin_profile_url: lead.linkedin_profile_url,
custom_data: {
lead_id: lead.id,
step: 'initial_message',
identity_id: identityId
}
}],
callback: {
url: 'https://your-app.com/webhooks/edges'
}
})
});
return response.json();
}
// Handle callback in your webhook endpoint
app.post('/webhooks/edges', async (req, res) => {
const { run_id, status, output, custom_data } = req.body;
// custom_data contains exactly what you sent
const { lead_id, step, identity_id } = custom_data;
if (status === 'SUCCESS') {
await updateLead(lead_id, {
sequence_status: 'WAITING_REPLY',
last_contacted_at: new Date().toISOString()
});
} else {
await logOutreachError(lead_id, output.error_label);
}
res.status(200).send('OK');
});
```
```python Python theme={null}
import requests
from flask import Flask, request
# Send async action with custom_data
async def send_async_message(identity_id: str, lead):
response = requests.post(
'https://api.edges.run/v1/actions/linkedin-message-profile/run/async',
headers={
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
json={
'identity_ids': [identity_id],
'parameters': {
'message': 'Hi! I wanted to connect...'
},
'inputs': [{
'linkedin_profile_url': lead.linkedin_profile_url,
'custom_data': {
'lead_id': str(lead.id),
'step': 'initial_message',
'identity_id': identity_id
}
}],
'callback': {
'url': 'https://your-app.com/webhooks/edges'
}
}
)
return response.json()
# Handle callback in your webhook endpoint
app = Flask(__name__)
@app.route('/webhooks/edges', methods=['POST'])
async def handle_edges_callback():
data = request.json
run_id = data['run_id']
status = data['status']
output = data.get('output', {})
custom_data = data.get('custom_data', {})
# custom_data contains exactly what you sent
lead_id = custom_data.get('lead_id')
step = custom_data.get('step')
identity_id = custom_data.get('identity_id')
if status == 'SUCCESS':
await update_lead(lead_id,
sequence_status='WAITING_REPLY',
last_contacted_at=datetime.now().isoformat()
)
else:
await log_outreach_error(lead_id, output.get('error_label'))
return 'OK', 200
```
See [Callbacks Documentation →](/v1/runs/callbacks) for full callback payload structure.
***
## Section 13: Rate Limits & Scaling
Understand both types of limits:
### 1. LinkedIn Smart Limits (per identity, per action)
These protect individual LinkedIn accounts from restrictions:
* 25-30 connections/day
* 50-250 messages/day
* See [LinkedIn Smart Limits →](/v1/linkedin/limits)
### 2. API Rate Limits (per workspace)
Based on your Edges plan tier. See [API Rate Limits →](/v1/runs/rate-limits)
### Choosing the Right Execution Mode
Choose based on **when and how** you need the action executed:
| Mode | Use Case | Example |
| ---------- | -------------------------------------------------- | ------------------------------------- |
| `live` | Real-time user action, need immediate result | User clicks "Send Message" in your UI |
| `async` | Background bulk operations, process callback later | Import 500 leads and message them all |
| `schedule` | Recurring automation on a schedule | Daily sync of connections at 9am |
**Implementation stays the same regardless of user count**. Your architecture should support all modes from day 1 based on feature requirements, not scale.
**Best practice:** Space out requests. Instead of bursting 50 messages at once, spread them across the day using job queues or scheduled runs.
***
## Section 14: When to Sync Data
Knowing **when** to sync connections and conversations is critical for accurate outreach.
### Connections Extraction
| When | Why | Mode |
| ------------------------------------ | ------------------------------------- | ------------------------ |
| **After identity connects LinkedIn** | Initial cache of existing connections | `live` or `async` |
| **Daily (e.g., 9am)** | Catch newly accepted requests | `schedule` |
| **Before starting a new lead** | Check if already connected | Part of lead import flow |
```javascript JavaScript theme={null}
// Schedule daily connections sync (use /run/schedule)
// Or call this on a cron job using /run/async
async function syncConnections(identityId) {
const response = await fetch('https://api.edges.run/v1/actions/linkedin-extract-connections/run/live', {
method: 'POST',
headers: {
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
body: JSON.stringify({
identity_ids: [identityId]
})
});
const connections = await response.json();
// Update your connections cache
for (const conn of connections) {
await upsertConnection({
identity_id: identityId,
linkedin_profile_id: conn.linkedin_profile_id,
linkedin_profile_url: conn.linkedin_profile_url,
connected_at: conn.connected_at
});
}
return connections.length;
}
```
```python Python theme={null}
async def sync_connections(identity_id: str):
"""Sync connections - call daily or after identity setup."""
response = requests.post(
'https://api.edges.run/v1/actions/linkedin-extract-connections/run/live',
headers={
'X-API-Key': EDGES_API_KEY,
'Content-Type': 'application/json'
},
json={
'identity_ids': [identity_id]
}
)
connections = response.json()
# Update your connections cache
for conn in connections:
await upsert_connection(
identity_id=identity_id,
linkedin_profile_id=conn['linkedin_profile_id'],
linkedin_profile_url=conn['linkedin_profile_url'],
connected_at=conn.get('connected_at')
)
return len(connections)
```
### Conversations Extraction
| When | Why | Mode |
| ------------------------------ | --------------------------------------------- | ------------------ |
| **Before sending any message** | Check for replies (sync-before-send) | `live` |
| **Every 1-4 hours** | Update conversation cache, detect new replies | `schedule` or cron |
| **On demand (inbox UI)** | Show user their latest conversations | `live` |
**Schedule mode is ideal for recurring syncs.** Set up a schedule to run `extract-connections` daily and `extract-conversations` every few hours. See [Scheduled Runs →](/v1/runs/schedules)
***
## Section 15: Sync-Before-Send Pattern (Race Condition Prevention)
**Problem:** If you sync at 8:00am but send at 8:05am, lead could have replied at 8:03am.
**Solution:** Always sync immediately before sending:
```javascript JavaScript theme={null}
async function executeOutreachWithSync(identityId, lead) {
// Step 1: Fresh sync for this specific lead
// Note: extractConversations() wraps the API call from Section 8
const conversations = await extractConversations(identityId, { max_results: 50 });
// Step 2: Check if lead replied since last sync
const userProfileId = await getUserProfileId(identityId);
for (const conv of conversations) {
const participantId = conv.participants[0]?.linkedin_profile_id;
if (participantId === lead.linkedin_profile_id) {
if (conv.last_message?.linkedin_profile_id !== userProfileId) {
// Lead replied! Abort outreach
await updateLead(lead.id, { sequence_status: 'REPLIED' });
return { status: 'skipped', reason: 'LEAD_REPLIED_SINCE_SYNC' };
}
}
}
// Step 3: Safe to send
return await sendMessage(identityId, lead);
}
```
```python Python theme={null}
async def execute_outreach_with_sync(identity_id: str, lead):
"""Sync conversations immediately before sending to prevent race conditions."""
# Step 1: Fresh sync for this specific lead
# Note: extract_conversations() wraps the API call from Section 8
conversations = await extract_conversations(identity_id, max_results=50)
# Step 2: Check if lead replied since last sync
user_profile_id = await get_user_profile_id(identity_id)
for conv in conversations:
participant_id = conv['participants'][0].get('linkedin_profile_id')
if participant_id == lead.linkedin_profile_id:
last_sender = conv.get('last_message', {}).get('linkedin_profile_id')
if last_sender != user_profile_id:
# Lead replied! Abort outreach
await update_lead(lead.id, sequence_status='REPLIED')
return {'status': 'skipped', 'reason': 'LEAD_REPLIED_SINCE_SYNC'}
# Step 3: Safe to send
return await send_message(identity_id, lead)
```
**Acceptable staleness:** 5-10 minutes for most use cases. For high-value leads, sync immediately before.
***
## Section 16: Identity Failure Recovery
**Prerequisite:** Set up integration webhooks. See [Monitor LinkedIn Integration Status →](/v1/identities/linkedin#monitor-linkedin-integration-status-with-webhooks)
**When identity auth expires (webhook event: `AUTH_EXPIRED`):**
```javascript JavaScript theme={null}
// Webhook handler for identity status changes
app.post('/webhooks/edges/integration', async (req, res) => {
const { identity_uid, event_type, integration_type } = req.body;
if (integration_type !== 'linkedin') {
return res.status(200).send('OK');
}
switch (event_type) {
case 'AUTH_EXPIRED':
await handleIdentityAuthExpired(identity_uid);
break;
case 'AUTH_SUCCESS':
await handleIdentityAuthRestored(identity_uid);
break;
}
res.status(200).send('OK');
});
async function handleIdentityAuthExpired(identityUid) {
// 1. Pause all leads in active sequences for this identity
await db.leads.updateMany({
where: {
identity_id: identityUid,
sequence_status: { in: ['IN_SEQUENCE', 'WAITING_REPLY'] }
},
data: {
sequence_status: 'PAUSED',
paused_reason: 'IDENTITY_AUTH_EXPIRED',
paused_at: new Date()
}
});
// 2. Notify user to re-authenticate
const user = await getUserFromIdentity(identityUid);
await sendNotification(user.id, {
type: 'LINKEDIN_AUTH_EXPIRED',
message: 'Your LinkedIn connection expired. Please reconnect to resume sequences.',
action_url: '/settings/integrations'
});
}
async function handleIdentityAuthRestored(identityUid) {
// Resume paused leads
await db.leads.updateMany({
where: {
identity_id: identityUid,
sequence_status: 'PAUSED',
paused_reason: 'IDENTITY_AUTH_EXPIRED'
},
data: {
sequence_status: 'IN_SEQUENCE',
paused_reason: null,
next_action_at: new Date() // Reschedule immediately
}
});
}
```
```python Python theme={null}
from flask import Flask, request
app = Flask(__name__)
@app.route('/webhooks/edges/integration', methods=['POST'])
async def handle_integration_webhook():
data = request.json
identity_uid = data['identity_uid']
event_type = data['event_type']
integration_type = data['integration_type']
if integration_type != 'linkedin':
return 'OK', 200
if event_type == 'AUTH_EXPIRED':
await handle_identity_auth_expired(identity_uid)
elif event_type == 'AUTH_SUCCESS':
await handle_identity_auth_restored(identity_uid)
return 'OK', 200
async def handle_identity_auth_expired(identity_uid: str):
"""Pause all sequences when LinkedIn auth expires."""
# 1. Pause all leads in active sequences for this identity
await db.execute("""
UPDATE leads
SET sequence_status = 'PAUSED',
paused_reason = 'IDENTITY_AUTH_EXPIRED',
paused_at = NOW()
WHERE identity_id = %s
AND sequence_status IN ('IN_SEQUENCE', 'WAITING_REPLY')
""", [identity_uid])
# 2. Notify user to re-authenticate
user = await get_user_from_identity(identity_uid)
await send_notification(
user_id=user.id,
notification_type='LINKEDIN_AUTH_EXPIRED',
message='Your LinkedIn connection expired. Please reconnect to resume sequences.',
action_url='/settings/integrations'
)
async def handle_identity_auth_restored(identity_uid: str):
"""Resume paused leads when LinkedIn auth is restored."""
await db.execute("""
UPDATE leads
SET sequence_status = 'IN_SEQUENCE',
paused_reason = NULL,
next_action_at = NOW()
WHERE identity_id = %s
AND sequence_status = 'PAUSED'
AND paused_reason = 'IDENTITY_AUTH_EXPIRED'
""", [identity_uid])
```
***
## Section 17: Handling Edge Cases (Already Connected, Lead Messaged First)
When adding a lead to a sequence, check for existing relationships:
```javascript JavaScript theme={null}
async function addLeadToSequence(identityId, lead) {
// Step 1: Check if already connected
// Note: Helper functions wrap the API calls shown in previous sections
const connections = await extractConnections(identityId, { max_results: 1000 });
const isConnected = connections.some(
conn => conn.linkedin_profile_id === lead.linkedin_profile_id
);
if (isConnected) {
lead.connection_status = 'CONNECTED';
lead.current_step = 3; // Skip to message step
await saveLead(lead);
return;
}
// Step 2: Check if conversation exists (lead may have messaged first)
const conversations = await extractConversations(identityId, { max_results: 100 });
for (const conv of conversations) {
if (conv.participants[0]?.linkedin_profile_id === lead.linkedin_profile_id) {
// Existing conversation! Check who initiated
const userProfileId = await getUserProfileId(identityId);
if (conv.last_message?.linkedin_profile_id === lead.linkedin_profile_id) {
lead.sequence_status = 'REPLIED';
lead.replied_at = conv.last_message.delivered_at;
await saveLead(lead);
return; // Don't start sequence, human should handle
}
}
}
// Step 3: Normal start
lead.sequence_status = 'IN_SEQUENCE';
lead.current_step = 0;
lead.next_action_at = new Date();
await saveLead(lead);
}
```
```python Python theme={null}
async def add_lead_to_sequence(identity_id: str, lead):
"""Add a lead to sequence with edge case handling."""
# Step 1: Check if already connected
# Note: Helper functions wrap the API calls shown in previous sections
connections = await extract_connections(identity_id, max_results=1000)
is_connected = any(
conn['linkedin_profile_id'] == lead.linkedin_profile_id
for conn in connections
)
if is_connected:
lead.connection_status = 'CONNECTED'
lead.current_step = 3 # Skip to message step
await save_lead(lead)
return
# Step 2: Check if conversation exists (lead may have messaged first)
conversations = await extract_conversations(identity_id, max_results=100)
for conv in conversations:
participant_id = conv['participants'][0].get('linkedin_profile_id')
if participant_id == lead.linkedin_profile_id:
# Existing conversation! Check who initiated
user_profile_id = await get_user_profile_id(identity_id)
last_sender = conv.get('last_message', {}).get('linkedin_profile_id')
if last_sender == lead.linkedin_profile_id:
lead.sequence_status = 'REPLIED'
lead.replied_at = conv['last_message']['delivered_at']
await save_lead(lead)
return # Don't start sequence, human should handle
# Step 3: Normal start
lead.sequence_status = 'IN_SEQUENCE'
lead.current_step = 0
lead.next_action_at = datetime.now()
await save_lead(lead)
```
***
## Section 18: Complete Orchestrator Example
This ties everything together. A daily cron job that processes all leads for all identities.
```javascript JavaScript theme={null}
// Main orchestrator - runs daily via cron
async function runDailyOutreachForAllIdentities() {
// Get all active identities
const identities = await db.identities.findMany({
where: { status: 'ACTIVE', linkedin_connected: true }
});
for (const identity of identities) {
try {
await runOutreachForIdentity(identity.id);
} catch (error) {
console.error(`Failed for identity ${identity.id}:`, error);
// Continue with other identities
}
}
}
async function runOutreachForIdentity(identityId) {
// Step 1: Sync conversations to detect replies (Section 8)
await syncConversations(identityId);
// Step 2: Get leads due for action today
const leads = await db.leads.findMany({
where: {
identity_id: identityId,
sequence_status: 'IN_SEQUENCE',
next_action_at: { lte: new Date() }
}
});
for (const lead of leads) {
// Step 3: De-dup check (Section 10)
const { canSend, reason } = await canSendOutreach(identityId, lead);
if (!canSend) {
console.log(`Skipping lead ${lead.id}: ${reason}`);
continue;
}
// Step 4: Sync-before-send for race condition prevention (Section 14)
const replyCheck = await checkForRecentReply(identityId, lead);
if (replyCheck.hasReplied) {
await updateLead(lead.id, { sequence_status: 'REPLIED' });
continue;
}
// Step 5: Calculate send time for business hours (Section 11)
const sendTime = getSendTime(lead);
if (sendTime > new Date()) {
await updateLead(lead.id, { next_action_at: sendTime });
continue;
}
// Step 6: Execute the action (Section 9) with error handling (Section 7)
await executeWithErrorHandling(identityId, lead, async () => {
return await executeSequenceStep(identityId, lead);
});
}
}
async function executeSequenceStep(identityId, lead) {
const SEQUENCE_STEPS = [
{ step: 0, action: 'visit', delayDays: 1 },
{ step: 1, action: 'connect', delayDays: 3 },
{ step: 2, action: 'follow_up_connect', delayDays: 4 }, // If not connected after 3 days
{ step: 3, action: 'message', delayDays: 7 }, // After connection accepted
{ step: 4, action: 'follow_up_1', delayDays: 7 },
{ step: 5, action: 'follow_up_2', delayDays: 7 },
{ step: 6, action: 'archive', delayDays: 0 }
];
const currentStep = SEQUENCE_STEPS[lead.current_step];
// Check connection status for message steps
if (['message', 'follow_up_1', 'follow_up_2'].includes(currentStep.action)) {
if (lead.connection_status !== 'CONNECTED') {
// Not connected yet - check if connection was accepted
// Note: Helper functions wrap API calls (see Section 9)
const connections = await extractConnections(identityId, { max_results: 100 });
const isNowConnected = connections.some(
c => c.linkedin_profile_id === lead.linkedin_profile_id
);
if (isNowConnected) {
await updateLead(lead.id, { connection_status: 'CONNECTED' });
} else {
// Still not connected - wait or move to follow-up connect
return;
}
}
}
switch (currentStep.action) {
case 'visit':
await visitProfile(identityId, lead);
break;
case 'connect':
await sendConnectionRequest(identityId, lead);
break;
case 'message':
case 'follow_up_1':
case 'follow_up_2':
const template = MESSAGE_TEMPLATES[currentStep.action];
const message = personalizeMessage(template, lead);
await sendMessage(identityId, lead, message);
break;
case 'archive':
await updateLead(lead.id, { sequence_status: 'ARCHIVED' });
return;
}
// Advance to next step
await updateLead(lead.id, {
current_step: lead.current_step + 1,
next_action_at: addDays(new Date(), currentStep.delayDays),
last_contacted_at: new Date()
});
}
// Message templates
const MESSAGE_TEMPLATES = {
message: "Hi {{first_name}}, I noticed you're working at {{company_name}}. I'd love to connect and share some ideas about...",
follow_up_1: "Hi {{first_name}}, just following up on my previous message. Would love to hear your thoughts on...",
follow_up_2: "{{first_name}}, last follow-up from me! If now isn't a good time, no worries. Feel free to reach out whenever..."
};
function personalizeMessage(template, lead) {
return template
.replace('{{first_name}}', lead.full_name?.split(' ')[0] || 'there')
.replace('{{company_name}}', lead.company_name || 'your company');
}
function addDays(date, days) {
const result = new Date(date);
result.setDate(result.getDate() + days);
return result;
}
```
```python Python theme={null}
from datetime import datetime, timedelta
# Main orchestrator - runs daily via cron
async def run_daily_outreach_for_all_identities():
"""Process outreach for all active identities."""
identities = await db.identities.find_many(
status='ACTIVE', linkedin_connected=True
)
for identity in identities:
try:
await run_outreach_for_identity(identity.id)
except Exception as e:
print(f"Failed for identity {identity.id}: {e}")
async def run_outreach_for_identity(identity_id: str):
"""Process all due leads for a single identity."""
# Step 1: Sync conversations to detect replies (Section 8)
await sync_conversations(identity_id)
# Step 2: Get leads due for action today
leads = await db.leads.find_many(
identity_id=identity_id,
sequence_status='IN_SEQUENCE',
next_action_at__lte=datetime.now()
)
for lead in leads:
# Step 3: De-dup check (Section 10)
result = await can_send_outreach(identity_id, lead)
if not result['can_send']:
print(f"Skipping lead {lead.id}: {result['reason']}")
continue
# Step 4: Sync-before-send for race condition prevention (Section 14)
if await check_for_recent_reply(identity_id, lead):
await update_lead(lead.id, sequence_status='REPLIED')
continue
# Step 5: Calculate send time for business hours (Section 11)
send_time = get_send_time(lead)
if send_time > datetime.now():
await update_lead(lead.id, next_action_at=send_time)
continue
# Step 6: Execute with error handling (Section 7)
await execute_with_error_handling(
identity_id, lead,
lambda: execute_sequence_step(identity_id, lead)
)
SEQUENCE_STEPS = [
{'step': 0, 'action': 'visit', 'delay_days': 1},
{'step': 1, 'action': 'connect', 'delay_days': 3},
{'step': 2, 'action': 'follow_up_connect', 'delay_days': 4},
{'step': 3, 'action': 'message', 'delay_days': 7},
{'step': 4, 'action': 'follow_up_1', 'delay_days': 7},
{'step': 5, 'action': 'follow_up_2', 'delay_days': 7},
{'step': 6, 'action': 'archive', 'delay_days': 0}
]
MESSAGE_TEMPLATES = {
'message': "Hi {{first_name}}, I noticed you're working at {{company_name}}. I'd love to connect and share some ideas about...",
'follow_up_1': "Hi {{first_name}}, just following up on my previous message. Would love to hear your thoughts on...",
'follow_up_2': "{{first_name}}, last follow-up from me! If now isn't a good time, no worries. Feel free to reach out whenever..."
}
async def execute_sequence_step(identity_id: str, lead):
"""Execute the current step in the lead's sequence."""
current_step = SEQUENCE_STEPS[lead.current_step]
action = current_step['action']
# Check connection status for message steps
if action in ['message', 'follow_up_1', 'follow_up_2']:
if lead.connection_status != 'CONNECTED':
# Note: Helper functions wrap API calls (see Section 9)
connections = await extract_connections(identity_id, max_results=100)
is_now_connected = any(
c['linkedin_profile_id'] == lead.linkedin_profile_id
for c in connections
)
if is_now_connected:
await update_lead(lead.id, connection_status='CONNECTED')
else:
return # Still not connected - wait
if action == 'visit':
await visit_profile(identity_id, lead)
elif action == 'connect':
await send_connection_request(identity_id, lead)
elif action in ['message', 'follow_up_1', 'follow_up_2']:
template = MESSAGE_TEMPLATES[action]
message = personalize_message(template, lead)
await send_message(identity_id, lead, message)
elif action == 'archive':
await update_lead(lead.id, sequence_status='ARCHIVED')
return
# Advance to next step
await update_lead(lead.id,
current_step=lead.current_step + 1,
next_action_at=datetime.now() + timedelta(days=current_step['delay_days']),
last_contacted_at=datetime.now()
)
def personalize_message(template: str, lead) -> str:
"""Replace placeholders with lead data."""
first_name = lead.full_name.split(' ')[0] if lead.full_name else 'there'
company = lead.company_name or 'your company'
return template \
.replace('{{first_name}}', first_name) \
.replace('{{company_name}}', company)
```
**How to run:**
* Set up a cron job: `0 8 * * * python run_daily_outreach.py`
* Or use a job queue (Bull, Celery, etc.) for better control
***
## Section 19: Testing Strategy
**Challenge:** Can't spam real LinkedIn users during development.
### Approach 1: Test Accounts
* Create 2-3 LinkedIn test accounts (personal accounts you control)
* Use these as "leads" for end-to-end testing
* Verify messages arrive, connections work
### Approach 2: Dry Run Mode
```javascript JavaScript theme={null}
async function executeAction(identityId, lead, dryRun = false) {
if (dryRun) {
// Log what WOULD happen without calling API
console.log(`DRY RUN: Would send ${lead.current_step} to ${lead.full_name}`);
return { status: 'dry_run', action: lead.current_step };
}
// Real execution
return await executeSequenceStep(identityId, lead);
}
```
```python Python theme={null}
async def execute_action(identity_id: str, lead, dry_run: bool = False):
if dry_run:
# Log what WOULD happen without calling API
print(f"DRY RUN: Would send step {lead.current_step} to {lead.full_name}")
return {'status': 'dry_run', 'action': lead.current_step}
# Real execution
return await execute_sequence_step(identity_id, lead)
```
### Approach 3: Staging Environment
* Use a separate Edges workspace for testing
* Connect test identities only
* Isolates production data
### Approach 4: Unit Test Mocks
```javascript JavaScript theme={null}
// Jest example - mock Edges responses
describe('Outreach Sequence', () => {
beforeEach(() => {
jest.spyOn(global, 'fetch').mockImplementation((url) => {
if (url.includes('extract-conversations')) {
return Promise.resolve({
ok: true,
json: () => Promise.resolve({
output: {
results: [{
linkedin_thread_id: '2-test-thread',
last_message: {
linkedin_profile_id: 123456789, // Simulated lead reply
delivered_at: '2024-01-20T10:00:00Z'
},
participants: [{ linkedin_profile_id: 123456789 }]
}]
}
})
});
}
// Add more mock responses...
});
});
test('detects lead reply correctly', async () => {
const result = await syncConversations('identity_123');
// Assert lead was marked as replied
});
});
```
```python Python theme={null}
# pytest example - mock Edges responses
import pytest
from unittest.mock import patch, MagicMock
@pytest.fixture
def mock_edges_client():
with patch('requests.post') as mock_post:
mock_response = MagicMock()
mock_response.ok = True
mock_response.json.return_value = {
'output': {
'results': [{
'linkedin_thread_id': '2-test-thread',
'last_message': {
'linkedin_profile_id': 123456789, # Simulated lead reply
'delivered_at': '2024-01-20T10:00:00Z'
},
'participants': [{'linkedin_profile_id': 123456789}]
}]
}
}
mock_post.return_value = mock_response
yield mock_post
def test_detects_lead_reply(mock_edges_client):
# Your test here
result = sync_conversations('identity_123')
# Assert lead was marked as replied
```
***
## Section 20: API Reference & SDK
### TypeScript SDK
For a cleaner developer experience, use the official TypeScript SDK:
```bash theme={null}
npm install @edgesrun/sdk
```
See [SDK Documentation →](/v1/sdk)
### Key Actions Reference
Don't duplicate schemas - refer to live action docs for full request/response examples:
| Action | Key Fields You Need | Live Docs |
| -------------------------------- | -------------------------------------------------------- | -------------------------------------------------------- |
| `linkedin-me` | `linkedin_profile_id` (user's own ID) | [View →](/v1/api/actions/linkedin-me) |
| `linkedin-extract-conversations` | `last_message.linkedin_profile_id`, `linkedin_thread_id` | [View →](/v1/api/actions/linkedin-extract-conversations) |
| `linkedin-extract-connections` | `linkedin_profile_id`, `connected_at` | [View →](/v1/api/actions/linkedin-extract-connections) |
| `linkedin-message-profile` | `linkedin_thread_id` (returned on success) | [View →](/v1/api/actions/linkedin-message-profile) |
| `linkedin-connect-profile` | Success/failure status | [View →](/v1/api/actions/linkedin-connect-profile) |
| `linkedin-visit-profile` | Success/failure status | [View →](/v1/api/actions/linkedin-visit-profile) |
| `linkedin-search-people` | `linkedin_profile_url`, `full_name`, `headline` | [View →](/v1/api/actions/linkedin-search-people) |
***
## Summary
You now have everything needed to build a production-ready LinkedIn outreach sequence:
| Feature | Section |
| ----------------------------------- | ------- |
| Identity setup & concepts | 1 |
| Lead sourcing | 2 |
| User profile ID for reply detection | 3 |
| Database schema | 4 |
| State machine for lead lifecycle | 5 |
| Action limits | 6 |
| Error handling per action | 7 |
| Optimized sync with `last_message` | 8 |
| Sending messages & connections | 9 |
| De-duplication logic | 10 |
| Business hours sending | 11 |
| Callback correlation | 12 |
| Rate limits & scaling | 13 |
| Sync-before-send pattern | 14 |
| Identity failure recovery | 15 |
| Edge case handling | 16 |
| Complete orchestrator | 17 |
| Testing strategies | 18 |
| API reference & SDK | 19 |
**Questions?** Check the [FAQ](/v1/faq) or reach out to [support](https://support.edges.run).