move_leads_to_campaign_or_list
Transfer or duplicate leads between email campaigns and contact lists using specific IDs, search filters, or status criteria to organize outreach data.
Instructions
Move or copy leads between campaigns/lists.
Runs as background job for large operations.
Source selection (use one):
ids: Specific lead IDs to move
search + campaign/list_id: Filter leads by search term
filter + campaign/list_id: Filter by status
Destination (use one):
to_campaign_id: Target campaign
to_list_id: Target list
Set copy_leads=true to copy instead of move.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
Implementation Reference
- src/instantly_mcp/tools/leads.py:357-421 (handler)Main handler function that implements the tool logic by constructing a request body from MoveLeadsInput parameters and posting to the /leads/move API endpoint.
async def move_leads_to_campaign_or_list(params: MoveLeadsInput) -> str: """ Move or copy leads between campaigns/lists. Runs as background job for large operations. Source selection (use one): - ids: Specific lead IDs to move - search + campaign/list_id: Filter leads by search term - filter + campaign/list_id: Filter by status Destination (use one): - to_campaign_id: Target campaign - to_list_id: Target list Set copy_leads=true to copy instead of move. """ client = get_client() body: dict[str, Any] = {} if params.to_campaign_id: body["to_campaign_id"] = params.to_campaign_id if params.to_list_id: body["to_list_id"] = params.to_list_id if params.ids: body["ids"] = params.ids if params.search: body["search"] = params.search if params.filter: body["filter"] = params.filter if params.campaign: body["campaign"] = params.campaign if params.list_id: body["list_id"] = params.list_id if params.in_campaign is not None: body["in_campaign"] = params.in_campaign if params.in_list is not None: body["in_list"] = params.in_list if params.queries: body["queries"] = params.queries if params.excluded_ids: body["excluded_ids"] = params.excluded_ids if params.contacts: body["contacts"] = params.contacts if params.check_duplicates_in_campaigns is not None: body["check_duplicates_in_campaigns"] = params.check_duplicates_in_campaigns if params.skip_leads_in_verification is not None: body["skip_leads_in_verification"] = params.skip_leads_in_verification if params.limit is not None: body["limit"] = params.limit if params.assigned_to: body["assigned_to"] = params.assigned_to if params.esp_code is not None: body["esp_code"] = params.esp_code if params.esg_code is not None: body["esg_code"] = params.esg_code if params.copy_leads is not None: body["copy_leads"] = params.copy_leads if params.check_duplicates is not None: body["check_duplicates"] = params.check_duplicates result = await client.post("/leads/move", json=body) return json.dumps(result, indent=2) - Pydantic schema defining the input parameters for the move_leads_to_campaign_or_list tool, including source selection, destination, and various options.
class MoveLeadsInput(BaseModel): """ Input for moving or copying leads between campaigns/lists. Runs as background job for large operations. """ model_config = ConfigDict(str_strip_whitespace=True, extra="ignore") to_campaign_id: Optional[str] = Field( default=None, description="Destination (OR to_list_id)" ) to_list_id: Optional[str] = Field( default=None, description="Destination (OR to_campaign_id)" ) ids: Optional[list[str]] = Field(default=None, description="Lead IDs") search: Optional[str] = Field(default=None) filter: Optional[str] = Field(default=None, description="Contact status filter") campaign: Optional[str] = Field(default=None, description="Source campaign") list_id: Optional[str] = Field(default=None, description="Source list") in_campaign: Optional[bool] = Field(default=None) in_list: Optional[bool] = Field(default=None) queries: Optional[list[dict[str, Any]]] = Field(default=None) excluded_ids: Optional[list[str]] = Field(default=None) contacts: Optional[list[str]] = Field(default=None) check_duplicates_in_campaigns: Optional[bool] = Field(default=None) skip_leads_in_verification: Optional[bool] = Field(default=None) limit: Optional[int] = Field(default=None) assigned_to: Optional[str] = Field(default=None) esp_code: Optional[int] = Field( default=None, description="0=Queue, 1=Google, 2=MS, etc." ) esg_code: Optional[int] = Field( default=None, description="0=Queue, 1=Barracuda, etc." ) copy_leads: Optional[bool] = Field(default=None, description="Copy instead of move") check_duplicates: Optional[bool] = Field(default=None) - src/instantly_mcp/server.py:67-128 (registration)TOOL_ANNOTATIONS dictionary entry for 'move_leads_to_campaign_or_list' with destructiveHint: False, and the registration loop that applies annotations to all tools including this one.
TOOL_ANNOTATIONS = { # Account tools "list_accounts": {"readOnlyHint": True}, "get_account": {"readOnlyHint": True}, "create_account": {"destructiveHint": False}, "update_account": {"destructiveHint": False}, "manage_account_state": {"destructiveHint": False}, "delete_account": {"destructiveHint": True, "confirmationRequiredHint": True}, # Campaign tools "create_campaign": {"destructiveHint": False}, "list_campaigns": {"readOnlyHint": True}, "get_campaign": {"readOnlyHint": True}, "update_campaign": {"destructiveHint": False}, "activate_campaign": {"destructiveHint": False}, "pause_campaign": {"destructiveHint": False}, "delete_campaign": {"destructiveHint": True, "confirmationRequiredHint": True}, "search_campaigns_by_contact": {"readOnlyHint": True}, # Lead tools "list_leads": {"readOnlyHint": True}, "get_lead": {"readOnlyHint": True}, "create_lead": {"destructiveHint": False}, "update_lead": {"destructiveHint": False}, "list_lead_lists": {"readOnlyHint": True}, "create_lead_list": {"destructiveHint": False}, "update_lead_list": {"destructiveHint": False}, "get_verification_stats_for_lead_list": {"readOnlyHint": True}, "add_leads_to_campaign_or_list_bulk": {"destructiveHint": False}, "delete_lead": {"destructiveHint": True, "confirmationRequiredHint": True}, "delete_lead_list": {"destructiveHint": True, "confirmationRequiredHint": True}, "move_leads_to_campaign_or_list": {"destructiveHint": False}, # Email tools "list_emails": {"readOnlyHint": True}, "get_email": {"readOnlyHint": True}, "reply_to_email": {"destructiveHint": True, "confirmationRequiredHint": True}, "count_unread_emails": {"readOnlyHint": True}, "verify_email": {"readOnlyHint": True}, "mark_thread_as_read": {"destructiveHint": False}, # Analytics tools "get_campaign_analytics": {"readOnlyHint": True}, "get_daily_campaign_analytics": {"readOnlyHint": True}, "get_warmup_analytics": {"readOnlyHint": True}, # Background job tools "list_background_jobs": {"readOnlyHint": True}, "get_background_job": {"readOnlyHint": True}, } 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 exports the move_leads_to_campaign_or_list function for inclusion in get_all_tools() used by 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, ]