list_leads
Retrieve and manage leads from Instantly.ai campaigns with pagination, filtering by status, and deduplication by email.
Instructions
List leads with cursor-based pagination (100 per page).
PAGINATION: If response contains pagination.next_starting_after, there are MORE results. Call again with starting_after= to get next page. Continue until pagination.next_starting_after is null.
Filter values:
FILTER_VAL_CONTACTED: Leads that have been contacted
FILTER_VAL_NOT_CONTACTED: Leads not yet contacted
FILTER_VAL_COMPLETED: Leads that completed sequence
FILTER_VAL_ACTIVE: Currently active leads
Use distinct_contacts=true to deduplicate by email.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | No |
Implementation Reference
- src/instantly_mcp/tools/leads.py:28-87 (handler)The primary handler function implementing the list_leads tool logic. It constructs the API request from input parameters, calls the /leads/list endpoint, enhances the response with pagination hints, and serializes to JSON.
async def list_leads(params: Optional[ListLeadsInput] = None) -> str: """ List leads with cursor-based pagination (100 per page). PAGINATION: If response contains pagination.next_starting_after, there are MORE results. Call again with starting_after=<that value> to get next page. Continue until pagination.next_starting_after is null. Filter values: - FILTER_VAL_CONTACTED: Leads that have been contacted - FILTER_VAL_NOT_CONTACTED: Leads not yet contacted - FILTER_VAL_COMPLETED: Leads that completed sequence - FILTER_VAL_ACTIVE: Currently active leads Use distinct_contacts=true to deduplicate by email. """ client = get_client() # Handle case where params is None (for OpenAI/non-Claude clients) # Set default limit=100 to return more results by default if params is None: params = ListLeadsInput(limit=100) # Build request body for POST /leads/list body: dict[str, Any] = {} # Default to 100 results if no limit specified body["limit"] = params.limit or 100 if params.campaign: body["campaign"] = params.campaign if params.list_id: body["list_id"] = params.list_id if params.list_ids: body["list_ids"] = params.list_ids if params.status: body["status"] = params.status if params.created_after: body["created_after"] = params.created_after if params.created_before: body["created_before"] = params.created_before if params.search: body["search"] = params.search if params.filter: body["filter"] = params.filter if params.distinct_contacts is not None: body["distinct_contacts"] = params.distinct_contacts if params.starting_after: body["starting_after"] = params.starting_after result = await client.post("/leads/list", json=body) # Add pagination guidance for LLMs if isinstance(result, dict): pagination = result.get("pagination", {}) next_cursor = pagination.get("next_starting_after") if next_cursor: result["_pagination_hint"] = f"MORE RESULTS AVAILABLE. Call list_leads with starting_after='{next_cursor}' to get next page." return json.dumps(result, indent=2) - Pydantic BaseModel defining the input schema (ListLeadsInput) for the list_leads tool, including fields for filtering, pagination, and search.
class ListLeadsInput(BaseModel): """Input for listing leads with pagination and filtering.""" # Use extra="ignore" to be tolerant of unexpected fields from LLMs model_config = ConfigDict(str_strip_whitespace=True, extra="ignore") campaign: Optional[str] = Field(default=None, description="Campaign UUID") list_id: Optional[str] = Field(default=None, description="List UUID") list_ids: Optional[list[str]] = Field(default=None, description="Multiple list UUIDs") status: Optional[str] = Field(default=None) created_after: Optional[str] = Field(default=None, description="YYYY-MM-DD") created_before: Optional[str] = Field(default=None, description="YYYY-MM-DD") search: Optional[str] = Field(default=None, description="Name or email") filter: Optional[str] = Field( default=None, description="FILTER_VAL_CONTACTED, FILTER_VAL_NOT_CONTACTED, FILTER_VAL_COMPLETED, FILTER_VAL_ACTIVE, etc." ) distinct_contacts: Optional[bool] = Field(default=None, description="Dedupe by email") limit: Optional[int] = Field(default=100, ge=1, le=100, description="Results per page (1-100, default: 100)") starting_after: Optional[str] = Field( default=None, description="Pagination cursor - use value from pagination.next_starting_after to get next page" ) - src/instantly_mcp/server.py:118-128 (registration)Dynamic registration loop in register_tools() where list_leads is registered with FastMCP using its name and readOnlyHint annotation.
for tool_func in tools: tool_name = tool_func.__name__ annotations = TOOL_ANNOTATIONS.get(tool_name, {}) # Register tool with FastMCP mcp.tool( name=tool_name, annotations=annotations, )(tool_func) print(f"[Instantly MCP] ✅ Registered {len(tools)} tools", file=sys.stderr) - src/instantly_mcp/tools/leads.py:450-463 (registration)LEAD_TOOLS list that includes the list_leads function, used by get_all_tools() for server registration.
LEAD_TOOLS = [ list_leads, get_lead, create_lead, update_lead, list_lead_lists, create_lead_list, update_lead_list, get_verification_stats_for_lead_list, add_leads_to_campaign_or_list_bulk, delete_lead, delete_lead_list, move_leads_to_campaign_or_list, ] - Re-export of ListLeadsInput schema in models package __init__.
from .leads import ( ListLeadsInput,