source-connector-implementation

Building a Source Connector in Airweave

Overview

A source connector in Airweave is a Python module that extracts data from an external service and transforms it into searchable entities. This guide covers everything you need to build a production-ready connector.

There are two types of source connectors:

1. Standard (Sync-Based): Extracts and syncs all data from the source to Airweave's vector database
2. Federated Search: Searches the source's API at query time without syncing data

Core Components

Every source connector requires three main components:

1. Source implementation (backend/airweave/platform/sources/{short_name}.py)
2. Entity schemas (backend/airweave/platform/entities/{short_name}.py)
3. OAuth configuration (backend/airweave/platform/auth/yaml/dev.integrations.yaml)

---

Part 1: Entity Schemas

Start with entities because they define your data model.

File Location

backend/airweave/platform/entities/{short_name}.py

Entity Types

There are two base entity types:

1. ChunkEntity - Text-based entities (tasks, messages, documents, etc.)
2. FileEntity - File attachments (PDFs, images, etc.)

Basic Structure

"""Entity schemas for {Connector Name}."""

from datetime import datetime
from typing import Any, Dict, List, Optional

from pydantic import Field

from airweave.platform.entities._airweave_field import AirweaveField
from airweave.platform.entities._base import ChunkEntity, FileEntity


class MyConnectorEntity(ChunkEntity):
    """Schema for primary entity type."""

    # Required fields
    name: str = AirweaveField(
        ...,
        description="Display name of the entity",
        embeddable=True  # This field will be embedded for search
    )

    # Timestamps (critical for incremental sync)
    created_at: Optional[datetime] = AirweaveField(
        None,
        description="When this entity was created",
        embeddable=True,
        is_created_at=True  # Marks this as the creation timestamp
    )

    modified_at: Optional[datetime] = AirweaveField(
        None,
        description="When this entity was last modified",
        embeddable=True,
        is_updated_at=True  # Marks this as the update timestamp
    )

    # Content fields
    content: Optional[str] = AirweaveField(
        None,
        description="The main text content",
        embeddable=True  # Make searchable
    )

    # Metadata fields (not embeddable)
    external_id: str = Field(
        ...,
        description="Unique ID from the external system"
    )

    permalink_url: Optional[str] = Field(
        None,
        description="Direct link to view in external system"
    )