TypeScript SDK

The TypeScript SDK (@getomnico/connector) provides everything you need to build custom connectors that integrate with Omni’s search and AI platform.

Installation

The SDK is not published to npm. Connectors live in the Omni monorepo and depend on the SDK as a local file path. To start a new TypeScript connector, fork or clone omni and create a directory under connectors/:

git clone https://github.com/getomnico/omni.git
cd omni/connectors
mkdir my-connector && cd my-connector

In your package.json, reference the SDK with a file: link — copy from connectors/linear/package.json:

{
  "name": "@getomnico/my-connector",
  "version": "1.0.0",
  "type": "module",
  "dependencies": {
    "@getomnico/connector": "file:../../sdk/typescript"
  }
}

Then install:

npm install

Requirements:

Node.js >= 20.0.0

Transitive dependencies pulled in by the SDK:

express - Web framework
pino - Structured logging
zod - Schema validation

Run /build-connector <service name> from Claude Code inside the omni repo to scaffold the entire connector structure (sync logic, manifest, Dockerfile, frontend wiring, Terraform, integration tests) following the conventions used by built-in connectors. See the SDK overview for details.

Quick Start

import {
  Connector,
  Document,
  DocumentMetadata,
  DocumentPermissions,
  SyncContext,
} from '@getomnico/connector';

class MyConnector extends Connector {
  get name(): string {
    return 'my-connector';
  }

  get version(): string {
    return '1.0.0';
  }

  get sourceTypes(): string[] {
    return ['my_source_type'];
  }

  get syncModes(): string[] {
    return ['full', 'incremental'];
  }

  async sync(
    sourceConfig: Record<string, unknown>,
    credentials: Record<string, unknown>,
    state: Record<string, unknown> | null,
    ctx: SyncContext
  ): Promise<void> {
    // Your sync logic here
    await ctx.complete();
  }
}

const connector = new MyConnector();
connector.serve({ port: 8000 });

When running under Docker, set CONNECTOR_MANAGER_URL, CONNECTOR_HOST_NAME, and PORT. The SDK uses CONNECTOR_HOST_NAME plus PORT to register the connector URL that connector-manager calls for syncs, actions, MCP resources, and prompts.

Core Concepts

Connector Class

The Connector abstract class defines the interface for all connectors. Required Properties:

Property	Type	Description
`name`	`string`	Unique connector identifier (e.g., `"my-connector"`)
`version`	`string`	Semantic version (e.g., `"1.0.0"`)
`sourceTypes`	`string[]`	Source types the connector handles

Optional Properties:

Property	Type	Default	Description
`syncModes`	`string[]`	`["full"]`	Supported modes: `"full"`, `"incremental"`
`actions`	`ActionDefinition[]`	`[]`	Custom actions the connector supports

Required Methods:

async sync(
  sourceConfig: Record<string, unknown>,  // Configuration from Omni UI
  credentials: Record<string, unknown>,   // Stored credentials
  state: Record<string, unknown> | null,  // Latest per-run checkpoint
  ctx: SyncContext                        // Sync context utilities
): Promise<void>

Optional Methods:

cancel(syncRunId: string): void
// Handle sync cancellation request

async executeAction(
  action: string,
  params: Record<string, unknown>,
  credentials: Record<string, unknown>
): Promise<Record<string, unknown>>
// Execute a custom action

SyncContext

The SyncContext object is passed to your sync method and provides utilities for the sync operation. Properties:

Property	Type	Description
`syncRunId`	`string`	Unique identifier for this sync run
`sourceId`	`string`	Source identifier
`sourceType`	`string \| null`	Source type dispatched by connector-manager
`state`	`Record<string, unknown>`	Latest per-run checkpoint when resuming
`syncMode`	`SyncMode`	Dispatched mode: `full`, `incremental`, or `realtime`
`isResume`	`boolean`	Whether this run was resumed after an interruption
`contentStorage`	`ContentStorage`	Storage for document content
`documentsEmitted`	`number`	Count of emitted documents, seeded from the running sync on resume
`documentsScanned`	`number`	Count of scanned items, seeded from the running sync on resume

Methods:

// Emit a new document
await ctx.emit(document: Document): Promise<void>

// Emit an updated document
await ctx.emitUpdated(document: Document): Promise<void>

// Mark a document as deleted
await ctx.emitDeleted(externalId: string): Promise<void>

// Emit source group membership for permission filtering
await ctx.emitGroupMembership(
  groupEmail: string,
  memberEmails: string[],
  groupName?: string,
): Promise<void>

// Report a non-fatal error for a document
ctx.emitError(externalId: string, error: string): void

// Increment scanned counter (also sends heartbeat)
await ctx.incrementScanned(): Promise<void>

// Increment the server-side updated counter
await ctx.incrementUpdated(count: number): Promise<void>

// Check source-level user allow/deny settings before emitting user-owned docs
ctx.shouldIndexUser(userEmail: string): boolean

// Save a per-run checkpoint for resumability
await ctx.saveCheckpoint(checkpoint: Record<string, unknown>): Promise<void>

// Alias for saveCheckpoint
await ctx.saveState(state: Record<string, unknown>): Promise<void>

// Mark sync as completed, optionally saving the final checkpoint first
await ctx.complete(checkpoint?: Record<string, unknown>): Promise<void>

// Mark sync as failed
await ctx.fail(error: string): Promise<void>

// Check if sync was cancelled
ctx.isCancelled(): boolean

ContentStorage

The ContentStorage class handles storing document content.

// Store text content
const contentId = await ctx.contentStorage.save(
  content: string,
  contentType?: string  // default: "text/plain"
): Promise<string>

// Store binary content (base64 encoded)
const contentId = await ctx.contentStorage.saveBinary(
  content: Buffer,
  contentType: string
): Promise<string>

// Extract text from a binary file (PDF, DOCX, PPTX, XLSX, images, ...)
// via connector-manager's /sdk/extract-text endpoint, without persisting.
// Returns the extracted text.
const text = await ctx.contentStorage.extractText(
  data: Buffer,
  mimeType: string,
  filename: string,
): Promise<string>

// Same as extractText, but also stores the extracted text and returns
// the resulting contentId.
const contentId = await ctx.contentStorage.extractAndStoreContent(
  data: Buffer,
  mimeType: string,
  filename: string,
): Promise<string>

The returned contentId is a ULID that references the stored content.

Always prefer extractText / extractAndStoreContent over shelling out to your own PDF/Office parser. The connector-manager routes these calls through the centralized Docling service when the admin has enabled it (with the configured quality preset), and falls back to a lightweight built-in extractor otherwise — so your connector automatically honors the instance-wide document-conversion setting.

Document Model

The Document interface represents a searchable document.

import { Document, DocumentMetadata, DocumentPermissions } from '@getomnico/connector';

const document: Document = {
  external_id: 'unique-id-from-source',
  title: 'Document Title',
  content_id: contentId,  // From contentStorage.save()
  metadata: {
    title: 'Document Title',
    author: 'Author Name',
    created_at: new Date().toISOString(),
    updated_at: new Date().toISOString(),
    mime_type: 'text/plain',
    size: 1234,
    url: 'https://source.com/doc/123',
    path: '/folder/document.txt',
    extra: { custom_field: 'value' },
  },
  permissions: {
    public: true,
    users: ['user@example.com'],
    groups: ['engineering'],
  },
  attributes: { category: 'documentation' },
};

Document Fields:

Field	Type	Required	Description
`external_id`	`string`	Yes	Unique ID from the source system
`title`	`string`	Yes	Document title for display
`content_id`	`string`	Yes	Reference to stored content
`metadata`	`DocumentMetadata`	No	Document metadata
`permissions`	`DocumentPermissions`	No	Access control
`attributes`	`Record<string, unknown>`	No	Custom searchable attributes

Actions

Connectors can define custom actions that users can trigger from the Omni UI.

import { ActionDefinition } from '@getomnico/connector';

class MyConnector extends Connector {
  get actions(): ActionDefinition[] {
    return [
      {
        name: 'validate_connection',
        description: 'Test the API connection',
        input_schema: {
          type: 'object',
          properties: {
            timeout: {
              type: 'integer',
              description: 'Connection timeout in seconds',
            },
          },
        },
      },
    ];
  }

  async executeAction(
    action: string,
    params: Record<string, unknown>,
    credentials: Record<string, unknown>
  ): Promise<Record<string, unknown>> {
    if (action === 'validate_connection') {
      // Perform validation
      return { status: 'success', message: 'Connection valid' };
    }
    throw new Error(`Unknown action: ${action}`);
  }
}

RSS Connector Example

Here’s a complete example of an RSS feed connector:

import {
  ActionDefinition,
  Connector,
  Document,
  DocumentMetadata,
  DocumentPermissions,
  SyncContext,
} from '@getomnico/connector';

interface FeedEntry {
  id?: string;
  link?: string;
  title?: string;
  content?: string;
  summary?: string;
  author?: string;
  published?: string;
}

interface Feed {
  title?: string;
  entries: FeedEntry[];
}

class RSSConnector extends Connector {
  get name(): string {
    return 'rss';
  }

  get version(): string {
    return '1.0.0';
  }

  get sourceTypes(): string[] {
    return ['rss'];
  }

  get syncModes(): string[] {
    return ['full', 'incremental'];
  }

  get actions(): ActionDefinition[] {
    return [
      {
        name: 'validate_feed',
        description: 'Validate that a feed URL is accessible',
        input_schema: {
          type: 'object',
          properties: {
            url: {
              type: 'string',
              description: 'The RSS/Atom feed URL to validate',
            },
          },
          required: ['url'],
        },
      },
    ];
  }

  async sync(
    sourceConfig: Record<string, unknown>,
    credentials: Record<string, unknown>,
    state: Record<string, unknown> | null,
    ctx: SyncContext
  ): Promise<void> {
    const feedUrl = sourceConfig.feed_url as string;
    if (!feedUrl) {
      await ctx.fail('Missing feed_url in source configuration');
      return;
    }

    // Fetch and parse the feed
    const feed = await this.parseFeed(feedUrl);

    // Get last sync time for incremental sync
    const lastSyncTime = state?.last_sync_time
      ? new Date(state.last_sync_time as string)
      : null;

    for (const entry of feed.entries) {
      if (ctx.isCancelled()) {
        return;
      }

      await ctx.incrementScanned();

      // Skip old entries in incremental mode
      if (lastSyncTime && entry.published) {
        const entryDate = new Date(entry.published);
        if (entryDate <= lastSyncTime) {
          continue;
        }
      }

      // Build content from entry
      const content = entry.content || entry.summary || entry.title || '';
      const contentId = await ctx.contentStorage.save(content);

      // Create document
      const doc: Document = {
        external_id: entry.id || entry.link || '',
        title: entry.title || 'Untitled',
        content_id: contentId,
        metadata: {
          author: entry.author,
          url: entry.link,
          created_at: entry.published,
          extra: {
            feed_title: feed.title,
            feed_url: feedUrl,
          },
        },
        permissions: {
          public: true,
        },
      };

      await ctx.emit(doc);
    }

    // Complete with updated state
    await ctx.complete({ last_sync_time: new Date().toISOString() });
  }

  async executeAction(
    action: string,
    params: Record<string, unknown>,
    credentials: Record<string, unknown>
  ): Promise<Record<string, unknown>> {
    if (action === 'validate_feed') {
      const url = params.url as string;
      if (!url) {
        return { status: 'error', error: 'Missing URL parameter' };
      }

      try {
        const feed = await this.parseFeed(url);
        return {
          status: 'success',
          result: {
            title: feed.title,
            entry_count: feed.entries.length,
          },
        };
      } catch (error) {
        return { status: 'error', error: String(error) };
      }
    }

    return { status: 'error', error: `Unknown action: ${action}` };
  }

  private async parseFeed(url: string): Promise<Feed> {
    // Simple XML parsing (use rss-parser library in production)
    const response = await fetch(url);
    const text = await response.text();

    const entries: FeedEntry[] = [];
    const itemRegex = /<item>([\s\S]*?)<\/item>/g;
    let match;

    while ((match = itemRegex.exec(text)) !== null) {
      const item = match[1];
      entries.push({
        title: this.extractTag(item, 'title'),
        link: this.extractTag(item, 'link'),
        content: this.extractTag(item, 'content:encoded') ||
                 this.extractTag(item, 'description'),
        author: this.extractTag(item, 'author') ||
                this.extractTag(item, 'dc:creator'),
        published: this.extractTag(item, 'pubDate'),
        id: this.extractTag(item, 'guid') ||
            this.extractTag(item, 'link'),
      });
    }

    return {
      title: this.extractTag(text, 'title'),
      entries,
    };
  }

  private extractTag(xml: string, tag: string): string | undefined {
    const regex = new RegExp(`<${tag}[^>]*>([\\s\\S]*?)</${tag}>`, 'i');
    const match = regex.exec(xml);
    return match ? match[1].trim() : undefined;
  }
}

const connector = new RSSConnector();
connector.serve({ port: 8000 });

Error Handling

The SDK provides custom error classes:

import {
  ConnectorError,      // Base error
  SdkClientError,      // Communication errors (includes statusCode)
  SyncCancelledError,  // Sync was cancelled
  ConfigurationError,  // Configuration issues
} from '@getomnico/connector';

Handling errors in sync:

async sync(
  sourceConfig: Record<string, unknown>,
  credentials: Record<string, unknown>,
  state: Record<string, unknown> | null,
  ctx: SyncContext
): Promise<void> {
  try {
    // Your sync logic
    await ctx.complete();
  } catch (error) {
    if (error instanceof SyncCancelledError) {
      // Handle cancellation gracefully
      return;
    }
    await ctx.fail(String(error));
  }
}

Development

Project Setup

# Install dependencies
npm install

# Build
npm run build

# Run tests
npm test

# Watch mode for tests
npm run test:watch

# Lint
npm run lint

TypeScript Configuration

The SDK uses strict TypeScript settings. Your tsconfig.json should include:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "ESNext",
    "moduleResolution": "bundler",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true
  }
}

API Reference

Exports

The SDK exports the following: Core Classes:

Connector - Abstract base class
SyncContext - Sync operation context
ContentStorage - Content storage interface
SdkClient - SDK client for connector-manager

Data Models (with Zod schemas):

Document, DocumentMetadata, DocumentPermissions
ConnectorEvent, EventType
ActionDefinition
ActionRequest, ActionResponse
ConnectorManifest, SyncMode
SyncRequest, SyncResponse
CancelRequest, CancelResponse

Error Classes:

ConnectorError, SdkClientError, SyncCancelledError, ConfigurationError

Zod Schemas

All data models include Zod schemas for runtime validation:

import { DocumentSchema, SyncRequestSchema } from '@getomnico/connector';

// Validate incoming data
const result = SyncRequestSchema.safeParse(data);
if (!result.success) {
  console.error('Validation failed:', result.error);
}

Getting Started

Deployment

Connectors

Developers

User Guide

Administration

Installation

Quick Start

Core Concepts

Connector Class

SyncContext

ContentStorage

Document Model

Actions

RSS Connector Example

Error Handling

Development

Project Setup

TypeScript Configuration

API Reference

Exports

Zod Schemas

What’s Next

SDK Overview

Python SDK

​Installation

​Quick Start

​Core Concepts

​Connector Class

​SyncContext

​ContentStorage

​Document Model

​Actions

​RSS Connector Example

​Error Handling

​Development

​Project Setup

​TypeScript Configuration

​API Reference

​Exports

​Zod Schemas

​What’s Next

SDK Overview

Python SDK

Installation

Quick Start

Core Concepts

Connector Class

SyncContext

ContentStorage

Document Model

Actions

RSS Connector Example

Error Handling

Development

Project Setup

TypeScript Configuration

API Reference

Exports

Zod Schemas

What’s Next