Data Ingestion Service
Relevant source files
The following files were used as context for generating this wiki page:
The Data Ingestion Service is a specialized component within the weaviate-mcp package designed to automate the process of downloading, processing, and vectorizing documents from external sources into a Weaviate vector database. It provides a robust pipeline that handles multiple content types, including HTML, plain text, and JSON, while ensuring content is optimized for Large Language Model (LLM) consumption through intelligent chunking strategies.
The service acts as a bridge between raw data sources and Weaviate, implementing token-aware semantic splitting to preserve context and performance. It is exposed to AI assistants via the Model Context Protocol (MCP) as discrete tools, allowing models to dynamically expand their knowledge base by ingesting relevant URLs or raw text.
Sources: packages/weaviate-mcp/src/weaviate_mcp/services/ingestion_service.py:1-12, packages/weaviate-mcp/src/weaviate_mcp/tools/ingestion_tools.py:1-12, packages/weaviate-mcp/README.md:144-154
Core Architecture and Data Flow
The ingestion system is structured into a tool layer and a service layer. The tool layer (ingestion_tools.py) validates inputs and manages high-level execution, while the service layer (ingestion_service.py) contains the core logic for content extraction and chunking.
Ingestion Pipeline Flow
This diagram illustrates the sequence of operations when a URL is ingested into the system.
The process begins with an asynchronous download using httpx. Once the content is retrieved, the service detects the MIME type to apply appropriate extraction logic. For HTML, it utilizes BeautifulSoup to strip boilerplate elements like scripts, styles, and navigation menus.
Sources: packages/weaviate-mcp/src/weaviate_mcp/services/ingestion_service.py:35-85, packages/weaviate-mcp/src/weaviate_mcp/tools/ingestion_tools.py:45-103
Technical Implementation
IngestionService Class
The IngestionService is the primary orchestrator. It is initialized with a WeaviateService and utilizes tiktoken with the cl100k_base encoding (compatible with GPT-4) for precise token counting.
| Method | Description |
|---|---|
ingest_from_url | Coordinates download, extraction, chunking, and storage for a given URL. |
_create_optimal_chunks | Implements the core semantic splitting logic using token limits. |
_extract_html_content | Cleans HTML and extracts metadata like title, description, and author. |
_split_large_text | Recursive method to break down text that exceeds token limits. |
Sources: packages/weaviate-mcp/src/weaviate_mcp/services/ingestion_service.py:21-33, packages/weaviate-mcp/src/weaviate_mcp/services/ingestion_service.py:175-188
Optimal Chunking Strategy
The service employs a multi-stage chunking strategy to ensure semantic coherence:
- Paragraph Splitting: Initially attempts to split text by
\n\n. - Sentence Splitting: If paragraphs are too large or absent, it falls back to regex-based sentence detection.
- Token Validation: Uses
tiktokento ensure no chunk exceeds themax_tokens_per_chunklimit. - Overlap Management: Prepends a portion of the previous chunk to the current one based on
overlap_tokensto maintain context across boundaries.
Sources: packages/weaviate-mcp/src/weaviate_mcp/services/ingestion_service.py:246-300
Exposed MCP Tools
The service is accessible through two primary MCP tools defined in ingestion_tools.py.
1. weaviate_ingest_from_url
Downloads and processes content from a public URL.
| Parameter | Type | Default | Description |
|---|---|---|---|
url | string | - | The source URL (HTML, Text, or JSON). |
collection_name | string | - | The target Weaviate collection. |
chunk_size | integer | 1000 | Target character count (fallback). |
chunk_overlap | integer | 100 | Character/Token overlap for context. |
max_tokens_per_chunk | integer | 500 | Hard token limit per chunk. |
Sources: packages/weaviate-mcp/src/weaviate_mcp/tools/ingestion_tools.py:21-43
2. weaviate_ingest_text_content
Allows direct ingestion of raw text strings, useful for documents already present in memory or provided by the user.
# Implementation snippet from ingestion_tools.py
objects = []
for i, chunk in enumerate(chunks):
obj = {
"content": chunk,
"source_url": source_identifier,
"chunk_index": i,
"total_chunks": len(chunks),
"content_type": "text/plain",
"title": title or f"Text Content {source_identifier}",
}
objects.append(obj)
Sources: packages/weaviate-mcp/src/weaviate_mcp/tools/ingestion_tools.py:168-177
Data Consistency and Error Handling
The service implements duplicate prevention by using a combination of source_url (or source_identifier) and chunk_index as unique properties during batch insertion. This ensures that re-ingesting the same URL does not create duplicate entries in the vector database.
Error Propagation
- Validation: Tools validate that URLs and collection names are not empty and that token counts are positive.
- Service Errors: Errors during HTTP download (e.g., 404, timeouts) are captured and returned as descriptive messages in a dictionary containing
error: True. - Unexpected Exceptions: Any uncaught exceptions are wrapped in a
ValueErrorwith the original error message to adhere to the MCP error handling pattern.
Sources: packages/weaviate-mcp/src/weaviate_mcp/tools/ingestion_tools.py:92-103, tests/weaviate-mcp/unit/tools/test_ingestion_tools.py:195-215
Conclusion
The Data Ingestion Service provides a robust, production-ready pipeline for transforming unstructured web content into structured vector data. By combining intelligent HTML cleaning with token-aware semantic chunking, it ensures that AI assistants can leverage external data with high accuracy and contextual integrity within the Weaviate ecosystem.
Sources: packages/weaviate-mcp/README.md:21-30, packages/weaviate-mcp/src/weaviate_mcp/services/ingestion_service.py:1-10