Instantly MCP Server

Instructions

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)

• src/instantly_mcp/models/leads.py:42-75 (schema)

  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},