What Enterprise Data Sources Can Semantica Ingest? A Complete Guide to the Ingest Module

Semantica can ingest enterprise data from 12+ source categories including Snowflake warehouses, Kafka streams, cloud storage (S3/GCS), relational databases, MCP servers, Git repositories, and web APIs through a unified ingest() API backed by pluggable ingestor classes.

The semantica-agi/semantica repository provides a comprehensive ingestion framework designed to handle diverse enterprise data sources through a single, extensible interface. The ingest module consolidates cloud storage connectors, database drivers, streaming clients, and web crawlers into standardized ingestor classes that expose a common API while managing source-specific authentication and connection pooling.

Core Architecture of the Ingest Module

The ingestion framework centers on a method registry pattern implemented in semantica/ingest/registry.py. This registry maps source types and file extensions to specific ingestor implementations, enabling the high-level ingest() function to automatically route requests to the correct handler.

Each ingestor follows a consistent lifecycle:

1. Source Detection – The system inspects input identifiers (file paths, URLs, or connection strings) to select the appropriate ingestor class (e.g., FileIngestor, DBIngestor, MCPIngestor).

2. Connector Initialization – Ingestors instantiate thin connector objects that handle enterprise-grade concerns including authentication, connection pooling, and timeout management. Configuration is centralized in semantica/ingest/config.py, supporting both environment variables and YAML configuration files.

3. Processing Mode – For high-volume enterprise sources, ingestors support both streaming (lazy iteration with back-pressure control) and batch processing modes.

4. Unified Output – All ingestors return standardized data objects (file metadata dictionaries, pandas-like tables, or resource lists) that downstream pipelines consume without source-specific adapters.

Supported Enterprise Data Sources

File Systems and Cloud Storage

The FileIngestor class in semantica/ingest/file_ingestor.py handles local filesystem paths and cloud storage protocols including s3://, gs://, and Azure Blob Storage. It supports binary blobs, PDFs, Office documents, and structured formats.

For analytical workloads, the ParquetIngestor processes columnar data files and Hive-style directory structures, enabling bulk ingestion of event logs and analytical datasets without schema inference overhead.

Databases and Data Warehouses

Relational databases are supported through DBIngestor in semantica/ingest/db_ingestor.py, which leverages SQLAlchemy to connect to PostgreSQL, MySQL, and SQLite instances using standard connection strings.

For enterprise data warehousing, SnowflakeIngestor in semantica/ingest/snowflake_ingestor.py provides native Snowflake account integration, supporting warehouse-specific authentication and compute resource allocation.

Streaming Platforms

Real-time event ingestion is handled by StreamIngestor in semantica/ingest/stream_ingestor.py. This implementation supports enterprise streaming platforms including Apache Kafka, RabbitMQ, AWS Kinesis, and Apache Pulsar. The ingestor yields messages lazily to maintain back-pressure control and implements configurable batching for high-throughput scenarios.

Web and API Sources

Web-based enterprise data sources utilize three specialized ingestors:

• WebIngestor (semantica/ingest/web_ingestor.py): Crawls URLs, sitemaps, and entire domains while respecting robots.txt and implementing configurable crawl delays.

• PublicAPIIngestor (semantica/ingest/public_api_ingestor.py): Consumes REST endpoints with automatic content-type detection (JSON, CSV, XML) for open data portals and demo APIs.

• FeedIngestor (semantica/ingest/feed_ingestor.py): Parses RSS and Atom feeds for news aggregation and alert stream processing.

Specialized Enterprise Sources

Several ingestors target specific enterprise infrastructure:

• RepoIngestor (semantica/ingest/repo_ingestor.py): Clones Git repositories and enumerates versioned files with commit metadata for codebase analysis.

• EmailIngestor (semantica/ingest/email_ingestor.py): Connects to IMAP and POP3 servers to retrieve messages, analyze threads, and extract attachments for ticketing systems and compliance archives.

• MCPIngestor (semantica/ingest/mcp_ingestor.py): Implements the Model-Context-Protocol to ingest resources from Python/FastMCP servers, providing unified access to custom enterprise services including proprietary databases and SaaS APIs.

Code Examples for Enterprise Pipelines

Ingesting Mixed Cloud and Local Files

from semantica.ingest import ingest_file

files = ingest_file(
    ["./reports/annual.pdf", "s3://finance-bucket/q1.xlsx"],
    method="cloud",
    provider="s3",
    access_key_id="YOUR_KEY",
    secret_access_key="YOUR_SECRET",
)
print(f"Ingested {len(files)} files")

Querying Snowflake Data Warehouses

from semantica.ingest import ingest_database, SnowflakeIngestor

sf = SnowflakeIngestor()
sf_conn = sf.connect(
    account="myaccount",
    user="etl_user",
    password="********",
    warehouse="COMPUTE_WH",
    database="FINANCE",
    schema="PUBLIC",
)
table = ingest_database(sf_conn, table="TRANSACTIONS", limit=5000)
print(table.row_count, table.columns)

Streaming Kafka Events to Vector Storage

from semantica.ingest import ingest_stream
from semantica.vector_store import FaissStore

kafka_cfg = {"topic": "events", "bootstrap_servers": ["kafka:9092"]}
processor = ingest_stream(kafka_cfg, method="kafka")
faiss = FaissStore(dim=768)

def handle(msg):
    vec = msg.embedding
    faiss.upsert([vec], ids=[msg.id])

processor.set_message_handler(handle)
processor.start_consuming()

Crawling Corporate Intranets

from semantica.ingest import ingest_web, WebIngestor

crawler = WebIngestor(
    delay=1.5, 
    respect_robots=True, 
    user_agent="SemanticaBot/1.0"
)
pages = ingest_web(
    "https://intranet.example.com", 
    method="crawl", 
    max_pages=200
)
print(f"Crawled {len(pages)} pages")

Accessing MCP Server Resources

from semantica.ingest import MCPIngestor

mcp = MCPIngestor()
mcp.connect(
    "doc_server", 
    url="https://docs.example.com/mcp",
    headers={"Authorization": "Bearer <token>"}
)
pdfs = mcp.ingest_resources(
    "doc_server",
    filter_func=lambda r: r.uri.endswith(".pdf")
)
print(f"Found {len(pdfs)} PDFs")

Summary

• Unified API: The ingest() function in semantica-agi/semantica routes to 12+ source-specific ingestor classes through a central registry.
• Enterprise Coverage: Native support for Snowflake, Kafka, S3/GCS, PostgreSQL, Git, IMAP, and MCP servers without custom adapters.
• Extensible Architecture: New enterprise sources are added by registering classes in semantica/ingest/registry.py while maintaining the same connector pattern.
• Configuration Management: Enterprise authentication and connection pooling are handled via semantica/ingest/config.py using environment variables or YAML files.

Frequently Asked Questions

How does Semantica handle authentication for enterprise data sources?

Authentication is managed through the semantica/ingest/config.py configuration layer, which supports environment variables and YAML configuration files. Each ingestor initializes connectors that read credentials from this centralized config, enabling enterprise-grade security patterns including connection pooling and timeout handling without hardcoding secrets in ingestion scripts.

Can Semantica ingest from proprietary or custom internal APIs?

Yes. While PublicAPIIngestor handles standard REST endpoints, the MCPIngestor class in semantica/ingest/mcp_ingestor.py connects to Python/FastMCP servers that can wrap proprietary services. Additionally, developers can extend the framework by implementing the ingestor interface and registering the new class in semantica/ingest/registry.py to ingest from custom enterprise systems.

What is the difference between batch and streaming ingestion in Semantica?

Batch processing loads complete datasets into memory (suitable for Snowflake queries or Parquet files), while streaming ingestion through StreamIngestor yields items lazily from sources like Kafka or RabbitMQ. The streaming mode implements back-pressure control and configurable batch sizes, allowing continuous ingestion of high-volume enterprise event streams without memory exhaustion.

Does Semantica support schema inference for ingested data?

The ingestors return unified data objects (file metadata dictionaries, pandas-like tables, or resource lists) that standardize output regardless of source. For structured sources like databases, the DBIngestor preserves SQLAlchemy column types, while ParquetIngestor maintains native columnar schemas. Downstream components receive consistent interfaces without requiring source-specific parsing logic.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}