Salesforce

API

The Salesforce connector is an API connector for Salesforce CRM. It authenticates via the OAuth username-password flow, queries objects via SOQL, and produces structural claims for Opportunities, Contacts, and Cases. Case descriptions are run through text extraction.

How it works

Authenticates via OAuth 2.0 username-password flow to obtain an access token
Queries Salesforce objects using SOQL with a configurable lookback window
Supports three object types by default: Opportunity, Contact, and Case
Produces structural claims per object (stage, amount, owner, status, priority)
Case descriptions are optionally run through text extraction for richer claims
Pagination handled automatically via nextRecordsUrl

Claim patterns

Opportunities

Subject	Predicate	Object
`(opp name, opportunity)`	`has_stage`	`(stage, stage)`
`(opp name, opportunity)`	`has_amount`	`(amount, currency)`
`(opp name, opportunity)`	`belongs_to`	`(account, account)`
`(opp name, opportunity)`	`owned_by`	`(owner, person)`
`(opp name, opportunity)`	`closes_on`	`(date, date)`

Contacts

Subject	Predicate	Object
`(name, contact)`	`has_role`	`(title, role)`
`(name, contact)`	`works_at`	`(account, company)`

Cases

Subject	Predicate	Object
`(case number, case)`	`has_status`	`(status, status)`
`(case number, case)`	`has_priority`	`(priority, level)`
`(case number, case)`	`submitted_by`	`(contact, person)`
`(case number, case)`	`belongs_to`	`(account, account)`

1 Create a Connected App

In Salesforce Setup, navigate to App Manager and create a new Connected App with OAuth settings enabled. Select the api and refresh_token scopes. Note the Consumer Key (client ID) and Consumer Secret (client secret).

2 Get your security token

If you don't have IP whitelisting, you need a security token. Go to Settings > My Personal Information > Reset My Security Token in your Salesforce account. The token is emailed to you.

3 Install requests

$ pip install requests

4 Connect

conn = db.connect("salesforce",
    username=os.environ["SF_USERNAME"],
    password=os.environ["SF_PASSWORD"],
    security_token=os.environ["SF_SECURITY_TOKEN"],
    client_id=os.environ["SF_CLIENT_ID"],
    client_secret=os.environ["SF_CLIENT_SECRET"],
)
result = conn.run(db)

Parameter	Required	Default	Description
`username`	Yes	—	Salesforce username
`password`	Yes	—	Salesforce password
`security_token`	Yes	—	Salesforce security token (appended to password for auth)
`client_id`	Yes	—	Connected App Consumer Key
`client_secret`	Yes	—	Connected App Consumer Secret
`instance_url`	No	Auto (from auth)	Override Salesforce instance URL
`objects`	No	`["Opportunity", "Contact", "Case"]`	List of Salesforce objects to query
`days`	No	`90`	Lookback window in days
`max_items`	No	`500`	Maximum records per object type
`extraction`	No	`heuristic`	Text extraction mode for case descriptions: `heuristic` or `llm`
`save`	No	`False`	Encrypt and persist credentials

Basic usage

conn = db.connect("salesforce",
    username=os.environ["SF_USERNAME"],
    password=os.environ["SF_PASSWORD"],
    security_token=os.environ["SF_SECURITY_TOKEN"],
    client_id=os.environ["SF_CLIENT_ID"],
    client_secret=os.environ["SF_CLIENT_SECRET"],
)
result = conn.run(db)
print(f"Ingested {result.claims_ingested} claims")

Contacts only

conn = db.connect("salesforce",
    username=os.environ["SF_USERNAME"],
    password=os.environ["SF_PASSWORD"],
    security_token=os.environ["SF_SECURITY_TOKEN"],
    client_id=os.environ["SF_CLIENT_ID"],
    client_secret=os.environ["SF_CLIENT_SECRET"],
    objects=["Contact"],
    max_items=1000,
)
result = conn.run(db)

With custom lookback

conn = db.connect("salesforce",
    username=os.environ["SF_USERNAME"],
    password=os.environ["SF_PASSWORD"],
    security_token=os.environ["SF_SECURITY_TOKEN"],
    client_id=os.environ["SF_CLIENT_ID"],
    client_secret=os.environ["SF_CLIENT_SECRET"],
    days=30,
    extraction="llm",
    save=True,
)
result = conn.run(db)