create_lead
Add a single lead to an email outreach campaign with custom variables and duplicate prevention options.
Instructions
Create a single lead with custom variables.
Use skip_if_in_campaign=true to prevent duplicates (recommended).
Custom variables must match field names defined in the campaign. Example: {"industry": "Technology", "company_size": "50-100"}
For bulk imports (10+ leads), use add_leads_to_campaign_or_list_bulk instead.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
Implementation Reference
- src/instantly_mcp/tools/leads.py:106-159 (handler)The handler function that implements the create_lead tool. It constructs a request body from the input parameters and sends a POST request to the Instantly API's /leads endpoint to create a new lead.async def create_lead(params: CreateLeadInput) -> str: """ Create a single lead with custom variables. Use skip_if_in_campaign=true to prevent duplicates (recommended). Custom variables must match field names defined in the campaign. Example: {"industry": "Technology", "company_size": "50-100"} For bulk imports (10+ leads), use add_leads_to_campaign_or_list_bulk instead. """ client = get_client() body: dict[str, Any] = { "email": params.email, } if params.campaign: body["campaign"] = params.campaign if params.first_name: body["first_name"] = params.first_name if params.last_name: body["last_name"] = params.last_name if params.company_name: body["company_name"] = params.company_name if params.phone: body["phone"] = params.phone if params.website: body["website"] = params.website if params.personalization: body["personalization"] = params.personalization if params.lt_interest_status is not None: body["lt_interest_status"] = params.lt_interest_status if params.pl_value_lead: body["pl_value_lead"] = params.pl_value_lead if params.list_id: body["list_id"] = params.list_id if params.assigned_to: body["assigned_to"] = params.assigned_to if params.skip_if_in_workspace is not None: body["skip_if_in_workspace"] = params.skip_if_in_workspace if params.skip_if_in_campaign is not None: body["skip_if_in_campaign"] = params.skip_if_in_campaign if params.skip_if_in_list is not None: body["skip_if_in_list"] = params.skip_if_in_list if params.blocklist_id: body["blocklist_id"] = params.blocklist_id if params.verify_leads_on_import is not None: body["verify_leads_on_import"] = params.verify_leads_on_import if params.custom_variables: body["custom_variables"] = params.custom_variables result = await client.post("/leads", json=body) return json.dumps(result, indent=2)
- Pydantic model defining the input schema (parameters) for the create_lead tool.class CreateLeadInput(BaseModel): """ Input for creating a lead with custom variables. Use skip_if_in_campaign to prevent duplicates. """ model_config = ConfigDict(str_strip_whitespace=True, extra="ignore") campaign: Optional[str] = Field(default=None, description="Campaign UUID") email: str = Field(..., description="Required - lead email address") first_name: Optional[str] = Field(default=None) last_name: Optional[str] = Field(default=None) company_name: Optional[str] = Field(default=None) phone: Optional[str] = Field(default=None) website: Optional[str] = Field(default=None) personalization: Optional[str] = Field(default=None) lt_interest_status: Optional[int] = Field( default=None, ge=-3, le=4, description="Interest status (-3 to 4)" ) pl_value_lead: Optional[str] = Field(default=None, description="Pipeline value") list_id: Optional[str] = Field(default=None) assigned_to: Optional[str] = Field(default=None, description="User UUID") skip_if_in_workspace: Optional[bool] = Field(default=None) skip_if_in_campaign: Optional[bool] = Field(default=None, description="Recommended") skip_if_in_list: Optional[bool] = Field(default=None) blocklist_id: Optional[str] = Field(default=None) verify_leads_on_import: Optional[bool] = Field(default=None) custom_variables: Optional[dict[str, Any]] = Field( default=None, description="Match campaign field names" )
- src/instantly_mcp/server.py:118-126 (registration)The dynamic registration loop in register_tools() that registers the create_lead function (imported from tools/leads.py) as an MCP tool with its annotations.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)
- src/instantly_mcp/server.py:89-89 (registration)TOOL_ANNOTATIONS dictionary entry providing metadata (destructiveHint: False) for the create_lead tool used during registration."create_lead": {"destructiveHint": False},