Skip to main content
The Omni Connector SDKs enable you to build custom connectors that integrate any data source with Omni’s search and AI capabilities. Whether you’re connecting an internal database, a SaaS application, or a proprietary system, the SDKs provide everything you need.

​
Why Build a Custom Connector?

While Omni includes connectors for popular platforms (Google Workspace, Slack, Atlassian, HubSpot), you may need to integrate:
  • Internal systems — Databases, wikis, or custom applications
  • Niche SaaS tools — Industry-specific platforms not yet supported
  • Proprietary data sources — File formats or APIs unique to your organization
  • Legacy systems — Older systems with custom APIs

​
SDK Features

Both SDKs provide the same core capabilities:
FeatureDescription
Connector Base ClassAbstract base with sync lifecycle management
Sync ContextUtilities for emitting documents and tracking progress
Content StorageStore document content for indexing
SDK ClientCommunication with the connector-manager service
HTTP ServerBuilt-in server exposing standard connector endpoints
Data ModelsType-safe models for documents, events, and metadata

​
Available SDKs

Python SDK

Build connectors in Python with FastAPI

TypeScript SDK

Build connectors in TypeScript with Express

​
Choosing an SDK

Choose Python if…Choose TypeScript if…
Your team is experienced with PythonYour team prefers TypeScript/Node.js
The data source has a Python SDKThe data source has a JavaScript/TypeScript SDK
You’re prototyping quicklyYou need strong type safety
You prefer FastAPI/async patternsYou prefer Express patterns
Both SDKs are functionally equivalent. Choose the language your team is most comfortable with.

​
Architecture

Custom connectors integrate with Omni through the connector-manager service: 
┌─────────────────┐         ┌───────────────────┐         ┌─────────────┐
│  Your Connector │ ←HTTP→  │ connector-manager │ ←Queue→ │ omni-indexer│
│   (SDK-based)   │         │   (orchestrator)  │         │  (indexing) │
└─────────────────┘         └───────────────────┘         └─────────────┘
        │                            │
        │                            │
        ↓                            ↓
┌─────────────────┐         ┌───────────────────┐
│   Data Source   │         │    PostgreSQL     │
│   (your API)    │         │   (storage/queue) │
└─────────────────┘         └───────────────────┘

​
Connector Lifecycle

  1. connector-manager triggers a sync via HTTP POST /sync
  2. Your connector receives the request with source configuration and credentials
  3. Your connector fetches data from the external source
  4. Documents are emitted through the SDK, which sends them to connector-manager
  5. connector-manager queues events for the indexer
  6. Your connector reports completion or failure

​
HTTP Endpoints

The SDK automatically exposes these endpoints:
EndpointMethodPurpose
/healthGETHealth check for container orchestration
/manifestGETReturns connector metadata and capabilities
/syncPOSTTriggers a sync operation
/cancelPOSTCancels a running sync
/actionPOSTExecutes connector-specific actions

​
Quick Start Example

Here’s a minimal connector implementation in Python: 
from omni_connector import Connector, Document, DocumentMetadata, SyncContext

class MyConnector(Connector):
    @property
    def name(self) -> str:
        return "my-connector"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def sync_modes(self) -> list[str]:
        return ["full"]

    async def sync(
        self,
        source_config: dict,
        credentials: dict,
        state: dict | None,
        ctx: SyncContext
    ) -> None:
        # Fetch items from your data source
        items = await self.fetch_items(credentials["api_key"])

        for item in items:
            # Store content
            content_id = await ctx.content_storage.save(item["content"])

            # Emit document
            await ctx.emit(Document(
                external_id=item["id"],
                title=item["title"],
                content_id=content_id,
                metadata=DocumentMetadata(
                    url=item["url"],
                    created_at=item["created_at"],
                ),
            ))

            await ctx.increment_scanned()

        await ctx.complete()

if __name__ == "__main__":
    MyConnector().serve(port=8000)

​
Environment Variables

All SDK-based connectors require:
VariableRequiredDescription
CONNECTOR_MANAGER_URLYesURL of the connector-manager service
PORTNoHTTP server port (default: 8000)
DATABASE_URLConditionalPostgreSQL connection string
Database connection can also be configured with individual variables:
  • DATABASE_HOST, DATABASE_PORT, DATABASE_USERNAME, DATABASE_PASSWORD, DATABASE_NAME

​
Example Connectors

Both SDKs include an RSS connector example that demonstrates:
  • Fetching data from an external API
  • Parsing and transforming content
  • Emitting documents with metadata
  • Implementing incremental sync
  • Handling errors and cancellation
  • Implementing custom actions
See the SDK-specific documentation for the complete example walkthrough.

​
Deploying Your Connector

​
Docker Deployment

Create a Dockerfile for your connector: 
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "main.py"]
Add to your docker-compose.override.yml: 
services:
  my-connector:
    build: ./connectors/my-connector
    environment:
      CONNECTOR_MANAGER_URL: http://connector-manager:8090
      PORT: 8000
    networks:
      - omni-network

​
Registering with Connector Manager

After deploying, add the connector URL to connector-manager’s environment: 
CONNECTOR_MY_CONNECTOR_URL=http://my-connector:8000

​
What’s Next

Python SDK

Detailed Python SDK documentation

TypeScript SDK

Detailed TypeScript SDK documentation