Salesforce

This connector is optimized for AI agents. For the data replication connector, see Salesforce.

The Salesforce agent connector is a Python package that equips AI agents to interact with Salesforce through strongly typed, well-documented tools. It's ready to use directly in your Python app, in an agent framework, or exposed through an MCP.

Salesforce is a cloud-based CRM platform that helps businesses manage customer relationships, sales pipelines, and business operations. This connector provides access to accounts, contacts, leads, opportunities, tasks, events, campaigns, cases, notes, and attachments for sales analytics and customer relationship management.

Example prompts

The Salesforce connector is optimized to handle prompts like these.

List recent contacts in my Salesforce account
List open cases in my Salesforce account
Show me the notes and attachments for a recent account
List all available reports in Salesforce
Run my quarterly revenue report and show the results
Show me my top 5 opportunities this month
List all contacts from {company} in the last quarter
Search for leads in the technology sector with revenue over $10M
What trends can you identify in my recent sales pipeline?
Summarize the open cases for my key accounts
Find upcoming events related to my most important opportunities
Analyze the performance of my recent marketing campaigns
Identify the highest value opportunities I'm currently tracking

Unsupported prompts

The Salesforce connector isn't currently able to handle prompts like these.

Create a new lead for {person}
Update the status of my sales opportunity
Schedule a follow-up meeting with {customer}
Delete this old contact record
Send an email to all contacts in this campaign

Entities and actions

This connector supports the following entities and actions. For more details, see this connector's full reference documentation.

Entity	Actions
Sobjects	List
Accounts	List, Get, API Search, Context Store Search
Contacts	List, Get, API Search, Context Store Search
Leads	List, Get, API Search, Context Store Search
Opportunities	List, Get, API Search, Context Store Search
Tasks	List, Get, API Search, Context Store Search
Events	List, Get, API Search
Campaigns	List, Get, API Search
Cases	List, Get, API Search
Notes	List, Get, API Search
Content Versions	List, Get, Download
Attachments	List, Get, Download
Reports	List, Get
Users	List, Get, Context Store Search
Opportunity Stages	List, Get, Context Store Search
Query	List

Salesforce API docs

See the official Salesforce API reference.

SDK installation

uv pip install airbyte-agent-sdk

SDK usage

Connectors can run in hosted or open source mode.

Hosted

In hosted mode, API credentials are stored securely in Airbyte Agents. You provide your Airbyte credentials instead. If your Airbyte client can access multiple organizations, also set organization_id.

This example assumes you've already authenticated your connector with Airbyte. See Authentication to learn more about authenticating. If you need a step-by-step guide, see the hosted execution tutorial.

The connect() factory returns a fully typed SalesforceConnector and reads AIRBYTE_CLIENT_ID / AIRBYTE_CLIENT_SECRET from the environment:

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector

connector = connect("salesforce", workspace_name="<your_workspace_name>")

agent = Agent("openai:gpt-4o")

@agent.tool_plain
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector

connector = connect("salesforce", workspace_name="<your_workspace_name>")

@tool
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector

connector = connect("salesforce", workspace_name="<your_workspace_name>")

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@SalesforceConnector.tool_utils(framework="openai_agents")
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Salesforce Assistant", tools=[salesforce_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector

connector = connect("salesforce", workspace_name="<your_workspace_name>")

mcp = FastMCP("Salesforce Agent")

@mcp.tool
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Or pass credentials explicitly (equivalent, useful when you're not loading them from the environment):

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = SalesforceConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

agent = Agent("openai:gpt-4o")

@agent.tool_plain
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = SalesforceConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

@tool
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = SalesforceConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@SalesforceConnector.tool_utils(framework="openai_agents")
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Salesforce Assistant", tools=[salesforce_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = SalesforceConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

mcp = FastMCP("Salesforce Agent")

@mcp.tool
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Open source

In open source mode, you provide API credentials directly to the connector.

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.connectors.salesforce.models import SalesforceAuthConfig

connector = SalesforceConnector(
    auth_config=SalesforceAuthConfig(
        refresh_token="<OAuth refresh token for automatic token renewal>",
        client_id="<Connected App Consumer Key>",
        client_secret="<Connected App Consumer Secret>"
    )
)

agent = Agent("openai:gpt-4o")

@agent.tool_plain
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.connectors.salesforce.models import SalesforceAuthConfig

connector = SalesforceConnector(
    auth_config=SalesforceAuthConfig(
        refresh_token="<OAuth refresh token for automatic token renewal>",
        client_id="<Connected App Consumer Key>",
        client_secret="<Connected App Consumer Secret>"
    )
)

@tool
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.connectors.salesforce.models import SalesforceAuthConfig

connector = SalesforceConnector(
    auth_config=SalesforceAuthConfig(
        refresh_token="<OAuth refresh token for automatic token renewal>",
        client_id="<Connected App Consumer Key>",
        client_secret="<Connected App Consumer Secret>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@SalesforceConnector.tool_utils(framework="openai_agents")
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Salesforce Assistant", tools=[salesforce_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.salesforce import SalesforceConnector
from airbyte_agent_sdk.connectors.salesforce.models import SalesforceAuthConfig

connector = SalesforceConnector(
    auth_config=SalesforceAuthConfig(
        refresh_token="<OAuth refresh token for automatic token renewal>",
        client_id="<Connected App Consumer Key>",
        client_secret="<Connected App Consumer Secret>"
    )
)

mcp = FastMCP("Salesforce Agent")

@mcp.tool
@SalesforceConnector.tool_utils
async def salesforce_execute(entity: str, action: str, params: dict | None = None):
    """Execute Salesforce connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Authentication

For all authentication options, see the connector's authentication documentation.

Version information

Connector version: 1.0.18

Example prompts​

Unsupported prompts​

Entities and actions​

Salesforce API docs​

SDK installation​

SDK usage​

Hosted​

Open source​

Authentication​

Version information​