This Python script is a Tool in Relevance AI and is designed for use by the FA | Attio Agent. The script integrates with the Attio API to manage and update lead records associated with people in a database. It performs the following tasks:
Search for a Person by Email or Name:
The script first tries to find an existing person in Attio by querying their email address.
If no match is found by email, it falls back to searching by the person's full name.
Check for Existing Leads:
Once the person is identified, the script checks whether there is an existing lead for the person in a specific pipeline (e.g., "Recruiter" pipeline).
Create or Update Leads:
If a lead exists, it updates the lead using a PATCH request, modifying fields such as the lead's name, status, and pipeline.
If no existing lead is found, the script creates a new lead with the provided information using a POST request.
Data Preparation:
The script builds a payload for the lead that includes key attributes such as the person's name, pipeline, and status.
It ensures that no empty fields are included in the payload.
Response Handling:
After the lead is created or updated, the script prints the result and provides a URL to view the lead in Attio.
In case of an error, the script captures and prints an error message.
This code automates the process of checking, creating, and updating leads in Attio, simplifying lead management and ensuring that existing records are updated efficiently.
Key Components:
Attio API Token: Used to authenticate requests.
Search Functions: Look up people by email or name.
Lead Management: Checks for existing leads and either updates or creates new ones.
PATCH/POST Requests: Sends updates (PATCH) or creates new records (POST) via Attio's API.
This script is useful for managing leads in systems where leads need to be continuously tracked and updated based on new data.
import requests
import json
from datetime import datetime
# Attio API access token
access_token = "2adaba20c6efafdf8e15a514ff47435eccd31c78a0e5966a6924918363d82818"
headers = {
"Authorization": f"Bearer {access_token}",
"accept": "application/json",
"content-type": "application/json"
}
# Object IDs
leads_object_id = "1fb52cd1-5ac8-40e6-808a-74fd94937319" # Object ID for 'Leads'
people_object_id = "b9d9b6c4-fe72-4374-bff6-fcc4a24c7e43" # Object ID for 'People'
# Attribute IDs for lead creation
name_attribute_id = "0f159b2d-972d-43fc-8d50-5ee4217f9d85"
status_attribute_id = "132bf713-2c0d-4576-a144-17f2a4c3cbaf"
pipeline_attribute_id = "d893d182-0fa3-403a-9428-37093ff59e13"
type_attribute_id = "4e6c3511-634d-420d-996e-22d17defc527"
people_attribute_id = "24629fb6-ac59-4a84-9147-4c81a0175eeb" # Correct attribute ID for People
# Function to search for a Person record based on email
def search_person_by_email(email):
print(f"Searching for person by email: {email}")
query_payload = {
"filter": {
"email_addresses": {
"email_address": {
"$eq": email
}
}
},
"limit": 1
}
print(f"Query Payload: {json.dumps(query_payload, indent=2)}")
response = requests.post(f"https://api.attio.com/v2/objects/{people_object_id}/records/query", headers=headers, json=query_payload)
print(f"Response Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
if response.status_code == 200:
person_data = response.json()
if person_data.get("data", []):
person_id = person_data["data"][0]["id"]["record_id"]
print(f"Person found with ID: {person_id}")
return person_id # Return the record ID
else:
print(f"Person not found by email: {email}")
return None
else:
print(f"Error searching person by email: {response.status_code}, {response.text}")
return None
# Function to search for a Person record based on name
def search_person_by_name(first_name, last_name):
print(f"Searching for person by name: {first_name} {last_name}")
query_payload = {
"filter": {
"name": {
"full_name": {
"$eq": f"{first_name} {last_name}"
}
}
},
"limit": 1
}
print(f"Query Payload: {json.dumps(query_payload, indent=2)}")
response = requests.post(f"https://api.attio.com/v2/objects/{people_object_id}/records/query", headers=headers, json=query_payload)
print(f"Response Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
if response.status_code == 200:
person_data = response.json()
if person_data.get("data", []):
person_id = person_data["data"][0]["id"]["record_id"]
print(f"Person found with ID: {person_id}")
return person_id # Return the record ID
else:
print(f"Person not found by name: {first_name} {last_name}")
return None
else:
print(f"Error searching person by name: {response.status_code}, {response.text}")
return None
# Function to check if a Lead exists for a person in a given pipeline
def query_existing_lead(person_id, pipeline):
print(f"Querying existing lead for person ID {person_id} in pipeline {pipeline}")
query_url = f"https://api.attio.com/v2/objects/{leads_object_id}/records/query"
query_payload = {
"filter": {
"$and": [
{
"people": {
"target_record_id": person_id
}
},
{
"pipeline": pipeline
}
]
}
}
print(f"Query Payload: {json.dumps(query_payload, indent=2)}")
response = requests.post(query_url, headers=headers, json=query_payload)
print(f"Response Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
if response.status_code == 200:
data = response.json()
if data.get("data", []):
lead_id = data["data"][0]["id"]["record_id"]
print(f"Existing lead found with ID: {lead_id}")
return lead_id # Lead exists, return the record ID
else:
print("No existing lead found.")
return None # No lead found
else:
print(f"Error querying lead: {response.status_code}, {response.text}")
return None
# Function to assert (create or update) a Lead associated with a Person
def assert_lead(params):
print(f"Asserting lead with params: {params}")
status = params.get('status')
pipeline = params.get('pipeline')
first_name = params.get('first_name')
last_name = params.get('last_name')
email = params.get('email')
# Check if required parameters are provided
if not all([pipeline, first_name, last_name, email]):
error_msg = "Missing required parameters: pipeline, first_name, last_name, or email"
print(error_msg)
return {"error": error_msg}
# Step 0: Generate the lead name based on the naming convention
current_date = datetime.now()
lead_name = f"{first_name[0]}{last_name[0]} - {pipeline} - {current_date.strftime('%b %y')}"
# Step 1: Try to find the Person by email
person_id = search_person_by_email(email)
# Step 2: If no person found by email, fallback to searching by full name
if not person_id:
person_id = search_person_by_name(first_name, last_name)
if not person_id:
error_msg = "Person not found. Make sure the person is in the system before creating a lead."
print(error_msg)
return {"error": error_msg}
# Step 3: Query if an existing lead exists for this person in the given pipeline
existing_lead_id = query_existing_lead(person_id, pipeline)
# Step 4: Prepare the payload
payload = {
"data": {
"values": {
name_attribute_id: lead_name,
pipeline_attribute_id: pipeline,
people_attribute_id: person_id, # This links the person to the lead
status_attribute_id: status if status else None,
}
}
}
# Remove keys with None values (i.e., if status is not provided)
payload["data"]["values"] = {k: v for k, v in payload["data"]["values"].items() if v is not None}
print(f"Payload for lead assertion: {json.dumps(payload, indent=2)}")
# Step 5: If a lead exists, update it using PATCH, otherwise create a new one
if existing_lead_id:
# Update the existing lead (PATCH)
print(f"Updating existing lead with ID: {existing_lead_id}")
patch_lead_url = f"https://api.attio.com/v2/objects/{leads_object_id}/records/{existing_lead_id}"
response = requests.patch(patch_lead_url, headers=headers, json=payload) # PATCH request is sent here
else:
# Create a new lead (POST)
print("Creating a new lead.")
create_lead_url = f"https://api.attio.com/v2/objects/{leads_object_id}/records"
response = requests.post(create_lead_url, headers=headers, json=payload) # POST request for creating new lead
print(f"Response Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
if response.status_code in [200, 201]:
lead_data = response.json()
lead_id = lead_data["data"]["id"]["record_id"]
print(f"Lead created/updated successfully with ID: {lead_id}")
return {"message": "Lead created/updated successfully", "lead_url": f"https://app.attio.com/sidebar/lead/{lead_id}/"}
else:
error_msg = f"Failed to assert lead: {response.status_code}, {response.text}"
print(error_msg)
return {"error": error_msg}
# Main function
def main(params):
try:
print("Starting main function.")
result = assert_lead(params)
print(f"Result: {result}")
return result
except Exception as e:
error_msg = f"An error occurred: {str(e)}"
print(error_msg)
return {"error": error_msg}
if __name__ == "__main__":
result = main(params)
print(result)