feat(ingest): add Salesforce ingestor

docs/docs.json

@@ -106,7 +106,8 @@
               "integrations/langchain",
               "integrations/docling",
               "integrations/snowflake",
-              "integrations/databricks"
+              "integrations/databricks",
+              "integrations/salesforce"
             ]
           },
           {

docs/integrations/salesforce.md

@@ -0,0 +1,350 @@
+---
+title: "Salesforce Integration"
+description: "Ingest CRM records from Salesforce sObjects and SOQL queries into Semantica's KG pipeline."
+icon: "cloud"
+---
+
+> Extract Accounts, Contacts, Opportunities, and custom objects from Salesforce into Semantica with username/password/security-token or session-based authentication.
+
+<Note>
+**JWT Bearer** and **Bulk API 2.0** authentication are not yet implemented. Use username/password/security-token for server-side integrations, or pass a pre-existing `session_id` + `instance_url` if your environment already manages OAuth tokens.
+</Note>
+
+
+## Installation
+
+```bash
+# Install with Salesforce support
+pip install "semantica[db-salesforce]"
+
+# Or install the connector separately
+pip install simple-salesforce>=1.12.0
+```
+
+
+## Basic Usage
+
+```python
+from semantica.ingest import SalesforceIngestor
+import os
+
+ingestor = SalesforceIngestor(
+    username=os.getenv("SALESFORCE_USERNAME"),
+    password=os.getenv("SALESFORCE_PASSWORD"),
+    security_token=os.getenv("SALESFORCE_SECURITY_TOKEN"),
+    domain=os.getenv("SALESFORCE_DOMAIN", "login"),   # "test" for sandbox
+)
+
+data = ingestor.ingest_sobject("Account", fields=["Id", "Name", "Industry"], limit=1000)
+print(f"Retrieved {data.row_count} of {data.total_size} matching records")
+print(f"Columns: {data.columns}")
+```
+
+<Tip>
+Use environment variables (or a `.env` file with `python-dotenv`) to keep credentials out of source code. `SalesforceIngestor()` with no arguments reads from `SALESFORCE_*` environment variables automatically.
+</Tip>
+
+
+## Authentication Methods
+
+<Tabs>
+  <Tab title="Username / Password / Security Token">
+    ```python
+    import os
+    from semantica.ingest import SalesforceIngestor
+
+    ingestor = SalesforceIngestor(
+        username=os.getenv("SALESFORCE_USERNAME"),
+        password=os.getenv("SALESFORCE_PASSWORD"),
+        security_token=os.getenv("SALESFORCE_SECURITY_TOKEN"),
+        domain="login",   # production; use "test" for sandbox
+    )
+    ```
+    Set the required environment variables before running:
+    ```bash
+    export SALESFORCE_USERNAME="your-username@example.com"
+    export SALESFORCE_PASSWORD="your-password"
+    export SALESFORCE_SECURITY_TOKEN="your-security-token"
+    ```
+    The standard server-side flow. The security token is appended to the
+    password during Salesforce SOAP login. Generate or reset it under
+    **Settings → My Personal Information → Reset My Security Token**.
+  </Tab>
+  <Tab title="Session ID + Instance URL">
+    ```python
+    ingestor = SalesforceIngestor(
+        session_id=os.getenv("SALESFORCE_SESSION_ID"),
+        instance_url=os.getenv("SALESFORCE_INSTANCE_URL"),
+    )
+    ```
+    Use this when your environment already manages the OAuth token
+    lifecycle (e.g. a connected app obtaining tokens via the web-server
+    or device flow). Pass the access token as `session_id` and the full
+    instance URL (e.g. `https://myorg.my.salesforce.com`) as
+    `instance_url`.
+  </Tab>
+  <Tab title="Sandbox">
+    ```python
+    import os
+    from semantica.ingest import SalesforceIngestor
+
+    ingestor = SalesforceIngestor(
+        username=os.getenv("SALESFORCE_USERNAME"),
+        password=os.getenv("SALESFORCE_PASSWORD"),
+        security_token=os.getenv("SALESFORCE_SECURITY_TOKEN"),
+        domain="test",   # routes to test.salesforce.com
+    )
+    ```
+    ```bash
+    export SALESFORCE_USERNAME="your-sandbox-username@example.com.sandbox"
+    export SALESFORCE_PASSWORD="your-password"
+    export SALESFORCE_SECURITY_TOKEN="your-security-token"
+    export SALESFORCE_DOMAIN="test"
+    ```
+    Replace `domain="login"` with `domain="test"` (or set
+    `SALESFORCE_DOMAIN=test` in your environment) to connect to a
+    developer or full sandbox.
+  </Tab>
+</Tabs>
+
+### Environment variables
+
+All constructor parameters have environment-variable fallbacks:
+
+| Variable | Parameter | Default |
+|---|---|---|
+| `SALESFORCE_USERNAME` | `username` | — |
+| `SALESFORCE_PASSWORD` | `password` | — |
+| `SALESFORCE_SECURITY_TOKEN` | `security_token` | — |
+| `SALESFORCE_DOMAIN` | `domain` | `"login"` |
+| `SALESFORCE_INSTANCE_URL` | `instance_url` | — |
+| `SALESFORCE_SESSION_ID` | `session_id` | — |
+| `SALESFORCE_API_VERSION` | `api_version` | library default (`59.0`) |
+
+
+## Object Ingestion
+
+### Ingest a standard object
+
+```python
+data = ingestor.ingest_sobject(
+    "Account",
+    fields=["Id", "Name", "Industry", "AnnualRevenue", "BillingCity"],
+    where="Type = 'Customer' AND AnnualRevenue > 1000000",
+    order_by="Name ASC",
+    limit=5000,
+)
+print(f"Retrieved {data.row_count} of {data.total_size} matching records")
+```
+
+<Note>
+`data.row_count` is the number of records in `data.data` (i.e. what was actually returned after any `limit`). `data.total_size` is Salesforce's `totalSize` — the number of records matching the query *before* the limit. Compare them to know whether you got all results.
+</Note>
+
+### Ingest a custom object
+
+Custom objects end with `__c` in their API name:
+
+```python
+data = ingestor.ingest_sobject(
+    "My_Custom_Object__c",
+    fields=["Id", "Name", "Custom_Field__c"],
+)
+```
+
+Relationship traversal fields (`Owner.Name`) are also supported:
+
+```python
+data = ingestor.ingest_sobject(
+    "Contact",
+    fields=["Id", "Name", "Email", "Account.Name", "Owner.Name"],
+    limit=10000,
+)
+```
+
+### Let Semantica choose the fields
+
+When `fields` is omitted, all selectable fields are fetched via `describe()`
+(one extra API call). Compound address and geolocation fields (`type=address`,
+`type=location`) are automatically excluded — select their components
+(`BillingStreet`, `BillingCity`, `Location__Latitude__s`, …) individually if
+you need them.
+
+```python
+data = ingestor.ingest_sobject("Opportunity")
+```
+
+
+## Raw SOQL Ingestion
+
+Pass any valid SOQL query verbatim — pagination is handled automatically:
+
+```python
+data = ingestor.ingest_query("""
+    SELECT Id, Name, StageName, Amount, CloseDate,
+           Account.Name, Owner.Name
+    FROM Opportunity
+    WHERE IsClosed = false
+    ORDER BY CloseDate ASC
+""")
+print(f"Open opportunities: {data.row_count}")
+```
+
+The query is passed to the Salesforce REST API unchanged. The caller is
+responsible for SOQL correctness and safety.
+
+<Warning>
+`ingest_query` does not validate or sanitise the SOQL string. Use
+`ingest_sobject` (which validates sObject names, field names, and WHERE/ORDER
+BY fragments) when building queries from application-controlled inputs.
+</Warning>
+
+
+## Document Export
+
+Convert ingested records to the Semantica document format for use with
+`GraphBuilder`:
+
+```python
+documents = ingestor.export_as_documents(
+    data,
+    id_field="Id",                            # default; Salesforce 18-char record Id
+    text_fields=["Name", "Description"],      # omit to join all string fields
+)
+
+print(f"Created {len(documents)} documents")
+# Each document:
+# {
+#   "id": "001xx000003GYk2AAG",
+#   "text": "Acme Corp Enterprise software company",
+#   "metadata": {
+#     "source": "salesforce",
+#     "sobject": "Account",
+#     "instance_url": "https://myorg.my.salesforce.com",
+#     "row_data": { ... full cleaned record ... }
+#   }
+# }
+```
+
+Feed the documents directly into `GraphBuilder`:
+
+```python
+from semantica.kg import GraphBuilder
+
+builder = GraphBuilder()
+kg = builder.build(documents)
+```
+
+
+## Object and Schema Discovery
+
+```python
+# List all accessible sObjects
+sobject_names = ingestor.list_sobjects()
+print(sobject_names[:10])   # ["Account", "Case", "Contact", ...]
+
+# Inspect fields for a specific sObject
+schema = ingestor.get_sobject_schema("Account")
+for field in schema["fields"]:
+    print(f"{field['name']}: {field['type']} (nillable={field['nillable']})")
+```
+
+
+## Context Manager
+
+Prefer the context manager for long-running jobs — it opens one connection on
+entry and closes it on exit, so every ingestion call inside the `with` block
+reuses the same authenticated session:
+
+```python
+with SalesforceIngestor(
+    username=os.getenv("SALESFORCE_USERNAME"),
+    password=os.getenv("SALESFORCE_PASSWORD"),
+    security_token=os.getenv("SALESFORCE_SECURITY_TOKEN"),
+) as sf:
+    accounts = sf.ingest_sobject("Account", limit=10000)
+    contacts = sf.ingest_sobject("Contact", limit=10000)
+    sobjects = sf.list_sobjects()
+```
+
+
+## Convenience Function
+
+Use `ingest_salesforce()` for one-liner ingestion:
+
+```python
+from semantica.ingest import ingest_salesforce
+
+# Fetch records
+data = ingest_salesforce(
+    method="sobject",
+    sobject_name="Account",
+    fields=["Id", "Name", "Industry"],
+    limit=500,
+)
+
+# Execute raw SOQL (credentials from environment variables)
+data = ingest_salesforce(
+    method="query",
+    soql="SELECT Id, Name FROM Contact WHERE IsActive = true",
+)
+
+# Ingest + export to documents in one step
+docs = ingest_salesforce(
+    method="documents",
+    sobject_name="Account",
+    text_fields=["Name", "Description"],
+    limit=1000,
+)
+
+# List accessible sObjects
+sobject_names = ingest_salesforce(method="list_sobjects")
+```
+
+Or use the unified `ingest()` dispatcher:
+
+```python
+from semantica.ingest import ingest
+
+result = ingest(
+    None,
+    source_type="salesforce",
+    method="sobject",
+    sobject_name="Account",
+    fields=["Id", "Name"],
+    limit=500,
+)
+data = result["data"]   # SalesforceData
+```
+
+
+## Troubleshooting
+
+```python
+import os
+from semantica.ingest import SalesforceConnector
+
+connector = SalesforceConnector(
+    username=os.getenv("SALESFORCE_USERNAME"),
+    password=os.getenv("SALESFORCE_PASSWORD"),
+    security_token=os.getenv("SALESFORCE_SECURITY_TOKEN"),
+)
+if not connector.test_connection():
+    print("Connection failed: check username, password, security token, and domain")
+```
+
+Common causes of authentication failures:
+
+- **Wrong domain**: production orgs use `domain="login"`; sandboxes use `domain="test"`.
+- **Stale security token**: reset it under **Settings → Reset My Security Token**. The new token is emailed to you.
+- **IP restriction**: your org's trusted IP ranges may block the originating IP. Check **Setup → Network Access**.
+- **API access disabled**: ensure the connected profile has the **API Enabled** permission.
+
+
+## See Also
+
+- [Ingest Module](../reference/ingest) — Full `SalesforceIngestor` API and all other ingestors.
+- [Snowflake Integration](snowflake) — Relational warehouse connector with a similar design.
+- [Databricks Integration](databricks) — Lakehouse connector.
+- [Installation](../installation) — All optional dependency extras.
+- [Knowledge Graph](../reference/kg) — Build a KG from ingested Salesforce data.

pyproject.toml

@@ -124,11 +124,12 @@ shacl = ["pyshacl>=0.25.0"]
 db-snowflake = ["snowflake-connector-python>=4.6.0", "cryptography>=49.0.0"]
 db-databricks = ["databricks-sdk>=0.60.0", "databricks-sql-connector>=4.0.0"]
 db-arrow = ["pyarrow>=24.0.0"]
+db-salesforce = ["simple-salesforce>=1.12.0"]
 ingest-parquet = ["pyarrow>=24.0.0"]
 ingest-arrow = ["pyarrow>=24.0.0"]
 
 db-all = [
-  "semantica[db-snowflake,db-databricks,db-arrow]"
+  "semantica[db-snowflake,db-databricks,db-salesforce,db-arrow]"
 ]
 
 # ---- Embedding / Models ----

semantica/ingest/__init__.py

@@ -152,6 +152,7 @@ from .methods import (
     ingest_parquet,
     ingest_public_api,
     ingest_repository,
+    ingest_salesforce,
     ingest_stream,
     ingest_web,
     ingest_xml,
@@ -231,6 +232,10 @@ _LAZY_EXPORTS: Dict[str, Tuple[str, str]] = {
     # XML ingestion
     "XMLIngestor": (".xml_ingestor", "XMLIngestor"),
     "XMLIngestionData": (".xml_ingestor", "XMLIngestionData"),
+    # Salesforce ingestion
+    "SalesforceIngestor": (".salesforce_ingestor", "SalesforceIngestor"),
+    "SalesforceData": (".salesforce_ingestor", "SalesforceData"),
+    "SalesforceConnector": (".salesforce_ingestor", "SalesforceConnector"),
 }
 
 _OPTIONAL_DEPENDENCY_MESSAGES = {
@@ -258,6 +263,11 @@ _OPTIONAL_DEPENDENCY_MESSAGES = {
         "Arrow ingestion requires optional dependency 'pyarrow'. "
         "Install it before importing ArrowIngestor or using ingest_arrow()."
     ),
+    ".salesforce_ingestor": (
+        "Salesforce ingestion requires optional dependency 'simple-salesforce'. "
+        "Install it with: pip install \"semantica[db-salesforce]\" "
+        "or: pip install simple-salesforce>=1.12.0"
+    ),
 }
 
 
@@ -272,7 +282,7 @@ def __getattr__(name: str) -> Any:
     except ModuleNotFoundError as exc:
         message = _OPTIONAL_DEPENDENCY_MESSAGES.get(module_name)
         missing_name = getattr(exc, "name", None)
-        if message and missing_name in {"git", "bs4", "pyarrow"}:
+        if message and missing_name in {"git", "bs4", "pyarrow", "simple_salesforce"}:
             raise ImportError(message) from exc
         raise
 
@@ -358,6 +368,10 @@ __all__ = [
     # XML ingestion
     "XMLIngestor",
     "XMLIngestionData",
+    # Salesforce ingestion
+    "SalesforceIngestor",
+    "SalesforceData",
+    "SalesforceConnector",
     # Registry and Methods
     "MethodRegistry",
     "method_registry",
@@ -369,6 +383,7 @@ __all__ = [
     "ingest_repository",
     "ingest_email",
     "ingest_database",
+    "ingest_salesforce",
     "ingest_ontology",
     "ingest_arrow",
     "ingest_parquet",

semantica/ingest/config.py

@@ -186,8 +186,13 @@ class IngestConfig:
         self._method_configs[method] = config
 
     def get_method_config(self, method: str) -> Dict:
-        """Get method-specific configuration."""
-        return self._method_configs.get(method, {})
+        """Get method-specific configuration.
+
+        Returns a **copy** of the stored method configuration so callers can
+        safely mutate it (e.g. to merge per-call options) without poisoning the
+        global configuration for subsequent calls.
+        """
+        return dict(self._method_configs.get(method, {}))
 
     def get_all(self) -> Dict[str, Any]:
         """Get all configuration."""

semantica/ingest/ingest_usage.md

@@ -875,6 +875,96 @@ schema = connector.get_schema(engine)
         print(f"  {table_name}: {[col['name'] for col in columns]}")
 ```
 
+## Salesforce CRM Ingestion
+
+Salesforce ingestion requires `simple-salesforce`:
+
+```bash
+pip install "semantica[db-salesforce]"
+```
+
+### Basic Usage
+
+```python
+from semantica.ingest import SalesforceIngestor
+import os
+
+ingestor = SalesforceIngestor(
+    username=os.getenv("SALESFORCE_USERNAME"),
+    password=os.getenv("SALESFORCE_PASSWORD"),
+    security_token=os.getenv("SALESFORCE_SECURITY_TOKEN"),
+    domain="login",   # "test" for sandbox
+)
+
+# Ingest Account records
+data = ingestor.ingest_sobject(
+    "Account",
+    fields=["Id", "Name", "Industry", "BillingCity"],
+    where="Type = 'Customer'",
+    limit=5000,
+)
+print(f"Retrieved {data.row_count} of {data.total_size} matching records")
+```
+
+`SalesforceIngestor()` with no arguments reads from `SALESFORCE_USERNAME`, `SALESFORCE_PASSWORD`, `SALESFORCE_SECURITY_TOKEN`, and `SALESFORCE_DOMAIN` environment variables automatically.
+
+### Custom Objects and Raw SOQL
+
+```python
+# Custom object (API name ends in __c)
+data = ingestor.ingest_sobject("My_Custom_Object__c", fields=["Id", "Name", "Custom_Field__c"])
+
+# Raw SOQL GÇö pagination is handled automatically
+data = ingestor.ingest_query("""
+    SELECT Id, Name, StageName, Amount
+    FROM Opportunity
+    WHERE IsClosed = false
+    ORDER BY CloseDate ASC
+""")
+print(f"Open opportunities: {data.row_count}")
+```
+
+### Document Export
+
+```python
+documents = ingestor.export_as_documents(
+    data,
+    id_field="Id",                      # Salesforce 18-char record Id
+    text_fields=["Name", "Description"],
+)
+# Each document: {"id": "001...", "text": "...", "metadata": {"source": "salesforce", ...}}
+```
+
+### Convenience Function
+
+```python
+from semantica.ingest import ingest_salesforce
+
+# Fetch records
+data = ingest_salesforce(
+    method="sobject",
+    sobject_name="Account",
+    fields=["Id", "Name"],
+    limit=500,
+)
+
+# Ingest and export as documents in one call
+docs = ingest_salesforce(
+    method="documents",
+    sobject_name="Account",
+    text_fields=["Name", "Description"],
+)
+
+# Using the unified dispatcher
+from semantica.ingest import ingest
+result = ingest(None, source_type="salesforce", method="sobject",
+                sobject_name="Account", fields=["Id", "Name"])
+data = result["data"]
+```
+
+See [Salesforce Integration](https://docs.getsemantica.ai/integrations/salesforce) for full documentation including sandbox, schema discovery, pagination details, and troubleshooting.
+
+
 ## MCP Server Ingestion
 
 **IMPORTANT**: This implementation supports **ONLY Python-based MCP servers and FastMCP servers**. Users can bring their own Python or FastMCP MCP servers via URL connections. JavaScript, TypeScript, C#, Java, and other language implementations are **NOT supported**.

semantica/ingest/methods.py

@@ -203,6 +203,7 @@ if TYPE_CHECKING:
     from .ontology_ingestor import OntologyData
     from .parquet_ingestor import ParquetData
     from .public_api_ingestor import PublicAPIDetection
+    from .salesforce_ingestor import SalesforceData
     from .stream_ingestor import StreamProcessor
     from .web_ingestor import WebContent
     from .xml_ingestor import XMLIngestionData
@@ -1126,6 +1127,213 @@ def ingest_database(
         raise
 
 
+def ingest_salesforce(
+    source: Optional[Dict[str, Any]] = None,
+    method: str = "sobject",
+    **kwargs,
+) -> Union["SalesforceData", List[Dict[str, Any]], Dict[str, Any]]:
+    """Ingest data from Salesforce CRM (convenience function).
+
+    A user-friendly wrapper around :class:`~semantica.ingest.SalesforceIngestor`
+    that connects, ingests, and returns data in a single call.
+
+    Args:
+        source: Optional credential/configuration dictionary.  Keys mirror the
+            :class:`~semantica.ingest.SalesforceConnector` constructor:
+            ``username``, ``password``, ``security_token``, ``domain``
+            (``"login"`` for production, ``"test"`` for sandbox),
+            ``instance_url``, ``session_id``, ``api_version``.
+            When ``None``, credentials are read from environment variables
+            (``SALESFORCE_USERNAME`` / ``SALESFORCE_PASSWORD`` /
+            ``SALESFORCE_SECURITY_TOKEN`` etc.).
+        method: Ingestion method:
+
+            * ``"sobject"`` *(default)* — fetch records from a named sObject
+              (requires ``sobject_name`` kwarg).
+            * ``"query"`` — execute a raw SOQL query string (requires
+              ``soql`` kwarg).
+            * ``"list_sobjects"`` — return a sorted list of accessible sObject
+              API names.
+            * ``"schema"`` — return field metadata for a named sObject
+              (requires ``sobject_name`` kwarg).
+            * ``"documents"`` — ingest an sObject and convert to the Semantica
+              document format in one step (requires ``sobject_name`` kwarg;
+              optional ``text_fields`` and ``id_field`` kwargs).
+
+        **kwargs: Additional options forwarded to the ingestor method.
+            Common kwargs for ``"sobject"`` / ``"documents"``:
+
+            * ``sobject_name`` — Salesforce sObject API name (e.g.
+              ``"Account"``, ``"My_Custom__c"``).
+            * ``fields`` — list of field API names to select.  When omitted
+              all selectable fields are fetched via ``describe()``.
+            * ``where`` — SOQL ``WHERE`` clause fragment (trusted input only).
+            * ``order_by`` — SOQL ``ORDER BY`` clause fragment.
+            * ``limit`` — maximum number of records.
+
+            For ``"query"``:
+
+            * ``soql`` — full SOQL query string.
+
+            For ``"schema"``:
+
+            * ``sobject_name`` — sObject to describe.
+
+    Returns:
+        * ``"sobject"`` / ``"query"`` → :class:`~semantica.ingest.SalesforceData`
+        * ``"documents"`` → ``List[Dict[str, Any]]`` (Semantica document format)
+        * ``"list_sobjects"`` → ``List[str]``
+        * ``"schema"`` → ``Dict[str, Any]``
+
+    Raises:
+        :class:`~semantica.utils.exceptions.ConfigurationError`: If
+            ``simple-salesforce`` is not installed.
+        :class:`~semantica.utils.exceptions.ValidationError`: If credentials
+            are incomplete or an sObject / field name is invalid.
+        :class:`~semantica.utils.exceptions.ProcessingError`: If the
+            Salesforce API call fails.
+
+    Examples::
+
+        >>> from semantica.ingest import ingest_salesforce
+
+        >>> # Fetch Account records (credentials from env vars)
+        >>> data = ingest_salesforce(
+        ...     method="sobject",
+        ...     sobject_name="Account",
+        ...     fields=["Id", "Name", "Industry"],
+        ...     limit=500,
+        ... )
+
+        >>> # Execute a raw SOQL query (credentials from environment variables)
+        >>> data = ingest_salesforce(
+        ...     method="query",
+        ...     soql="SELECT Id, Name FROM Contact WHERE IsActive = true",
+        ... )
+
+        >>> # Ingest and export as documents for GraphBuilder in one step
+        >>> docs = ingest_salesforce(
+        ...     method="documents",
+        ...     sobject_name="Account",
+        ...     text_fields=["Name", "Description"],
+        ...     limit=1000,
+        ... )
+
+        >>> # List all accessible sObjects in the connected org
+        >>> sobject_names = ingest_salesforce(method="list_sobjects")
+
+        >>> # Use sandbox org
+        >>> data = ingest_salesforce(
+        ...     method="sobject",
+        ...     sobject_name="Account",
+        ... )  # set SALESFORCE_DOMAIN=test in environment for sandbox
+    """
+    # Registry hook — allows callers to register a custom "salesforce" method
+    custom_method = method_registry.get("salesforce", method)
+    if custom_method and custom_method != ingest_salesforce:
+        fallback = kwargs.pop("fallback_on_custom_error", False)
+        result = call_custom_method(
+            logger, method, custom_method, source,
+            fallback_on_custom_error=fallback, **kwargs,
+        )
+        if result is not CUSTOM_METHOD_FELL_BACK:
+            return result
+
+    try:
+        from .salesforce_ingestor import SalesforceIngestor
+    except ModuleNotFoundError as exc:
+        if _is_missing_dependency(exc, "simple_salesforce"):
+            raise _missing_optional_dependency(
+                "Salesforce ingestion", "simple-salesforce"
+            ) from exc
+        raise
+
+    # Unpack credential dict (if given); everything else stays in kwargs.
+    creds: Dict[str, Any] = {}
+    if source is not None:
+        if not isinstance(source, dict):
+            raise ProcessingError(
+                "ingest_salesforce() source must be a credential dict or None. "
+                "Pass sobject_name / soql as keyword arguments."
+            )
+        creds = dict(source)
+
+    # Merge any ingest_config method config under "salesforce".
+    # get_method_config() now returns a copy, so this dict is safe to mutate.
+    # We build the final connector config in order of increasing priority:
+    #   1. base method config (lowest — global defaults set by operator)
+    #   2. per-call credential dict supplied via `source`
+    #   3. per-call connector params supplied as kwargs
+    # Credentials are extracted from kwargs and removed so they don't also
+    # flow into the ingest method call (which doesn't understand them).
+    _CONNECTOR_PARAMS = frozenset({
+        "username", "password", "security_token", "domain",
+        "instance_url", "session_id", "api_version",
+    })
+    connector_kwargs = {k: v for k, v in kwargs.items() if k in _CONNECTOR_PARAMS}
+    for k in _CONNECTOR_PARAMS:
+        kwargs.pop(k, None)
+
+    # Build a fresh per-call config dict — never mutate the global store.
+    config: Dict[str, Any] = {
+        **ingest_config.get_method_config("salesforce"),  # base (already a copy)
+        **creds,                                           # source dict credentials
+        **connector_kwargs,                                # kwarg credentials
+    }
+
+    ingestor = SalesforceIngestor(**config)
+
+    if method == "sobject":
+        sobject_name = kwargs.pop("sobject_name", None)
+        if not sobject_name:
+            raise ProcessingError(
+                "ingest_salesforce() with method='sobject' requires "
+                "sobject_name keyword argument."
+            )
+        return ingestor.ingest_sobject(sobject_name, **kwargs)
+
+    elif method == "query":
+        soql = kwargs.pop("soql", None)
+        if not soql:
+            raise ProcessingError(
+                "ingest_salesforce() with method='query' requires "
+                "soql keyword argument."
+            )
+        return ingestor.ingest_query(soql, **kwargs)
+
+    elif method == "list_sobjects":
+        return ingestor.list_sobjects()
+
+    elif method == "schema":
+        sobject_name = kwargs.pop("sobject_name", None)
+        if not sobject_name:
+            raise ProcessingError(
+                "ingest_salesforce() with method='schema' requires "
+                "sobject_name keyword argument."
+            )
+        return ingestor.get_sobject_schema(sobject_name)
+
+    elif method == "documents":
+        sobject_name = kwargs.pop("sobject_name", None)
+        if not sobject_name:
+            raise ProcessingError(
+                "ingest_salesforce() with method='documents' requires "
+                "sobject_name keyword argument."
+            )
+        id_field = kwargs.pop("id_field", "Id")
+        text_fields = kwargs.pop("text_fields", None)
+        data = ingestor.ingest_sobject(sobject_name, **kwargs)
+        return ingestor.export_as_documents(data, id_field=id_field,
+                                            text_fields=text_fields)
+
+    else:
+        raise ProcessingError(
+            f"Unknown ingest_salesforce method: {method!r}. "
+            "Valid methods: 'sobject', 'query', 'list_sobjects', 'schema', "
+            "'documents'."
+        )
+
+
 def ingest_mcp(
     source: Union[str, Dict[str, Any]],
     method: str = "resources",
@@ -1306,6 +1514,7 @@ def ingest(
             - "ontology": Ontology ingestion
             - "parquet": Apache Parquet file or directory ingestion
             - "xml": XML file or directory ingestion
+            - "salesforce": Salesforce CRM ingestion (pass credentials via kwargs)
         method: Optional specific ingestion method
         **kwargs: Additional options passed to ingestor
 
@@ -1428,6 +1637,9 @@ def ingest(
         return {"ontology": ingest_ontology(sources, method=method or "file", **kwargs)}
     elif source_type == "mcp":
         return {"data": ingest_mcp(sources, method=method or "resources", **kwargs)}
+    elif source_type == "salesforce":
+        return {"data": ingest_salesforce(sources if isinstance(sources, dict) else None,
+                                          method=method or "sobject", **kwargs)}
     else:
         raise ProcessingError(f"Unknown source type: {source_type}")
 
@@ -1540,3 +1752,9 @@ method_registry.register("ontology", "file", ingest_ontology)
 method_registry.register("ontology", "directory", ingest_ontology)
 method_registry.register("ingest", "default", ingest)
 method_registry.register("ingest", "unified", ingest)
+method_registry.register("salesforce", "default", ingest_salesforce)
+method_registry.register("salesforce", "sobject", ingest_salesforce)
+method_registry.register("salesforce", "query", ingest_salesforce)
+method_registry.register("salesforce", "list_sobjects", ingest_salesforce)
+method_registry.register("salesforce", "schema", ingest_salesforce)
+method_registry.register("salesforce", "documents", ingest_salesforce)

semantica/ingest/registry.py

@@ -66,6 +66,7 @@ class MethodRegistry:
         "parquet": {},
         "arrow": {},
         "xml": {},
+        "salesforce": {},
         "ingest": {},
     }