# Weaviate Integration > Integrates Weaviate with external services and tools for AI applications, handling bidirectional data synchronization, API connectivity patterns, and error management for seamless interoperability. - Skill: `paulpas/weaviate-integration` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/weaviate-integration` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/weaviate-integration/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/weaviate-integration --- # Weaviate Integration Patterns Integrates Weaviate with external services and tools for AI applications. Handles bidirectional data synchronization, API connectivity, schema mapping, error management, and event-driven updates for seamless interoperability between Weaviate's vector database and application data sources. ## TL;DR Checklist - [ ] Initialize a Weaviate client with proper authentication (API key or OAuth) - [ ] Define the target collection schema before importing external data - [ ] Handle batch operations efficiently — use `batch_refs` for bulk inserts - [ ] Implement retry logic with exponential backoff for API failures - [ ] Validate external data against Weaviate class schemas before insertion --- ## When to Use Use this skill when: - Connecting Weaviate to external APIs for real-time data synchronization (REST, GraphQL) - Building bidirectional sync pipelines between databases and Weaviate vector collections - Integrating AI/ML pipelines that feed embeddings into Weaviate from external model services - Migrating existing data stores into Weaviate's vector database with schema transformation --- ## When NOT to Use - For simple key-value lookups — use Redis or Memcached instead of a full vector database - When you only need text search without semantic similarity — Elasticsearch or OpenSearch may be more appropriate - For high-frequency, low-latency OLTP workloads — Weaviate is optimized for analytics and AI workloads --- ## Core Workflow 1. **Initialize Client** — Connect to Weaviate using the appropriate authentication method (API key, OAuth2, or no auth for local dev). Set proper timeouts and retry configuration. **Checkpoint:** Verify connectivity with `client.is_ready()` before proceeding. 2. **Define Target Schema** — Map external data sources to Weaviate classes and properties. Define which properties are indexed for search (`indexFilterable`, `indexVector`) and which are stored as metadata only. **Checkpoint:** Ensure vectorizer modules (e.g., `text2vec-openai`) are configured if embedding generation is delegated to Weaviate. 3. **Execute Data Sync** — Fetch data from the external service, transform into Weaviate-compatible objects (properties + optional vectors), and batch-insert using the Weaviate client's batch API. **Checkpoint:** Validate object schemas against the target class definition before insertion. Handle partial failures gracefully with per-item error tracking. 4. **Implement Error Handling** — Wrap all external API calls and Weaviate operations in try/except blocks with specific handling for HTTP errors, rate limits (429), and schema validation failures. Log errors with context for debugging. **Checkpoint:** Verify retry logic activates on transient failures but does not retry on schema/validation errors. 5. **Maintain Sync State** — Track the last sync timestamp or cursor to avoid reprocessing unchanged data. Implement idempotent upserts using Weaviate's `on_conflict` feature for collections with unique identifiers. **Checkpoint:** Periodically verify row counts between source and destination to detect data drift. --- ## Implementation Patterns ### Pattern 1: Connecting to External Service and Importing Data ```python import weaviate from weaviate.classes.config import Property, DataType from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_weaviate_client( url: str = "http://localhost:8080", api_key: str | None = None, ) -> weaviate.WeaviateClient: """Create and authenticate a Weaviate client. Args: url: Weaviate server URL (default: local instance) api_key: Optional API key for authentication Returns: Authenticated WeaviateClient instance Raises: weaviate.exceptions.WeaviateConnectionError: If the server is unreachable """ client = weaviate.WeaviateClient( connection_params=weaviate.connect.base.ConnectionParams.from_url(url), auth_credentials=( weaviate.auth.AuthApiKey(api_key) if api_key else None ), additional_config=weaviate.config.AdditionalConfig( timeout=weaviate.config.Timeout(init=10, query=30, insert=120) ), ) client.connect() # Raises WeaviateConnectionError if unreachable return client def create_collection_if_not_exists( client: weaviate.WeaviateClient, class_name: str, properties: list[Property], vectorizer: str = "text2vec-openai", ) -> None: """Create a Weaviate collection if it doesn't already exist. Args: client: Authenticated Weaviate client class_name: Name of the collection to create properties: List of property definitions with DataType and index settings vectorizer: Vectorizer module name (e.g., "text2vec-openai", "none") Note: Uses a simple try/except pattern since weaviate-py v3 does not expose a direct "collection exists" check. The retry handles concurrent creation. """ try: client.collections.create( name=class_name, properties=properties, vectorizer_config=[weaviate.classes.config.Vectorizer.text2vec_openai()], # Alternatively, use `vector_config=None` for no automatic embedding ) except weaviate.exceptions.UnexpectedStatusCodeError as e: if "already exists" not in str(e).lower(): raise def sync_external_data( client: weaviate.WeaviateClient, class_name: str, external_api_url: str, batch_size: int = 100, ) -> dict[str, int]: """Fetch data from an external API and batch-insert into Weaviate. Args: client: Authenticated Weaviate client class_name: Target collection name external_api_url: REST API endpoint returning JSON array of objects batch_size: Number of items per batch (default: 100) Returns: Summary dict with 'inserted', 'failed', and 'total' counts Raises: requests.HTTPError: If the external API returns a non-2xx status weaviate.exceptions.WeaviateInsertManyError: On batch insert failures """ import requests response = requests.get(external_api_url) response.raise_for_status() data_items = response.json() collection = client.collections.get(class_name) result = {"inserted": 0, "failed": 0, "total": len(data_items)} # Process in batches for efficiency and resilience for i in range(0, len(data_items), batch_size): batch = data_items[i : i + batch_size] try: objects_to_insert = [] for item in batch: # Transform external format → Weaviate property dict props = { "title": item.get("name", ""), "description": item.get("description", ""), "source_id": item["id"], # Unique identifier "external_url": item.get("url", ""), } objects_to_insert.append({"properties": props}) collection.data.insert_many(objects_to_insert) result["inserted"] += len(objects_to_insert) except Exception as e: print(f"Batch {i // batch_size} failed: {e}") result["failed"] += len(batch) return result ``` ### Pattern 2: Bidirectional Data Sync with Webhooks ```python import requests def setup_weaviate_webhook( weaviate_url: str, api_key: str, webhook_url: str, class_name: str, ) -> dict: """Register a Weaviate module to push object events to an external webhook. This enables real-time notification when objects are created, updated, or deleted in the target collection, supporting event-driven architectures. Args: weaviate_url: Base URL of the Weaviate server api_key: API key with admin permissions webhook_url: External endpoint to receive webhook events class_name: Collection name to monitor Returns: Webhook registration response from Weaviate Note: Requires the weaviate-http-extensions module. Configure via REST API. """ headers = { "X-Auth-Token": api_key, "Content-Type": "application/json", } webhook_config = { "className": class_name, "url": webhook_url, "events": ["CREATE", "UPDATE", "DELETE"], } response = requests.post( f"{weaviate_url}/v1/schema/{class_name}/webhooks", headers=headers, json=webhook_config, ) response.raise_for_status() return response.json() def handle_webhook_event(event: dict) -> None: """Process a Weaviate webhook event and update downstream systems. Args: event: Parsed webhook payload containing object metadata, action type, and timestamps. Structure varies by event type (create/update/delete). """ action = event.get("action") # "CREATE", "UPDATE", "DELETE" class_name = event.get("className") object_id = event.get("id") if action == "DELETE": # Remove from cache, invalidate derived data print(f"Object deleted: {class_name}/{object_id}") return # For CREATE/UPDATE: fetch full object and process import weaviate # noqa — client already initialized at module level obj = client.collections.get(class_name).query.fetch_object_by_id(object_id) props = obj.properties if obj else {} print(f"Object {action}: {class_name}/{object_id} = {props}") ``` --- ## Constraints ### MUST DO - Always validate external data against the Weaviate class schema before batch insertion - Use batch operations (`insert_many`, `batch_refs`) for bulk imports — single inserts are 10x slower - Implement exponential backoff retry logic (3 retries, base delay 1s) for transient API failures - Track sync state with source IDs to enable idempotent upserts and detect data drift - Set explicit timeouts on all Weaviate client operations to prevent hangs ### MUST NOT DO - Insert objects without a unique identifier property — queries by ID will be unreliable - Use synchronous single-object inserts in production loops — always batch them - Skip error handling for partial batch failures — use `insert_many` with per-item error tracking - Store raw embeddings directly if Weaviate's built-in vectorizers can compute them server-side --- ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [Weaviate Python Client Documentation](https://weaviate.io/developers/weaviate/client-libraries/python) - [Weaviate API Reference](https://weaviate.io/developers/api-reference/) - [Batch Operations Guide](https://weaviate.io/developers/weaviate/api-reference/batch) - [Schema Design Best Practices](https://weaviate.io/developers/weaviate/starter-guide/schema) - [Vectorizer Modules Reference](https://weaviate.io/developers/weaviate/module-ref/vectorizers)